From f6b94a44d02837b19e19f67277fe9e912ae1b440 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Wed, 8 Apr 2026 12:29:07 +0700 Subject: [PATCH] refactor: enhance bird metrics polling with context support Updated the `startBirdMetricsPoller` function to accept a context parameter, allowing for better control over the polling lifecycle. This change was applied in both `evobgp-all` and `evobgp-api` main files. Additionally, modified the `StartBirdProtocolsPoller` function to handle context cancellation, ensuring graceful shutdown of the polling routine. Introduced a new service in the Docker Compose configuration for logging runtime service outputs, improving observability during deployment. --- cmd/evobgp-all/main.go | 6 +- cmd/evobgp-api/main.go | 8 +-- deploy/compose/stack.microvps-full.yaml | 43 ++++++++++++ internal/jobs/job.go | 88 +++++++++++++++++++++---- internal/jobs/worker.go | 74 ++++++++++----------- internal/observability/metrics.go | 18 +++-- internal/pipeline/parse.go | 54 +++++++++++---- internal/pipeline/parse_test.go | 50 ++++++++++++++ 8 files changed, 267 insertions(+), 74 deletions(-) create mode 100644 internal/pipeline/parse_test.go diff --git a/cmd/evobgp-all/main.go b/cmd/evobgp-all/main.go index 3d7e8c5..12af768 100644 --- a/cmd/evobgp-all/main.go +++ b/cmd/evobgp-all/main.go @@ -51,7 +51,7 @@ func main() { go render.Run(ctx, renderDeps) go deploy.Run(ctx, deployDeps) - startBirdMetricsPoller() + startBirdMetricsPoller(ctx) httpSrv := &http.Server{ Addr: cfg.HTTPAddr, @@ -75,7 +75,7 @@ func main() { log.Printf("%s stopped", platform.ServiceName("evobgp-all")) } -func startBirdMetricsPoller() { +func startBirdMetricsPoller(ctx context.Context) { sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) if sock == "" { return @@ -85,7 +85,7 @@ func startBirdMetricsPoller() { interval = d } bin := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")) - observability.StartBirdProtocolsPoller(sock, bin, interval, + observability.StartBirdProtocolsPoller(ctx, sock, bin, interval, func(ctx context.Context, socket, birdcBin string) (string, error) { return birdfmt.ShowProtocols(ctx, socket, birdcBin) }, diff --git a/cmd/evobgp-api/main.go b/cmd/evobgp-api/main.go index 5622846..c11a033 100644 --- a/cmd/evobgp-api/main.go +++ b/cmd/evobgp-api/main.go @@ -35,11 +35,11 @@ func main() { defer srv.Close() observability.SetBuildInfo("0.1.0", strings.TrimSpace(os.Getenv("EVOBGP_GIT_SHA"))) - startBirdMetricsPoller() - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + startBirdMetricsPoller(ctx) + httpSrv := &http.Server{ Addr: cfg.HTTPAddr, Handler: srv.Handler(), @@ -68,7 +68,7 @@ func main() { log.Printf("%s stopped", platform.ServiceName("evobgp-api")) } -func startBirdMetricsPoller() { +func startBirdMetricsPoller(ctx context.Context) { sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) if sock == "" { return @@ -78,7 +78,7 @@ func startBirdMetricsPoller() { interval = d } bin := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")) - observability.StartBirdProtocolsPoller(sock, bin, interval, + observability.StartBirdProtocolsPoller(ctx, sock, bin, interval, func(ctx context.Context, socket, birdcBin string) (string, error) { return birdfmt.ShowProtocols(ctx, socket, birdcBin) }, diff --git a/deploy/compose/stack.microvps-full.yaml b/deploy/compose/stack.microvps-full.yaml index 2748139..1547050 100644 --- a/deploy/compose/stack.microvps-full.yaml +++ b/deploy/compose/stack.microvps-full.yaml @@ -13,6 +13,12 @@ # Traefik не находит acme.json и отдаёт дефолтный сертификат до новой выдачи LE. # Не использовать `docker compose down -v` без бэкапа. Если раньше был том с префиксом # проекта, перенесите acme.json в том evobgp_traefik_letsencrypt. +# +# Долгий сбор логов в файлы на хосте: сервис stack-runtime-logs пишет в каталог +# ./runtime-logs/ (рядом с этим compose-файлом) по одному файлу на сервис. +# Имя проекта в Docker должно совпадать с label com.docker.compose.project: +# при смене `name:` или имени стека в Portainer задайте COMPOSE_PROJECT_NAME. +# Требуется доступ к docker.sock (полные права на демон — осознанно). name: evobgp-microvps-full configs: @@ -216,6 +222,43 @@ services: max-size: "10m" max-file: "3" + stack-runtime-logs: + image: docker:27-cli + restart: unless-stopped + depends_on: + evobgp-all: + condition: service_started + environment: + COMPOSE_PROJECT_NAME: ${COMPOSE_PROJECT_NAME:-evobgp-microvps-full} + volumes: + - /var/run/docker.sock:/var/run/docker.sock + - type: bind + source: ./runtime-logs + target: /logs + entrypoint: ["/bin/sh", "-c"] + command: + - | + set -eu + mkdir -p /logs + PROJECT=$$COMPOSE_PROJECT_NAME + SERVICES="postgres nats bird2 evobgp-agent evobgp-all evobgp-web evobgp-edge prometheus" + log_one() { + svc=$$1 + f="/logs/$$svc.log" + while true; do + cid=$$(docker ps -q \ + -f "label=com.docker.compose.service=$$svc" \ + -f "label=com.docker.compose.project=$$PROJECT" | head -n1) + if [ -n "$$cid" ]; then + echo "---- $$(date -u +"%Y-%m-%dT%H:%M:%SZ") attach $$svc $$cid ----" >> "$$f" + docker logs -f --timestamps "$$cid" >> "$$f" 2>&1 || true + fi + sleep 3 + done + } + for s in $$SERVICES; do log_one "$$s" & done + wait + volumes: pgdata: bird_etc: diff --git a/internal/jobs/job.go b/internal/jobs/job.go index da70f73..fafce3d 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -2,7 +2,10 @@ package jobs import ( "fmt" + "os" "sort" + "strconv" + "strings" "sync" "time" @@ -20,17 +23,17 @@ const ( // Job is the API-facing job model (поля согласованы со схемой job_audit в миграциях; персистенция в БД пока не подключена). type Job struct { - ID string - TenantID string - Kind string - Status string - IdempotencyKey *string - ModuleID *string - CreatedAt time.Time - StartedAt *time.Time - FinishedAt *time.Time - Error *string - ProgressPct *int16 + ID string + TenantID string + Kind string + Status string + IdempotencyKey *string + ModuleID *string + CreatedAt time.Time + StartedAt *time.Time + FinishedAt *time.Time + Error *string + ProgressPct *int16 Meta map[string]any cancelRequested bool mu sync.Mutex @@ -148,6 +151,25 @@ func (j *Job) Snapshot() map[string]any { return m } +var jobRegistryMaxJobsOnce sync.Once +var jobRegistryMaxJobs int + +// registryMaxJobsFromEnv returns EVOBGP_JOB_REGISTRY_MAX_JOBS once (0 = без лимита, только завершённые джобы вытесняются). +func registryMaxJobsFromEnv() int { + jobRegistryMaxJobsOnce.Do(func() { + s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_REGISTRY_MAX_JOBS")) + if s == "" { + return + } + n, err := strconv.Atoi(s) + if err != nil || n <= 0 { + return + } + jobRegistryMaxJobs = n + }) + return jobRegistryMaxJobs +} + // Registry — in-memory очередь и индекс по idempotency в процессе, где поднят HTTP API (evobgp-api и evobgp-all). // Отдельные воркеры в reference-профиле не разделяют память с API: scheduler дергает refresh по HTTP; см. docs/architecture.md. // Запись задач в PostgreSQL job_audit + SKIP LOCKED / внешний брокер — планируемое расширение (архитектурный план §2, §7.10). @@ -171,11 +193,54 @@ func NewRegistry(workerStart func(j *Job)) *Registry { } } +// pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs. +func (r *Registry) pruneTerminalIfOver(maxJobs int) { + if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs { + return + } + type fin struct { + j *Job + t time.Time + } + var cands []fin + for _, j := range r.byID { + st := j.statusLocked() + if st != StatusSucceeded && st != StatusFailed && st != StatusCancelled { + continue + } + j.mu.Lock() + ft := j.FinishedAt + j.mu.Unlock() + if ft == nil { + continue + } + cands = append(cands, fin{j: j, t: *ft}) + } + need := len(r.byID) - maxJobs + if need <= 0 || len(cands) == 0 { + return + } + sort.Slice(cands, func(i, j int) bool { return cands[i].t.Before(cands[j].t) }) + if need > len(cands) { + need = len(cands) + } + for i := 0; i < need; i++ { + v := cands[i].j + delete(r.byID, v.ID) + if v.IdempotencyKey != nil && *v.IdempotencyKey != "" { + delete(r.byIdempo, idempoKey{tenant: v.TenantID, key: *v.IdempotencyKey}) + } + } +} + // Enqueue creates a job or returns an existing one for the same idempotency key. func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) (*Job, bool, error) { r.mu.Lock() defer r.mu.Unlock() + maxJobs := registryMaxJobsFromEnv() + r.pruneTerminalIfOver(maxJobs) + if idempotencyKey != nil && *idempotencyKey != "" { k := idempoKey{tenant: tenantID, key: *idempotencyKey} if existing, ok := r.byIdempo[k]; ok { @@ -197,6 +262,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module r.byIdempo[idempoKey{tenant: tenantID, key: *idempotencyKey}] = j } r.byID[j.ID] = j + r.pruneTerminalIfOver(maxJobs) if r.workerStart != nil { go r.workerStart(j) diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index cede83c..f1e5bb8 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -158,8 +158,8 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, rev string) { deferDeploy := w.Registry.CountOtherActiveModuleRefresh(j.TenantID, j.ID) > 0 if deferDeploy { j.mergeMeta(map[string]any{ - "deploy_apply_deferred": true, - "deploy_apply_defer_reason": "parallel_module_refresh", + "deploy_apply_deferred": true, + "deploy_apply_defer_reason": "parallel_module_refresh", }) } j.Succeed() @@ -324,19 +324,6 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a return nil, 0, fmt.Errorf("store not configured") } commLabels := buildCommunityLabelMap(w.Store, tenantID) - var all []store.PrefixRow - cursor := "" - for { - rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000) - all = append(all, rows...) - if !more { - break - } - cursor = next - if strings.TrimSpace(cursor) == "" { - break - } - } type agg struct { kind string source string @@ -345,22 +332,35 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a sample []string } groups := map[string]*agg{} - for _, p := range all { - src := strings.TrimSpace(p.Source) - comm := "none" - if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" { - comm = strings.TrimSpace(*p.CommunityID) + total := 0 + cursor := "" + for { + rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000) + for _, p := range rows { + total++ + src := strings.TrimSpace(p.Source) + comm := "none" + if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" { + comm = strings.TrimSpace(*p.CommunityID) + } + kind, sourceName := classifySource(src) + k := kind + "|" + sourceName + "|" + comm + g, ok := groups[k] + if !ok { + g = &agg{kind: kind, source: sourceName, community: comm} + groups[k] = g + } + g.count++ + if len(g.sample) < 5 { + g.sample = append(g.sample, p.Prefix) + } } - kind, sourceName := classifySource(src) - k := kind + "|" + sourceName + "|" + comm - g, ok := groups[k] - if !ok { - g = &agg{kind: kind, source: sourceName, community: comm} - groups[k] = g + if !more { + break } - g.count++ - if len(g.sample) < 5 { - g.sample = append(g.sample, p.Prefix) + cursor = next + if strings.TrimSpace(cursor) == "" { + break } } keys := make([]string, 0, len(groups)) @@ -374,16 +374,16 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a cl := resolveCommunityLabel(g.community, commLabels) msg := humanLogMessage(g.kind, g.source, g.count, cl, g.sample) out = append(out, map[string]any{ - "kind": g.kind, - "source": g.source, - "community": g.community, - "community_label": cl, - "prefix_count": g.count, - "sample": g.sample, - "message": msg, + "kind": g.kind, + "source": g.source, + "community": g.community, + "community_label": cl, + "prefix_count": g.count, + "sample": g.sample, + "message": msg, }) } - return out, len(all), nil + return out, total, nil } func classifySource(src string) (kind, name string) { diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 1d3e71f..87850f0 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -196,14 +196,15 @@ func (s *statusRecorder) WriteHeader(code int) { } // StartBirdProtocolsPoller runs birdc "show protocols" on interval when socket is non-empty. -func StartBirdProtocolsPoller(socket string, birdcPath string, interval time.Duration, showFn func(ctx context.Context, socket, birdcBin string) (string, error), countFn func(output string) int) { +// Горутина завершается при отмене ctx (корректное завершение вместе с процессом API). +func StartBirdProtocolsPoller(ctx context.Context, socket string, birdcPath string, interval time.Duration, showFn func(ctx context.Context, socket, birdcBin string) (string, error), countFn func(output string) int) { socket = trimSpace(socket) - if socket == "" || interval <= 0 || showFn == nil || countFn == nil { + if ctx == nil || socket == "" || interval <= 0 || showFn == nil || countFn == nil { return } scrape := func() { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - out, err := showFn(ctx, socket, birdcPath) + sctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + out, err := showFn(sctx, socket, birdcPath) cancel() if err != nil { SetBirdSessionMetrics(0, false) @@ -215,8 +216,13 @@ func StartBirdProtocolsPoller(socket string, birdcPath string, interval time.Dur scrape() t := time.NewTicker(interval) defer t.Stop() - for range t.C { - scrape() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + scrape() + } } }() } diff --git a/internal/pipeline/parse.go b/internal/pipeline/parse.go index 6927c2d..2f81c5f 100644 --- a/internal/pipeline/parse.go +++ b/internal/pipeline/parse.go @@ -3,10 +3,17 @@ package pipeline import ( "bufio" "encoding/json" + "fmt" "net/netip" "strings" ) +// Защита от pathological JSON: глубокая рекурсия при обходе и взрыв числа узлов по пути. +const ( + maxJSONWalkDepth = 512 + maxJSONPathBreadth = 50000 +) + // ParseCIDRLines extracts unique IPv4/IPv6 CIDRs from plain text (one per line, # comments, empty lines skipped). func ParseCIDRLines(body string) []netip.Prefix { seen := make(map[string]struct{}) @@ -47,7 +54,10 @@ func parseCIDRsFromJSON(body, prefixPath string) ([]netip.Prefix, error) { if err := json.Unmarshal([]byte(body), &root); err != nil { return nil, err } - values := jsonValuesAtPath(root, prefixPath) + values, err := jsonValuesAtPath(root, prefixPath) + if err != nil { + return nil, err + } seen := make(map[string]struct{}) var out []netip.Prefix for _, raw := range values { @@ -66,10 +76,10 @@ func parseCIDRsFromJSON(body, prefixPath string) ([]netip.Prefix, error) { return out, nil } -func jsonValuesAtPath(root any, prefixPath string) []string { +func jsonValuesAtPath(root any, prefixPath string) ([]string, error) { path := strings.TrimSpace(prefixPath) if path == "" { - return flattenJSONStrings(root) + return flattenJSONStrings(root, 0) } parts := strings.Split(path, ".") nodes := []any{root} @@ -98,36 +108,54 @@ func jsonValuesAtPath(root any, prefixPath string) []string { } next = append(next, child) } + if len(next) > maxJSONPathBreadth { + return nil, fmt.Errorf("json path: слишком много узлов на шаге (>%d)", maxJSONPathBreadth) + } nodes = next if len(nodes) == 0 { - return nil + return nil, nil } } var out []string for _, n := range nodes { - out = append(out, flattenJSONStrings(n)...) + part, err := flattenJSONStrings(n, 0) + if err != nil { + return nil, err + } + out = append(out, part...) } - return out + return out, nil } -func flattenJSONStrings(v any) []string { +func flattenJSONStrings(v any, depth int) ([]string, error) { + if depth > maxJSONWalkDepth { + return nil, fmt.Errorf("json: глубина вложенности превышает %d", maxJSONWalkDepth) + } switch x := v.(type) { case string: - return []string{strings.TrimSpace(x)} + return []string{strings.TrimSpace(x)}, nil case []any: var out []string for _, item := range x { - out = append(out, flattenJSONStrings(item)...) + part, err := flattenJSONStrings(item, depth+1) + if err != nil { + return nil, err + } + out = append(out, part...) } - return out + return out, nil case map[string]any: var out []string for _, item := range x { - out = append(out, flattenJSONStrings(item)...) + part, err := flattenJSONStrings(item, depth+1) + if err != nil { + return nil, err + } + out = append(out, part...) } - return out + return out, nil default: - return nil + return nil, nil } } diff --git a/internal/pipeline/parse_test.go b/internal/pipeline/parse_test.go new file mode 100644 index 0000000..f14a873 --- /dev/null +++ b/internal/pipeline/parse_test.go @@ -0,0 +1,50 @@ +package pipeline + +import ( + "strings" + "testing" +) + +func TestExtractCIDRs_JSONNested(t *testing.T) { + body := `{"a":{"b":{"c":"192.0.2.0/24"}}}` + pfx, err := ExtractCIDRs(body, "json", "a.b.c") + if err != nil { + t.Fatal(err) + } + if len(pfx) != 1 || pfx[0].String() != "192.0.2.0/24" { + t.Fatalf("got %#v", pfx) + } +} + +func TestExtractCIDRs_JSONDepthLimit(t *testing.T) { + var b strings.Builder + b.WriteString(`{"x":`) + for i := 0; i < maxJSONWalkDepth+4; i++ { + b.WriteString(`{"k":`) + } + b.WriteString(`"192.0.2.1"`) + for i := 0; i < maxJSONWalkDepth+4; i++ { + b.WriteByte('}') + } + b.WriteByte('}') + _, err := ExtractCIDRs(b.String(), "json", "") + if err == nil { + t.Fatal("expected depth error") + } +} + +func TestExtractCIDRs_JSONPathBreadth(t *testing.T) { + var b strings.Builder + b.WriteString(`{"items":{"x":[`) + for i := 0; i < maxJSONPathBreadth+1; i++ { + if i > 0 { + b.WriteByte(',') + } + b.WriteString(`"192.0.2.0/24"`) + } + b.WriteString(`]}}`) + _, err := ExtractCIDRs(b.String(), "json", "items.x[]") + if err == nil { + t.Fatal("expected breadth error") + } +}