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

@@ -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)
 		},

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)
 		},

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:

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)

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) {

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()
+			}
 		}
 	}()
 }

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
 	}
 }
 

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")
+	}
+}