package lkgateway // lkagent.go — подключение ядра к кабинету, когда у ядра нет адреса в интернете. // // Обычный порядок: кабинет на хостинге сам обращается к bj-server. Но ядро депозитария // часто стоит во внутренней сети без публичного имени, и «достучаться» до него снаружи // нельзя. Тогда связь разворачивается: ядро само подключается к кабинету по его // доменному имени, забирает накопившиеся запросы клиентов, выполняет их у себя и // возвращает ответы. Наружу из сети депозитария идёт только исходящее соединение по // TLS — открывать порты и пробрасывать адрес не нужно. // // Запрос выполняется тем же обработчиком /api/lk/*, что и в прямом режиме: ядро просто // прогоняет его через свой mux, как если бы кабинет пришёл по сети. import ( "bytes" "context" "encoding/json" "fmt" "io" "log" "net/http" "strings" "sync" "sync/atomic" "time" "git.zetit.ru/zuevav/Bridge-and-Join-s/modules/lk" ) // lkAgentState — состояние канала для страницы настроек. type lkAgentStateT struct { Connected atomic.Bool LastOK atomic.Int64 // unix-время последнего успешного обращения LastErr atomic.Value // string Served atomic.Uint64 // сколько запросов кабинета выполнено } var lkAgentState lkAgentStateT func (a *lkAgentStateT) fail(err error) { a.Connected.Store(false) a.LastErr.Store(err.Error()) } func (a *lkAgentStateT) ok() { a.Connected.Store(true) a.LastOK.Store(time.Now().Unix()) a.LastErr.Store("") } // LKAgentStatus — что показать администратору в настройках. type LKAgentStatus struct { Enabled bool `json:"enabled"` Cabinet string `json:"cabinet"` Connected bool `json:"connected"` LastOK string `json:"last_ok,omitempty"` LastErr string `json:"last_error,omitempty"` Served uint64 `json:"served"` } func lkAgentStatus(s Settings) LKAgentStatus { st := LKAgentStatus{ Enabled: s.LK.AgentMode, Cabinet: strings.TrimSpace(s.LK.CabinetURL), Connected: lkAgentState.Connected.Load(), Served: lkAgentState.Served.Load(), } if v := lkAgentState.LastOK.Load(); v > 0 { st.LastOK = time.Unix(v, 0).Format("02.01.2006 15:04:05") } if v, _ := lkAgentState.LastErr.Load().(string); v != "" { st.LastErr = v } return st } // lkAgentLoop — вечный цикл: подключиться к кабинету, забрать запрос, выполнить, ответить. // Запускается всегда, но работает только когда режим ожидания включён в настройках: // так администратор может включить и выключить его без перезапуска службы. func (s *Server) lkAgentLoop(ctx context.Context) { // Несколько подключений сразу: запрос к учётной системе может идти десятки секунд, // и один медленный клиент не должен задерживать остальных. var wg sync.WaitGroup for i := 0; i < lkAgentWorkers; i++ { wg.Add(1) go func() { defer wg.Done() s.lkAgentWorker(ctx) }() } wg.Wait() } // lkAgentWorkers — сколько запросов кабинета ядро обслуживает одновременно. const lkAgentWorkers = 4 func (s *Server) lkAgentWorker(ctx context.Context) { defer func() { if rec := recover(); rec != nil { log.Printf("кабинет: сбой при обслуживании запроса: %v", rec) } }() client := &http.Client{Timeout: 40 * time.Second} for { if ctx.Err() != nil { return } cfg := s.rc.Snapshot().LK cabinet := strings.TrimRight(strings.TrimSpace(cfg.CabinetURL), "/") if !cfg.AgentMode || cabinet == "" || strings.TrimSpace(cfg.APIKey) == "" { lkAgentState.Connected.Store(false) if !sleepCtx(ctx, 15*time.Second) { return } continue } task, err := lkAgentFetch(ctx, client, cabinet, cfg.APIKey) if err != nil { lkAgentState.fail(err) log.Printf("кабинет: нет связи с %s: %v", cabinet, err) if !sleepCtx(ctx, 20*time.Second) { // не долбим хостинг при аварии return } continue } lkAgentState.ok() if task == nil { continue // запросов не было, кабинет отпустил соединение — подключаемся заново } res := s.lkAgentServe(ctx, task) if err := lkAgentReply(ctx, client, cabinet, cfg.APIKey, res); err != nil { log.Printf("кабинет: ответ на запрос %s не доставлен: %v", task.ID, err) continue } lkAgentState.Served.Add(1) } } // lkAgentTask — запрос, ожидающий в кабинете: вызов API или страница в конверте. type lkAgentTask = lk.PageRequest // lkAgentFetch — забрать очередной запрос. nil без ошибки — запросов не было. func lkAgentFetch(ctx context.Context, c *http.Client, cabinet, key string) (*lkAgentTask, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, cabinet+"/agent/tasks", nil) if err != nil { return nil, err } req.Header.Set("X-BJ-Key", key) resp, err := c.Do(req) if err != nil { return nil, err } defer resp.Body.Close() switch resp.StatusCode { case http.StatusNoContent: return nil, nil case http.StatusOK: var t lkAgentTask if err := json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(&t); err != nil { return nil, err } if t.ID == "" || (t.Kind != "page" && !strings.HasPrefix(t.Path, "/api/lk/")) { return nil, fmt.Errorf("кабинет прислал неизвестный запрос") } return &t, nil case http.StatusForbidden: return nil, fmt.Errorf("кабинет не принял ключ модуля: сверьте BJ_KEY в /etc/bj-lk/lk.env") default: return nil, fmt.Errorf("кабинет ответил %d", resp.StatusCode) } } // lkAgentServe — выполнить запрос кабинета у себя: страницу — встроенным кабинетом, // вызов API — тем же обработчиком /api/lk/*, что и по сети. func (s *Server) lkAgentServe(ctx context.Context, t *lkAgentTask) lk.PageResponse { if t.Kind == "page" { if globalLKCabinet == nil { return lk.PageResponse{ID: t.ID, Status: http.StatusServiceUnavailable, Body: []byte("кабинет в ядре не запущен")} } req, _ := http.NewRequestWithContext(ctx, http.MethodGet, "/", nil) return globalLKCabinet.serve(req, *t) } cctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() req, err := http.NewRequestWithContext(cctx, t.Method, t.Path, bytes.NewReader(t.Body)) if err != nil { return lk.PageResponse{ID: t.ID, Status: http.StatusBadRequest, Body: []byte(`{"error":"некорректный запрос кабинета"}`)} } req.Header.Set("Content-Type", "application/json") req.Header.Set("X-BJ-Key", s.rc.Snapshot().LK.APIKey) // запрос пришёл по ключу, он же идёт дальше if t.Session != "" { req.Header.Set("X-BJ-Session", t.Session) } if t.ClientIP != "" { req.Header.Set("X-BJ-Client-IP", t.ClientIP) } if t.ClientAgent != "" { req.Header.Set("X-BJ-Client-Agent", t.ClientAgent) } req.RemoteAddr = "127.0.0.1:0" // обращение пришло по нашему исходящему каналу rec := &lkAgentRecorder{status: http.StatusOK} s.mux.ServeHTTP(rec, req) return lk.PageResponse{ID: t.ID, Status: rec.status, Body: rec.buf.Bytes()} } // lkAgentRecorder — приёмник ответа обработчика: пишем в память, а не в сеть. type lkAgentRecorder struct { status int buf bytes.Buffer hdr http.Header } func (w *lkAgentRecorder) Header() http.Header { if w.hdr == nil { w.hdr = http.Header{} } return w.hdr } func (w *lkAgentRecorder) Write(b []byte) (int, error) { return w.buf.Write(b) } func (w *lkAgentRecorder) WriteHeader(code int) { w.status = code } // lkAgentReply — вернуть кабинету ответ на его запрос. func lkAgentReply(ctx context.Context, c *http.Client, cabinet, key string, res lk.PageResponse) error { payload, _ := json.Marshal(res) req, err := http.NewRequestWithContext(ctx, http.MethodPost, cabinet+"/agent/result", bytes.NewReader(payload)) if err != nil { return err } req.Header.Set("X-BJ-Key", key) req.Header.Set("Content-Type", "application/json") resp, err := c.Do(req) if err != nil { return err } defer resp.Body.Close() _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) if resp.StatusCode >= 400 && resp.StatusCode != http.StatusGone { return fmt.Errorf("кабинет ответил %d", resp.StatusCode) } return nil } // sleepCtx — пауза, прерываемая остановкой службы. func sleepCtx(ctx context.Context, d time.Duration) bool { t := time.NewTimer(d) defer t.Stop() select { case <-ctx.Done(): return false case <-t.C: return true } }