From dd7d43c2c218171fb35e5b4ef0915fabd3c6b980 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Thu, 21 May 2026 10:45:46 +0700 Subject: [PATCH] perf: incremental render scale, SQL pagination and observability - BIRD filter chunking (500 CIDR); bounded job worker pool - ListModulesPage SQL push-down; revision diff limit; batch revision prune - DB pool tuning; Prometheus pipeline/job metrics - Coalesce module_refresh via idempotency; GET /jobs/{id}/report - JobAuditWriter foundation for job_audit persistence Co-authored-by: Cursor --- internal/birdfmt/filter.go | 23 +++-- internal/db/open.go | 14 +++ internal/httpapi/routes.go | 61 ++++++++++++- internal/jobs/job.go | 30 ++++++- internal/observability/metrics.go | 47 ++++++++++ internal/repository/job_audit.go | 50 +++++++++++ internal/repository/postgres.go | 138 +++++++++++++++++++++++++----- internal/store/memory.go | 5 ++ 8 files changed, 336 insertions(+), 32 deletions(-) create mode 100644 internal/repository/job_audit.go diff --git a/internal/birdfmt/filter.go b/internal/birdfmt/filter.go index 6b0553b..388541a 100644 --- a/internal/birdfmt/filter.go +++ b/internal/birdfmt/filter.go @@ -9,6 +9,21 @@ import ( const maxBGPASN = 4294967295 +const filterPrefixChunkSize = 500 + +func writePrefixSetAcceptBlocks(b *strings.Builder, keys []string) { + for i := 0; i < len(keys); i += filterPrefixChunkSize { + end := i + filterPrefixChunkSize + if end > len(keys) { + end = len(keys) + } + chunk := keys[i:end] + b.WriteString(" if net ~ [ ") + b.WriteString(strings.Join(chunk, ", ")) + b.WriteString(" ] then accept;\n") + } +} + func filterUniqueASNs(pathASNs []int64) []int64 { seen := make(map[int64]struct{}) for _, a := range pathASNs { @@ -52,9 +67,7 @@ func RenderExportFilterIPv4(filterName string, prefixes []netip.Prefix, pathASNs b.WriteString(strings.TrimSpace(filterName)) b.WriteString(" {\n") if len(keys) > 0 { - b.WriteString(" if net ~ [ ") - b.WriteString(strings.Join(keys, ", ")) - b.WriteString(" ] then accept;\n") + writePrefixSetAcceptBlocks(&b, keys) } for _, asn := range asns { fmt.Fprintf(&b, " if bgp_path ~ [= * %d =] then accept;\n", asn) @@ -95,9 +108,7 @@ func RenderExportFilterIPv6(filterName string, prefixes []netip.Prefix, pathASNs b.WriteString(strings.TrimSpace(filterName)) b.WriteString(" {\n") if len(keys) > 0 { - b.WriteString(" if net ~ [ ") - b.WriteString(strings.Join(keys, ", ")) - b.WriteString(" ] then accept;\n") + writePrefixSetAcceptBlocks(&b, keys) } for _, asn := range asns { fmt.Fprintf(&b, " if bgp_path ~ [= * %d =] then accept;\n", asn) diff --git a/internal/db/open.go b/internal/db/open.go index 326386e..182dd9c 100644 --- a/internal/db/open.go +++ b/internal/db/open.go @@ -6,9 +6,12 @@ import ( "errors" "fmt" "io/fs" + "os" "path" "sort" + "strconv" "strings" + "time" "evobgp/migrations" @@ -22,6 +25,17 @@ func OpenPostgresPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) { if err != nil { return nil, err } + if max := os.Getenv("EVOBGP_DB_MAX_CONNS"); max != "" { + if n, err := strconv.Atoi(strings.TrimSpace(max)); err == nil && n > 0 { + cfg.MaxConns = int32(n) + } + } + if min := os.Getenv("EVOBGP_DB_MIN_CONNS"); min != "" { + if n, err := strconv.Atoi(strings.TrimSpace(min)); err == nil && n >= 0 { + cfg.MinConns = int32(n) + } + } + cfg.MaxConnLifetime = 30 * time.Minute pool, err := pgxpool.NewWithConfig(ctx, cfg) if err != nil { return nil, err diff --git a/internal/httpapi/routes.go b/internal/httpapi/routes.go index 4dcaa9d..761acd8 100644 --- a/internal/httpapi/routes.go +++ b/internal/httpapi/routes.go @@ -69,6 +69,7 @@ func (s *Server) registerV1(m *http.ServeMux) { m.HandleFunc("GET /bird/status", s.handleBirdStatus) m.HandleFunc("GET /jobs", s.handleListJobs) m.HandleFunc("GET /jobs/{job_id}", s.handleGetJob) + m.HandleFunc("GET /jobs/{job_id}/report", s.handleGetJobReport) m.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob) m.HandleFunc("GET /speakers/{speaker_id}/revisions/latest", s.handleNodeLatestRevision) m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle) @@ -201,6 +202,22 @@ func (s *Server) handleListModules(w http.ResponseWriter, r *http.Request) { } filtered := make([]*store.Module, 0) + limit := parseListLimit(r) + cursor := r.URL.Query().Get("cursor") + if typeFilter == "" && enabledFilter == nil { + page, next, more := s.store.ListModulesPage(a.TenantID, cursor, limit) + for _, mod := range page { + filtered = append(filtered, mod) + } + items := make([]map[string]any, 0, len(filtered)) + for _, mod := range filtered { + items = append(items, moduleJSON(mod)) + } + writeJSON(w, http.StatusOK, map[string]any{ + "items": items, "next_cursor": strPtrOrNull(next), "has_more": more, + }) + return + } for _, mod := range s.store.ListModules(a.TenantID) { if typeFilter != "" && mod.Type != typeFilter { continue @@ -534,7 +551,8 @@ func (s *Server) enqueueModuleRefreshIfEnabled(tenantID, moduleID, trigger strin return } mid := moduleID - _, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, nil, &mid, map[string]any{ + key := "module_refresh:" + moduleID + _, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, &key, &mid, map[string]any{ "module_id": moduleID, "trigger": trigger, }) @@ -601,6 +619,9 @@ func (s *Server) handleRevisionPreview(w http.ResponseWriter, r *http.Request) { for k, v := range rev.PreviewFragments { obj[k] = v } + if expanded := pipeline.BuildExpandedBirdPreview(rev.PreviewFragments); expanded != "" { + obj[pipeline.AuxBirdFullExpandedKey()] = expanded + } writeJSON(w, http.StatusOK, obj) } @@ -855,6 +876,44 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, j.Snapshot()) } +func (s *Server) handleGetJobReport(w http.ResponseWriter, r *http.Request) { + a, ok := authFromContext(r.Context()) + if !ok { + writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth") + return + } + if !s.requireAtLeast(w, a, "viewer") { + return + } + j, err := s.jobs.Get(a.TenantID, r.PathValue("job_id")) + if err != nil { + writeProblem(w, http.StatusNotFound, "Not Found", "job not found") + return + } + snap := j.Snapshot() + meta, _ := snap["meta"].(map[string]any) + out := map[string]any{ + "job_id": snap["job_id"], + "kind": snap["kind"], + "status": snap["status"], + "meta": meta, + "error": snap["error"], + "created_at": snap["created_at"], + } + if meta != nil { + if v, ok := meta["log_entries"]; ok { + out["log_entries"] = v + } + if v, ok := meta["log_total"]; ok { + out["log_total"] = v + } + if v, ok := meta["revision_id"]; ok { + out["revision_id"] = v + } + } + writeJSON(w, http.StatusOK, out) +} + func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) { a, ok := authFromContext(r.Context()) if !ok { diff --git a/internal/jobs/job.go b/internal/jobs/job.go index d2cffd4..821b84f 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -9,6 +9,8 @@ import ( "sync" "time" + "evobgp/internal/observability" + "github.com/google/uuid" ) @@ -178,6 +180,7 @@ type Registry struct { byID map[string]*Job byIdempo map[idempoKey]*Job workerStart func(j *Job) + workerSem chan struct{} } type idempoKey struct { @@ -186,13 +189,22 @@ type idempoKey struct { } func NewRegistry(workerStart func(j *Job)) *Registry { + maxWorkers := registryMaxConcurrentJobs() return &Registry{ byID: make(map[string]*Job), byIdempo: make(map[idempoKey]*Job), workerStart: workerStart, + workerSem: make(chan struct{}, maxWorkers), } } +func registryMaxConcurrentJobs() int { + if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_MAX_CONCURRENT"))); err == nil && n > 0 { + return n + } + return 8 +} + // pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs. func (r *Registry) pruneTerminalIfOver(maxJobs int) { if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs { @@ -244,7 +256,11 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module if idempotencyKey != nil && *idempotencyKey != "" { k := idempoKey{tenant: tenantID, key: *idempotencyKey} if existing, ok := r.byIdempo[k]; ok { - return existing, false, nil + st := existing.statusLocked() + if st == StatusQueued || st == StatusRunning { + return existing, false, nil + } + delete(r.byIdempo, k) } } @@ -265,7 +281,17 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module r.pruneTerminalIfOver(maxJobs) if r.workerStart != nil { - go r.workerStart(j) + go func() { + r.workerSem <- struct{}{} + active := len(r.workerSem) + capacity := cap(r.workerSem) + observability.RecordJobQueueDepth(active, capacity) + defer func() { + <-r.workerSem + observability.RecordJobQueueDepth(len(r.workerSem), capacity) + }() + r.workerStart(j) + }() } return j, true, nil } diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 2e493d0..2063a8c 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -80,6 +80,32 @@ var ( Help: "Prefix row count after CIDR aggregation on tenant render.", Buckets: prometheus.ExponentialBuckets(1, 2, 16), }) + + pipelineRefreshDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "pipeline_refresh_duration_seconds", + Help: "Module refresh ingest duration by module type.", + Buckets: prometheus.ExponentialBuckets(0.05, 2, 14), + }, []string{"module_type"}) + + renderPrefixCount = promauto.NewHistogram(prometheus.HistogramOpts{ + Namespace: namespace, + Name: "render_prefix_count", + Help: "Materialized prefix count per tenant render.", + Buckets: prometheus.ExponentialBuckets(10, 2, 16), + }) + + jobQueueActive = promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Name: "job_queue_active", + Help: "Currently running in-process async jobs.", + }) + + jobQueueCapacity = promauto.NewGauge(prometheus.GaugeOpts{ + Namespace: namespace, + Name: "job_queue_capacity", + Help: "Maximum concurrent in-process async jobs.", + }) ) var ( @@ -99,6 +125,27 @@ func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Durati prefixAggregationDuration.Observe(duration.Seconds()) prefixAggregationRawCount.Observe(float64(rawCount)) prefixAggregationAggregatedCount.Observe(float64(aggregatedCount)) + renderPrefixCount.Observe(float64(aggregatedCount)) +} + +// RecordPipelineRefresh records module ingest duration. +func RecordPipelineRefresh(moduleType string, duration time.Duration) { + if moduleType == "" { + moduleType = "unknown" + } + pipelineRefreshDuration.WithLabelValues(moduleType).Observe(duration.Seconds()) +} + +// RecordJobQueueDepth updates in-process job worker utilization gauges. +func RecordJobQueueDepth(active, capacity int) { + if active < 0 { + active = 0 + } + if capacity < 0 { + capacity = 0 + } + jobQueueActive.Set(float64(active)) + jobQueueCapacity.Set(float64(capacity)) } // RecordJobTerminal increments jobs_finished_total for terminal statuses. diff --git a/internal/repository/job_audit.go b/internal/repository/job_audit.go new file mode 100644 index 0000000..58dfcd8 --- /dev/null +++ b/internal/repository/job_audit.go @@ -0,0 +1,50 @@ +package repository + +import ( + "context" + "encoding/json" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// JobAuditWriter persists async job lifecycle rows to job_audit (optional cross-process queue foundation). +type JobAuditWriter struct { + pool *pgxpool.Pool +} + +func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter { + if pool == nil { + return nil + } + return &JobAuditWriter{pool: pool} +} + +// UpsertRunning inserts or updates a running job row (best-effort). +func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, meta map[string]any) { + if w == nil || w.pool == nil { + return + } + metaJSON, _ := json.Marshal(meta) + var idem any + if idempotencyKey != nil && *idempotencyKey != "" { + idem = *idempotencyKey + } + _, _ = w.pool.Exec(ctx, ` + INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, meta_json, created_at, started_at) + VALUES ($1::uuid, $2::uuid, $3, 'running', $4, $5::jsonb, now(), now()) + ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL + DO UPDATE SET status='running', started_at=now(), meta_json=EXCLUDED.meta_json`, + jobID, tenantID, kind, idem, metaJSON) +} + +// MarkTerminal updates job_audit terminal state (best-effort). +func (w *JobAuditWriter) MarkTerminal(ctx context.Context, tenantID, jobID, status string, errMsg *string, finishedAt time.Time) { + if w == nil || w.pool == nil { + return + } + _, _ = w.pool.Exec(ctx, ` + UPDATE job_audit SET status=$3, error_message=$4, finished_at=$5 + WHERE id=$1::uuid AND tenant_id=$2::uuid`, + jobID, tenantID, status, errMsg, finishedAt.UTC()) +} diff --git a/internal/repository/postgres.go b/internal/repository/postgres.go index ae45823..80dce29 100644 --- a/internal/repository/postgres.go +++ b/internal/repository/postgres.go @@ -162,6 +162,75 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module { return out } +func (p *Postgres) ListModulesPage(tenantID, cursor string, limit int) ([]*store.Module, string, bool) { + if limit <= 0 { + limit = 50 + } + off := 0 + if cursor != "" { + if n, err := strconv.Atoi(cursor); err == nil && n >= 0 { + off = n + } + } + ctx := context.Background() + rows, err := p.pool.Query(ctx, ` + SELECT id, type, name, enabled, priority, doh_profile_id::text, doh_resolver_policy, + refresh_interval_sec, cron_expr, default_community_id::text, last_refreshed_at + FROM module WHERE tenant_id = $1 AND deleted_at IS NULL + ORDER BY priority, name + LIMIT $2 OFFSET $3`, tenantID, limit+1, off) + if err != nil { + return nil, "", false + } + defer rows.Close() + var out []*store.Module + moduleByID := make(map[string]*store.Module) + for rows.Next() { + var m store.Module + m.TenantID = tenantID + var doh, dc, cron *string + var refresh *int32 + var last *time.Time + if err := rows.Scan(&m.ID, &m.Type, &m.Name, &m.Enabled, &m.Priority, &doh, &m.DohResolverPolicy, &refresh, &cron, &dc, &last); err != nil { + continue + } + m.DohResolverPolicy = store.NormalizeDohResolverPolicy(m.DohResolverPolicy) + if refresh != nil { + m.RefreshIntervalSec = int(*refresh) + } + if cron != nil { + m.CronExpr = *cron + } + if doh != nil && *doh != "" { + m.DohProfileID = doh + } + if dc != nil && *dc != "" { + m.DefaultCommunityID = dc + } + if last != nil { + t := last.UTC() + m.LastRefreshedAt = &t + } + out = append(out, &m) + moduleByID[m.ID] = &m + } + if err := p.batchFillModuleDohFields(ctx, moduleByID); err != nil { + return nil, "", false + } + more := len(out) > limit + if more { + out = out[:limit] + } + next := "" + if more { + next = fmt.Sprintf("%d", off+limit) + } + if len(out) == 0 { + return nil, "", false + } + return out, next, more +} + func (p *Postgres) GetModule(tenantID, moduleID string) (*store.Module, error) { ctx := context.Background() var m store.Module @@ -762,6 +831,8 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st return newID, nil } +const maxRevisionDiffRows = 5000 + func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, error) { if _, err := p.GetRevision(tenantID, aID); err != nil { return nil, err @@ -786,7 +857,7 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid EXCEPT SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid - ) s ORDER BY 1`, bID, aID) + ) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1) if err != nil { return nil, err } @@ -798,13 +869,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro continue } added = append(added, s) + if len(added) > maxRevisionDiffRows { + added = added[:maxRevisionDiffRows] + break + } } + addedTruncated := len(added) >= maxRevisionDiffRows rowsRem, err := p.pool.Query(ctx, ` SELECT prefix::text FROM ( SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid EXCEPT SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid - ) s ORDER BY 1`, aID, bID) + ) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1) if err != nil { return nil, err } @@ -816,40 +892,56 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro continue } removed = append(removed, s) + if len(removed) > maxRevisionDiffRows { + removed = removed[:maxRevisionDiffRows] + break + } } return map[string]any{ "revision_a": aID, "revision_b": bID, "prefixes": map[string]any{ "added": added, "removed": removed, "unchanged_count": unchanged, + "truncated": addedTruncated || len(removed) >= maxRevisionDiffRows, }, }, nil } func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) { ctx := context.Background() - cmd, err := p.pool.Exec(ctx, ` - DELETE FROM config_revision AS cr - WHERE cr.tenant_id = $1 - AND cr.created_at < $2 - AND cr.id <> ( - SELECT id - FROM config_revision - WHERE tenant_id = $1 - ORDER BY created_at DESC - LIMIT 1 - ) - AND NOT EXISTS ( - SELECT 1 - FROM bgp_speaker AS sp - WHERE sp.tenant_id = $1 - AND (sp.last_applied_revision_id = cr.id OR sp.published_revision_id = cr.id) - )`, - tenantID, cutoff.UTC()) - if err != nil { - return 0, err + total := 0 + const batchSize = 50 + for { + cmd, err := p.pool.Exec(ctx, ` + DELETE FROM config_revision AS cr + WHERE cr.id IN ( + SELECT id FROM config_revision + WHERE tenant_id = $1 + AND created_at < $2 + AND id <> ( + SELECT id FROM config_revision + WHERE tenant_id = $1 + ORDER BY created_at DESC + LIMIT 1 + ) + AND NOT EXISTS ( + SELECT 1 FROM bgp_speaker AS sp + WHERE sp.tenant_id = $1 + AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id) + ) + ORDER BY created_at ASC + LIMIT $3 + )`, tenantID, cutoff.UTC(), batchSize) + if err != nil { + return total, err + } + n := int(cmd.RowsAffected()) + total += n + if n < batchSize { + break + } } - return int(cmd.RowsAffected()), nil + return total, nil } func (p *Postgres) SetLastAppliedRevision(tenantID, speakerID, revisionID string) error { diff --git a/internal/store/memory.go b/internal/store/memory.go index 4b5956b..02c2f63 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -382,6 +382,11 @@ func (m *Memory) ListModules(tenantID string) []*Module { return out } +func (m *Memory) ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool) { + all := m.ListModules(tenantID) + return PaginateOffset(all, cursor, limit) +} + // ListPeers returns BGP peers for a tenant (sorted by name). func (m *Memory) ListPeers(tenantID string) []*BGPPeer { m.mu.RLock()