package jobs import ( "context" "fmt" "net/http" "net/netip" "os" "sort" "strings" "sync" "time" "evobgp/internal/birddeploy" "evobgp/internal/birdfmt" "evobgp/internal/httpclient" "evobgp/internal/nodedispatch" "evobgp/internal/observability" "evobgp/internal/pipeline" "evobgp/internal/store" "github.com/jackc/pgx/v5/pgxpool" ) // mergeBirdPostApplyMeta attaches a birdc snapshot after deploy/reload (best-effort). func mergeBirdPostApplyMeta(j *Job) { if strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) == "" { j.mergeMeta(map[string]any{"bird_post_apply_check": "skipped_no_birdc_socket"}) return } ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second) defer cancel() st := birdfmt.InspectLocalBird(ctx) inner := map[string]any{ "bgp_established": st.BGPEstablished, "bgp_sessions_total": st.BGPSessionsTotal, } if st.Error != "" { inner["ok"] = false inner["error"] = st.Error } else { inner["ok"] = true } j.mergeMeta(map[string]any{"bird_post_apply": inner}) } const ( KindModuleRefresh = "module_refresh" KindTenantRefresh = "tenant_refresh" KindPeerReconcile = "peer_reconcile" KindDeployApply = "deploy_apply" KindRevisionRollback = "revision_rollback" KindBirdReload = "bird_reload" KindPostgresMetricsRefresh = "postgres_metrics_refresh" KindPostgresSlowQueryAgg = "postgres_slow_query_aggregate" KindPostgresTableBloat = "postgres_table_bloat_estimate" KindPostgresIndexUsage = "postgres_index_usage_analyze" KindPostgresAutovacuumLag = "postgres_autovacuum_lag_detect" KindPostgresVacuum = "postgres_vacuum" KindPostgresVacuumAnalyze = "postgres_vacuum_analyze" KindPostgresAnalyze = "postgres_analyze" KindPostgresReindex = "postgres_reindex" KindPostgresCleanup = "postgres_cleanup" KindMaintenancePolicyRun = "maintenance_policy_run" ) // Worker executes queued jobs against store.Backend (memory or SQL). type Worker struct { Store store.Backend PgPool *pgxpool.Pool HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout). // Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback. Registry *Registry // refreshGate serializes deploy_apply gating after module_refresh per tenant (see finishModuleRefreshSuccess). refreshGate sync.Map // map[string]*sync.Mutex } type revisionLogEntry struct { Kind string `json:"kind"` Source string `json:"source"` Community string `json:"community"` CommunityLabel string `json:"community_label"` PrefixCount int `json:"prefix_count"` Sample []string `json:"sample,omitempty"` Message string `json:"message"` } var defaultWorkerHTTP = httpclient.New(httpclient.DefaultTimeout) func (w *Worker) httpClient() *http.Client { if w != nil && w.HTTPClient != nil { return w.HTTPClient } return defaultWorkerHTTP } // Process is registered as Registry.workerStart. func (w *Worker) Process(j *Job) { defer func() { observability.RecordJobTerminal(j.Kind, j.statusLocked()) if w != nil && w.Registry != nil { w.Registry.fireTerminal(j) } }() if w == nil || w.Store == nil { j.MarkRunning() j.Fail("worker not configured") return } j.MarkRunning() if w != nil && w.Registry != nil { w.Registry.fireRunning(j) } if j.IsCancelRequested() { j.MarkCancelled() return } switch j.Kind { case KindModuleRefresh: w.runModuleRefresh(j) case KindTenantRefresh: w.runTenantRefresh(j) case KindPeerReconcile: w.runPeerReconcile(j) case KindDeployApply: w.runDeployApply(j) case KindRevisionRollback: w.runRollback(j) case KindBirdReload: sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")) if sock == "" { j.Succeed() return } ctx, cancel := j.workContext() defer cancel() ctl := &birdfmt.BirdCtl{ Socket: sock, Birdc: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")), } if err := ctl.Configure(ctx); err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } mergeBirdPostApplyMeta(j) j.Succeed() case KindPostgresMetricsRefresh: w.runPostgresMetricsRefresh(j) case KindPostgresSlowQueryAgg: w.runPostgresSlowQueryAgg(j) case KindPostgresTableBloat: w.runPostgresTableBloat(j) case KindPostgresIndexUsage: w.runPostgresIndexUsage(j) case KindPostgresAutovacuumLag: w.runPostgresAutovacuumLag(j) case KindPostgresVacuum: w.runPostgresMaint(j, "vacuum") case KindPostgresVacuumAnalyze: w.runPostgresMaint(j, "vacuum_analyze") case KindPostgresAnalyze: w.runPostgresMaint(j, "analyze") case KindPostgresReindex: w.runPostgresMaint(j, "reindex") case KindPostgresCleanup: w.runPostgresCleanup(j) case KindMaintenancePolicyRun: w.runMaintenancePolicy(j) default: j.Fail("unknown job kind") } } func (w *Worker) runPeerReconcile(j *Job) { if w == nil || w.Store == nil { j.Fail("worker not configured") return } const peerJobTitle = "Обновление BGP пиров" j.mergeMeta(map[string]any{"job_title": peerJobTitle}) var revID string latest, _, _ := w.Store.ListRevisions(j.TenantID, "", "", 1) triggerModuleID, err := w.peerTriggerModuleID(j.TenantID, latest) if err != nil { j.Fail(err.Error()) return } if len(latest) == 0 { ctx, cancel := j.workContext() defer cancel() rid, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID) if err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } revID = rid } else { baseRevID := latest[0].ID rows := make([]store.PrefixRow, 0, 1024) cursor := "" for { page, next, more := w.Store.ListRevisionPrefixes(j.TenantID, baseRevID, cursor, 2000) rows = append(rows, page...) if !more || strings.TrimSpace(next) == "" { break } cursor = next } ctx, cancel := j.workContext() defer cancel() rid, err := pipeline.RenderTenantRevisionFromPrefixes(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID, rows) if err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } revID = rid } j.mergeMeta(map[string]any{"revision_id": revID}) if entries, total, err := w.buildRevisionLogEntries(j.TenantID, revID); err == nil { j.mergeMeta(map[string]any{ "log_entries": entries, "log_total": total, "log_generated": time.Now().UTC().Format(time.RFC3339Nano), }) } else { j.mergeMeta(map[string]any{"log_build_error": err.Error()}) } w.enqueueDeployAllSpeakers(j, j.TenantID, revID) j.Succeed() } func (w *Worker) peerTriggerModuleID(tenantID string, latest []*store.Revision) (string, error) { if len(latest) > 0 { if mid := strings.TrimSpace(latest[0].ModuleID); mid != "" { return mid, nil } } for _, mod := range w.Store.ListModules(tenantID) { if mod == nil || !mod.Enabled { continue } if strings.TrimSpace(mod.ID) != "" { return mod.ID, nil } } for _, mod := range w.Store.ListModules(tenantID) { if mod == nil { continue } if strings.TrimSpace(mod.ID) != "" { return mod.ID, nil } } return "", fmt.Errorf("missing module_id for peer reconcile") } func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex { v, _ := w.refreshGate.LoadOrStore(tenantID, &sync.Mutex{}) return v.(*sync.Mutex) } // finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant, // creates one aggregate revision and enqueues a single deploy_apply. // runModuleRefresh handles a single module_refresh job and guarantees the per-tenant inflight // slot is released exactly once — even on failure/cancellation before finishModuleRefreshSuccess. func (w *Worker) runModuleRefresh(j *Job) { coalesceFinalized := false defer func() { if !coalesceFinalized && w != nil && w.Registry != nil { // Refresh failed/was cancelled before reaching finishModuleRefreshSuccess. // Decrement the counter under the tenant mutex so the "last one" logic stays sound. mu := w.tenantRefreshMu(j.TenantID) mu.Lock() w.Registry.finalizeRefreshCoalesce(j.TenantID) mu.Unlock() } }() mid := strings.TrimSpace(j.metaString("module_id")) if mid == "" { j.Fail("missing module_id in job meta") return } ctx, cancel := j.workContext() defer cancel() if ctx.Err() != nil { j.MarkCancelled() return } if err := pipeline.RefreshModuleIngest(ctx, w.Store, w.httpClient(), j.TenantID, mid); err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } w.finishModuleRefreshSuccess(j, mid) coalesceFinalized = true } func (w *Worker) runTenantRefresh(j *Job) { coalesceFinalized := false defer func() { if !coalesceFinalized && w != nil && w.Registry != nil { mu := w.tenantRefreshMu(j.TenantID) mu.Lock() w.Registry.finalizeRefreshCoalesce(j.TenantID) mu.Unlock() } }() moduleIDs := moduleIDsFromJobMeta(j.metaCopy()) if len(moduleIDs) == 0 { j.Fail("missing module_ids in job meta") return } ctx, cancel := j.workContext() defer cancel() if err := pipeline.RefreshTenantModules(ctx, w.Store, w.httpClient(), j.TenantID, moduleIDs); err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } trigger, err := pipeline.PickTenantRefreshTriggerModule(w.Store, j.TenantID, moduleIDs) if err != nil { j.Fail(err.Error()) return } j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)}) w.finishModuleRefreshSuccess(j, trigger) coalesceFinalized = true } func moduleIDsFromJobMeta(meta map[string]any) []string { if meta == nil { return nil } raw, ok := meta["module_ids"] if !ok { return nil } switch v := raw.(type) { case []string: return v case []any: var out []string for _, x := range v { if s, ok := x.(string); ok && strings.TrimSpace(s) != "" { out = append(out, strings.TrimSpace(s)) } } return out default: return nil } } func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) { if w == nil || w.Store == nil { if w != nil && w.Registry != nil { w.Registry.finalizeRefreshCoalesce(j.TenantID) } j.Succeed() return } mu := w.tenantRefreshMu(j.TenantID) mu.Lock() defer mu.Unlock() ctx, cancel := j.workContext() defer cancel() // Determine whether this is the last outstanding refresh for the tenant. The counter is // incremented in Enqueue (under r.mu) and decremented here, so the "last one" decision is // deterministic regardless of ingest timing — unlike the previous status-polling approach // (CountOtherActiveRefresh) which could race under -race. isLastRefresh := true if w.Registry != nil { isLastRefresh = w.Registry.finalizeRefreshCoalesce(j.TenantID) } if !isLastRefresh { j.mergeMeta(map[string]any{ "deploy_apply_deferred": true, "deploy_apply_defer_reason": "parallel_module_refresh", }) j.Succeed() return } rev, err := pipeline.RenderTenantRevision(ctx, w.Store, w.httpClient(), j.TenantID, triggerModuleID) if err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } j.mergeMeta(map[string]any{"revision_id": rev}) if entries, total, err := w.buildRevisionLogEntries(j.TenantID, rev); err == nil { j.mergeMeta(map[string]any{ "log_entries": entries, "log_total": total, "log_generated": time.Now().UTC().Format(time.RFC3339Nano), }) } else { j.mergeMeta(map[string]any{"log_build_error": err.Error()}) } w.enqueueDeployAllSpeakers(j, j.TenantID, rev) j.Succeed() } // enqueueDeployAllSpeakers queues the same work as POST /v1/apply (all speakers, no speaker_id). func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) { if w == nil || w.Registry == nil { return } revID = strings.TrimSpace(revID) if revID == "" { return } applyJob, _, err := w.Registry.Enqueue(tenantID, KindDeployApply, nil, nil, map[string]any{ "revision_id": revID, }) if err != nil { j.mergeMeta(map[string]any{"deploy_apply_enqueue_error": err.Error()}) return } if applyJob != nil { j.mergeMeta(map[string]any{"deploy_apply_job_id": applyJob.ID}) } } func (w *Worker) runDeployApply(j *Job) { revID := strings.TrimSpace(j.metaString("revision_id")) spk := strings.TrimSpace(j.metaString("speaker_id")) hasSpeaker := spk != "" if revID == "" { j.Fail("missing revision_id in job meta") return } ctx, cancel := j.workContext() defer cancel() activeDir := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR")) if activeDir != "" { revObj, err := w.Store.GetRevision(j.TenantID, revID) if err != nil { j.Fail(err.Error()) return } staging := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_STAGING_DIR")) if staging == "" { staging = os.TempDir() + "/evobgp-bird-staging" } cfg := birddeploy.Config{ ActiveDir: activeDir, StagingDir: staging, BirdBin: strings.TrimSpace(os.Getenv("EVOBGP_BIRD_BIN")), BirdcBin: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")), Socket: strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")), } ctl := &birdfmt.BirdCtl{Bird: cfg.BirdBin, Birdc: cfg.BirdcBin, Socket: cfg.Socket} if err := birddeploy.ApplyRevision(ctx, ctl, revObj, cfg); err != nil { if ctx.Err() != nil { j.MarkCancelled() return } j.Fail(err.Error()) return } } applied := make([]string, 0, 8) var dispatchResults []nodedispatch.Result applyOne := func(speakerID string) error { if err := w.Store.SetLastAppliedRevision(j.TenantID, speakerID, revID); err != nil { return err } // Replica / node pulls use LatestPublishedRevision; keep pointer in sync with successful deploy. if err := w.Store.PublishRevisionForSpeaker(speakerID, revID); err != nil { return err } applied = append(applied, speakerID) return nil } dispatchSpeaker := func(sp *store.Speaker) { if !nodedispatch.Enabled() || sp == nil { return } meta := store.ParseSpeakerMeta(sp.MetaJSON) if !store.SpeakerNeedsRemoteDispatch(sp.Role, meta) { return } ctx2, cancel := context.WithTimeout(ctx, 35*time.Second) defer cancel() res := nodedispatch.WakeSpeaker(ctx2, sp, nodedispatch.Options{RevisionID: revID}) dispatchResults = append(dispatchResults, res) patch := store.SpeakerMeta{ LastDispatchAt: time.Now().UTC().Format(time.RFC3339Nano), LastDispatchStatus: res.Status, } if res.Error != "" { patch.LastDispatchError = res.Error patch.SyncStatus = "error" } else if res.Status == "ok" { patch.LastDispatchError = "" patch.SyncStatus = "synced" } merged := store.MergeSpeakerMetaJSON(sp.MetaJSON, patch) _, _ = w.Store.UpdateSpeaker(j.TenantID, sp.ID, &store.SpeakerPatch{MetaJSON: &merged}) } if hasSpeaker && spk != "" { if err := applyOne(spk); err != nil { j.Fail(err.Error()) return } if sp, err := w.Store.GetSpeaker(j.TenantID, spk); err == nil { dispatchSpeaker(sp) } if len(dispatchResults) > 0 { j.mergeMeta(map[string]any{"node_dispatch": map[string]any{ "revision_id": revID, "results": dispatchResults, }}) } mergeBirdPostApplyMeta(j) j.Succeed() return } speakers := w.Store.ListSpeakersForTenant(j.TenantID) for _, sp := range speakers { if err := applyOne(sp.ID); err != nil { j.Fail(err.Error()) return } } for _, sp := range speakers { dispatchSpeaker(sp) } j.mergeMeta(map[string]any{ "apply_summary": map[string]any{ "revision_id": revID, "speakers_count": len(applied), "speaker_ids": applied, "message": fmt.Sprintf("Ревизия %s применена на %d спикерах", shortID(revID), len(applied)), }, }) if len(dispatchResults) > 0 { j.mergeMeta(map[string]any{"node_dispatch": map[string]any{ "revision_id": revID, "results": dispatchResults, }}) } mergeBirdPostApplyMeta(j) j.Succeed() } func (w *Worker) runRollback(j *Job) { src := strings.TrimSpace(j.metaString("source_revision_id")) if src == "" { j.Fail("missing source_revision_id in job meta") return } newID, err := w.Store.CreateRollbackRevision(j.TenantID, src) if err != nil { j.Fail(err.Error()) return } j.mergeMeta(map[string]any{"new_revision_id": newID}) j.mergeMeta(map[string]any{ "rollback_summary": map[string]any{ "source_revision_id": src, "new_revision_id": newID, "message": fmt.Sprintf("Откат %s → %s", shortID(src), shortID(newID)), }, }) w.enqueueDeployAllSpeakers(j, j.TenantID, newID) j.Succeed() } // buildCommunityLabelMap maps community UUID -> human-readable title (or BGP community string). func buildCommunityLabelMap(st store.Backend, tenantID string) map[string]string { out := make(map[string]string) if st == nil { return out } list, err := st.ListCommunities(tenantID) if err != nil || list == nil { return out } for _, c := range list { if c == nil { continue } label := strings.TrimSpace(c.Title) if label == "" { label = strings.TrimSpace(c.Community) } if label == "" { label = c.ID } out[c.ID] = label } return out } func resolveCommunityLabel(commID string, byID map[string]string) string { if commID == "" || commID == "none" { return "без community" } if lbl, ok := byID[commID]; ok && strings.TrimSpace(lbl) != "" { return strings.TrimSpace(lbl) } return commID } func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]any, int, error) { if w == nil || w.Store == nil { return nil, 0, fmt.Errorf("store not configured") } commLabels := buildCommunityLabelMap(w.Store, tenantID) type agg struct { kind string source string community string count int sample []string } groups := map[string]*agg{} 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) } } if !more { break } cursor = next if strings.TrimSpace(cursor) == "" { break } } keys := make([]string, 0, len(groups)) for k := range groups { keys = append(keys, k) } sort.Strings(keys) out := make([]map[string]any, 0, len(keys)) for _, k := range keys { g := groups[k] 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, }) } return out, total, nil } func classifySource(src string) (kind, name string) { switch { case strings.HasPrefix(src, "as:"): return "asn", strings.TrimPrefix(src, "as:") case strings.HasPrefix(src, "domain:"): return "domain", strings.TrimPrefix(src, "domain:") case strings.HasPrefix(src, "cdn:"): return "cdn", strings.TrimPrefix(src, "cdn:") case src == "ip_range": return "ip_range", "manual_ranges" default: if src == "" { return "unknown", "unknown" } return "source", src } } // humanLogMessage builds a Russian log line; communityLabel is already resolved (title or BGP value). func humanLogMessage(kind, source string, count int, communityLabel string, sample []string) string { switch kind { case "asn": return fmt.Sprintf("AS%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel) case "domain": ips := strings.Join(prettyDomainSample(sample), " ") if ips == "" { ips = "—" } return fmt.Sprintf("%s: IP (%s) → добавлено в сообщество «%s»", source, ips, communityLabel) case "cdn": return fmt.Sprintf("CDN «%s»: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel) case "ip_range": return fmt.Sprintf("Статические диапазоны: добавлено %d префиксов в сообщество «%s»", count, communityLabel) default: return fmt.Sprintf("%s: добавлено %d префиксов в сообщество «%s»", source, count, communityLabel) } } func prettyDomainSample(sample []string) []string { out := make([]string, 0, len(sample)) for _, s := range sample { p, err := netip.ParsePrefix(strings.TrimSpace(s)) if err != nil { out = append(out, s) continue } if (p.Addr().Is4() && p.Bits() == 32) || (p.Addr().Is6() && p.Bits() == 128) { out = append(out, p.Addr().String()) continue } out = append(out, s) } return out } func shortID(id string) string { s := strings.TrimSpace(id) if len(s) <= 8 { return s } return s[:8] }