From 8fe74c1d3b90d5b49c260f4c501bb8776d3be81e Mon Sep 17 00:00:00 2001 From: Denozordec Date: Fri, 31 Jul 2026 12:26:18 +0700 Subject: [PATCH] feat(jobs): add durable PG queue reclaim, slog, and richer metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit JSON slog в ключевых пакетах; Prometheus path_group, job_audit_depth, upstream breaker; job_audit ClaimQueued/ReclaimStaleRunning + Adopt loop для HA после рестарта. Co-authored-by: Cursor --- cmd/evobgp-api/main.go | 19 ++-- internal/broker/connect.go | 5 +- internal/deploy/run.go | 12 ++- internal/httpapi/server.go | 4 + internal/httpclient/circuit.go | 18 ++++ internal/httpclient/httpclient.go | 11 +- internal/ingest/run.go | 8 +- internal/jobs/durable.go | 137 +++++++++++++++++++++++++ internal/logging/logger.go | 38 +++++++ internal/maintenance/scheduler.go | 6 +- internal/observability/metrics.go | 58 ++++++++++- internal/pgmonitor/scheduler.go | 15 +-- internal/pipeline/collect_parallel.go | 4 +- internal/pipeline/collect_stale.go | 4 +- internal/render/publish.go | 7 +- internal/render/run.go | 8 +- internal/repository/job_audit_claim.go | 130 +++++++++++++++++++++++ internal/scheduler/run.go | 15 +-- 18 files changed, 448 insertions(+), 51 deletions(-) create mode 100644 internal/jobs/durable.go create mode 100644 internal/logging/logger.go create mode 100644 internal/repository/job_audit_claim.go diff --git a/cmd/evobgp-api/main.go b/cmd/evobgp-api/main.go index 37b1811..7637a31 100644 --- a/cmd/evobgp-api/main.go +++ b/cmd/evobgp-api/main.go @@ -14,6 +14,7 @@ import ( "evobgp/internal/config" "evobgp/internal/dbcli" "evobgp/internal/httpapi" + "evobgp/internal/logging" "evobgp/internal/observability" "evobgp/internal/platform" "evobgp/internal/version" @@ -65,13 +66,13 @@ func main() { } go func() { svc := platform.ServiceName("evobgp-api") - log.Printf("%s listening on %s", svc, cfg.HTTPAddr) - log.Printf("bundle signing public key (base64, set on evobgp-node): %s", srv.BundlePublicKeyBase64()) + logging.Default().Info("listening", "service", svc, "addr", cfg.HTTPAddr) + logging.Default().Info("bundle signing public key (base64, set on evobgp-node)", "key", srv.BundlePublicKeyBase64()) if opts.SeedDemo { tid, mCDN, mIP, rev, sp := srv.Store().DemoIDs() - log.Printf("demo tenant=%s module_cdn=%s module_ip_ranges=%s revision=%s speaker=%s", tid, mCDN, mIP, rev, sp) - log.Printf("example: EVOBGP_API_KEYS=op|%s|operator,node|%s|node", tid, tid) - log.Printf("demo auth: Authorization: Bearer dev (operator, demo tenant only)") + logging.Default().Info("demo ids", "tenant", tid, "module_cdn", mCDN, "module_ip_ranges", mIP, "revision", rev, "speaker", sp) + logging.Default().Info("example API keys", "hint", "EVOBGP_API_KEYS=op|"+tid+"|operator,node|"+tid+"|node") + logging.Default().Info("demo auth: Authorization: Bearer dev (operator, demo tenant only)") } if err := httpSrv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal(err) @@ -83,15 +84,15 @@ func main() { defer cancel() n := srv.Jobs().RequestCancelAll() if n > 0 { - log.Printf("draining %d job(s)…", n) + logging.Default().Info("draining jobs", "count", n) if err := srv.Jobs().Drain(shutdownCtx); err != nil { - log.Printf("job drain: %v", err) + logging.Default().Warn("job drain", "err", err) } } if err := httpSrv.Shutdown(shutdownCtx); err != nil { - log.Printf("HTTP shutdown: %v", err) + logging.Default().Warn("HTTP shutdown", "err", err) } - log.Printf("%s stopped", platform.ServiceName("evobgp-api")) + logging.Default().Info("stopped", "service", platform.ServiceName("evobgp-api")) } func firstNonEmpty(candidates ...string) string { diff --git a/internal/broker/connect.go b/internal/broker/connect.go index 72d5465..ce80670 100644 --- a/internal/broker/connect.go +++ b/internal/broker/connect.go @@ -4,7 +4,8 @@ package broker import ( "context" - "log" + "evobgp/internal/logging" + "fmt" "strings" ) @@ -14,6 +15,6 @@ func LogConnect(ctx context.Context, brokerURL string) { if u == "" { return } - log.Printf("broker: EVOBGP_BROKER_URL=%q set; control plane still uses in-process jobs.Registry (no JetStream/Redis consumer in this binary)", u) + logging.Default().Info(fmt.Sprintf("broker: EVOBGP_BROKER_URL=%q set; control plane still uses in-process jobs.Registry (no JetStream/Redis consumer in this binary)", u)) _ = ctx } diff --git a/internal/deploy/run.go b/internal/deploy/run.go index 300c370..0127f05 100644 --- a/internal/deploy/run.go +++ b/internal/deploy/run.go @@ -2,6 +2,8 @@ package deploy import ( "context" + "evobgp/internal/logging" + "fmt" "log" "os" "strings" @@ -22,18 +24,18 @@ func Run(ctx context.Context, deps *Deps) { cfg := config.Load() broker.LogConnect(ctx, cfg.BrokerURL) if d := strings.TrimSpace(os.Getenv("EVOBGP_BIRD_ACTIVE_DIR")); d != "" { - log.Printf("evobgp-deploy: EVOBGP_BIRD_ACTIVE_DIR=%q (apply via API jobs when API has same env)", d) + logging.Default().Info(fmt.Sprintf("evobgp-deploy: EVOBGP_BIRD_ACTIVE_DIR=%q (apply via API jobs when API has same env)", d)) } if deps == nil || deps.Store == nil { log.Fatalf("evobgp-deploy: missing store (pass deploy.Deps from BootstrapWorkers or evobgp-all)") } t := time.NewTicker(90 * time.Second) defer t.Stop() - log.Printf("evobgp-deploy: active (speaker published vs applied drift log)") + logging.Default().Info(fmt.Sprintf("evobgp-deploy: active (speaker published vs applied drift log)")) for { select { case <-ctx.Done(): - log.Printf("evobgp-deploy: stopped") + logging.Default().Info(fmt.Sprintf("evobgp-deploy: stopped")) return case <-t.C: logDrift(context.Background(), deps.Store) @@ -45,7 +47,7 @@ func logDrift(ctx context.Context, st store.Backend) { _ = ctx tenants, err := st.ListTenantIDs() if err != nil { - log.Printf("evobgp-deploy: list tenants: %v", err) + logging.Default().Info(fmt.Sprintf("evobgp-deploy: list tenants: %v", err)) return } for _, tid := range tenants { @@ -59,7 +61,7 @@ func logDrift(ctx context.Context, st store.Backend) { applied = *sp.LastAppliedRevisionID } if applied != "" && applied != pub { - log.Printf("evobgp-deploy: drift speaker=%s applied=%s published=%s", sp.ID, applied, pub) + logging.Default().Info(fmt.Sprintf("evobgp-deploy: drift speaker=%s applied=%s published=%s", sp.ID, applied, pub)) } } } diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index a319b61..7b7b98b 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -14,6 +14,7 @@ import ( "evobgp/internal/jobs" "evobgp/internal/maintenance" "evobgp/internal/pgmonitor" + "evobgp/internal/repository" "evobgp/internal/runtimelogs" "evobgp/internal/store" @@ -156,6 +157,9 @@ func (s *Server) Jobs() *jobs.Registry { return s.jobs } func (s *Server) StartBackground(ctx context.Context) { if s != nil && s.pgPool != nil { pgmonitor.StartScheduler(ctx, s.pgPool) + if s.jobs != nil { + jobs.StartDurableQueueLoop(ctx, s.jobs, repository.NewJobAuditWriter(s.pgPool)) + } } if s != nil && s.maintConfig != nil && s.jobs != nil { maintenance.StartScheduler(ctx, s.maintConfig, func(policyID string, dryRun bool, idem string) { diff --git a/internal/httpclient/circuit.go b/internal/httpclient/circuit.go index 949d8ba..89dac5b 100644 --- a/internal/httpclient/circuit.go +++ b/internal/httpclient/circuit.go @@ -49,6 +49,24 @@ func (b *hostBreaker) recordFailure() { } } +// SnapshotBreakers returns whether each known host breaker is currently open. +func SnapshotBreakers() map[string]bool { + out := map[string]bool{} + hostBreakers.Range(func(key, value any) bool { + host, _ := key.(string) + b, _ := value.(*hostBreaker) + if b == nil { + return true + } + b.mu.Lock() + open := time.Now().Before(b.openUntil) + b.mu.Unlock() + out[host] = open + return true + }) + return out +} + // ResetHostBreakers clears all circuit breakers (tests only). func ResetHostBreakers() { hostBreakers = sync.Map{} diff --git a/internal/httpclient/httpclient.go b/internal/httpclient/httpclient.go index 1edfa81..2cb7102 100644 --- a/internal/httpclient/httpclient.go +++ b/internal/httpclient/httpclient.go @@ -7,6 +7,8 @@ import ( "io" "net/http" "time" + + "evobgp/internal/observability" ) const DefaultTimeout = 45 * time.Second @@ -69,19 +71,24 @@ func DoWithBreaker(ctx context.Context, hc *http.Client, req *http.Request, maxA if req == nil || req.URL == nil { return nil, fmt.Errorf("httpclient: nil request") } - br := breakerForHost(req.URL.Hostname()) + host := req.URL.Hostname() + br := breakerForHost(host) if !br.allow() { - return nil, fmt.Errorf("httpclient: circuit open for %s", req.URL.Hostname()) + observability.SetUpstreamBreakerOpen(host, true) + return nil, fmt.Errorf("httpclient: circuit open for %s", host) } resp, err := DoWithRetry(ctx, hc, req, maxAttempts) if err != nil { br.recordFailure() + observability.SetUpstreamBreakerOpen(host, !br.allow()) return nil, err } if resp.StatusCode >= 500 { br.recordFailure() + observability.SetUpstreamBreakerOpen(host, !br.allow()) return resp, nil } br.recordSuccess() + observability.SetUpstreamBreakerOpen(host, false) return resp, nil } diff --git a/internal/ingest/run.go b/internal/ingest/run.go index 58a1579..463ffca 100644 --- a/internal/ingest/run.go +++ b/internal/ingest/run.go @@ -2,6 +2,8 @@ package ingest import ( "context" + "evobgp/internal/logging" + "fmt" "log" "time" @@ -27,18 +29,18 @@ func Run(ctx context.Context, deps *Deps) { hc := httpclient.New(httpclient.DefaultTimeout) t := time.NewTicker(60 * time.Second) defer t.Stop() - log.Printf("evobgp-ingest: active (CDN conditional GET / ETag prefetch)") + logging.Default().Info(fmt.Sprintf("evobgp-ingest: active (CDN conditional GET / ETag prefetch)")) for { select { case <-ctx.Done(): - log.Printf("evobgp-ingest: stopped") + logging.Default().Info(fmt.Sprintf("evobgp-ingest: stopped")) return case <-t.C: prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second) err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc) cancel() if err != nil { - log.Printf("evobgp-ingest: prefetch: %v", err) + logging.Default().Info(fmt.Sprintf("evobgp-ingest: prefetch: %v", err)) } } } diff --git a/internal/jobs/durable.go b/internal/jobs/durable.go new file mode 100644 index 0000000..c2ea9c3 --- /dev/null +++ b/internal/jobs/durable.go @@ -0,0 +1,137 @@ +package jobs + +import ( + "context" + "os" + "strconv" + "strings" + "time" + + "evobgp/internal/logging" + "evobgp/internal/observability" + "evobgp/internal/repository" +) + +// Adopt registers a durable job_audit claim into the in-process registry and starts the worker. +// Used after SKIP LOCKED claim so two API processes can share the queue without losing work on restart. +func (r *Registry) Adopt(j *Job) bool { + if r == nil || j == nil || j.ID == "" { + return false + } + r.mu.Lock() + if _, exists := r.byID[j.ID]; exists { + r.mu.Unlock() + return false + } + if j.Status == "" { + j.Status = StatusQueued + } + if j.Meta == nil { + j.Meta = map[string]any{} + } + r.byID[j.ID] = j + if j.IdempotencyKey != nil && *j.IdempotencyKey != "" { + r.byIdempo[idempoKey{tenant: j.TenantID, key: *j.IdempotencyKey}] = j + } + if isRefreshKind(j.Kind) { + r.inflightRefresh[j.TenantID]++ + } + workerStart := r.workerStart + r.mu.Unlock() + + if workerStart != nil { + 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) + }() + workerStart(j) + }() + } + return true +} + +// StartDurableQueueLoop periodically reclaims stale running rows and claims queued job_audit work (PG SKIP LOCKED). +// No-op when audit is nil. Interval from EVOBGP_JOB_RECLAIM_INTERVAL (default 20s); stale from EVOBGP_JOB_STALE_AFTER (default 15m). +func StartDurableQueueLoop(ctx context.Context, reg *Registry, audit *repository.JobAuditWriter) { + if ctx == nil || reg == nil || audit == nil { + return + } + interval := 20 * time.Second + if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_RECLAIM_INTERVAL")); s != "" { + if d, err := time.ParseDuration(s); err == nil && d > 0 { + interval = d + } + } + staleAfter := 15 * time.Minute + if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_STALE_AFTER")); s != "" { + if d, err := time.ParseDuration(s); err == nil && d > 0 { + staleAfter = d + } + } + limit := 8 + if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_CLAIM_LIMIT"))); err == nil && n > 0 { + limit = n + } + grace := 30 * time.Second + if s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_CLAIM_GRACE")); s != "" { + if d, err := time.ParseDuration(s); err == nil && d >= 0 { + grace = d + } + } + log := logging.With("component", "jobs.durable") + run := func() { + cctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + n, err := audit.ReclaimStaleRunning(cctx, staleAfter) + if err != nil { + log.Warn("reclaim stale running failed", "err", err) + } else if n > 0 { + log.Info("reclaimed stale running jobs", "count", n) + } + claimed, err := audit.ClaimQueued(cctx, limit, grace) + if err != nil { + log.Warn("claim queued failed", "err", err) + return + } + for i := range claimed { + c := claimed[i] + j := &Job{ + ID: c.ID, + TenantID: c.TenantID, + Kind: c.Kind, + Status: StatusQueued, + IdempotencyKey: c.IdempotencyKey, + ModuleID: c.ModuleID, + CreatedAt: c.CreatedAt, + Meta: c.Meta, + } + if !reg.Adopt(j) { + // Already local — leave DB running; local worker owns it. + continue + } + log.Info("adopted durable job", "job_id", j.ID, "kind", j.Kind, "tenant_id", j.TenantID) + } + counts, err := audit.CountByStatus(cctx) + if err == nil { + observability.RecordJobAuditDepth(counts) + } + } + go func() { + run() + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + run() + } + } + }() +} diff --git a/internal/logging/logger.go b/internal/logging/logger.go new file mode 100644 index 0000000..6710c40 --- /dev/null +++ b/internal/logging/logger.go @@ -0,0 +1,38 @@ +// Package logging provides a process-wide slog JSON logger for EvoBGP binaries. +package logging + +import ( + "log/slog" + "os" + "strings" + "sync" +) + +var ( + once sync.Once + logger *slog.Logger +) + +// Default returns the process JSON slog logger (stdout). Level from EVOBGP_LOG_LEVEL (debug|info|warn|error). +func Default() *slog.Logger { + once.Do(func() { + level := slog.LevelInfo + switch strings.ToLower(strings.TrimSpace(os.Getenv("EVOBGP_LOG_LEVEL"))) { + case "debug": + level = slog.LevelDebug + case "warn", "warning": + level = slog.LevelWarn + case "error": + level = slog.LevelError + } + handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: level}) + logger = slog.New(handler) + slog.SetDefault(logger) + }) + return logger +} + +// With returns Default().With(args...). +func With(args ...any) *slog.Logger { + return Default().With(args...) +} diff --git a/internal/maintenance/scheduler.go b/internal/maintenance/scheduler.go index 01f3947..308f89b 100644 --- a/internal/maintenance/scheduler.go +++ b/internal/maintenance/scheduler.go @@ -2,8 +2,8 @@ package maintenance import ( "context" + "evobgp/internal/logging" "fmt" - "log" "strings" "sync" "time" @@ -35,7 +35,7 @@ func StartScheduler(ctx context.Context, provider *ConfigProvider, enqueue func( } sched, err := parser.Parse(p.Schedule) if err != nil { - log.Printf("maintenance: invalid cron for policy %s: %v", p.ID, err) + logging.Default().Info(fmt.Sprintf("maintenance: invalid cron for policy %s: %v", p.ID, err)) continue } schedules[p.ID] = sched @@ -81,5 +81,5 @@ func StartScheduler(ctx context.Context, provider *ConfigProvider, enqueue func( } } }() - log.Printf("maintenance: policy scheduler started (tick=%s)", tick) + logging.Default().Info(fmt.Sprintf("maintenance: policy scheduler started (tick=%s)", tick)) } diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go index 2063a8c..b8995e2 100644 --- a/internal/observability/metrics.go +++ b/internal/observability/metrics.go @@ -5,6 +5,7 @@ import ( "context" "net/http" "strconv" + "strings" "sync" "sync/atomic" "time" @@ -30,7 +31,7 @@ var ( Name: "http_requests_total", Help: "HTTP requests handled by the API mux (excludes /metrics).", }, - []string{"method", "code"}, + []string{"method", "code", "path_group"}, ) jobsFinished = promauto.NewCounterVec( @@ -106,6 +107,18 @@ var ( Name: "job_queue_capacity", Help: "Maximum concurrent in-process async jobs.", }) + + jobAuditDepth = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: namespace, + Name: "job_audit_depth", + Help: "Rows in job_audit by status (durable queue).", + }, []string{"status"}) + + upstreamBreakerOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: namespace, + Name: "upstream_breaker_open", + Help: "1 if per-host HTTP circuit breaker is open.", + }, []string{"host"}) ) var ( @@ -158,6 +171,28 @@ func RecordJobTerminal(kind, status string) { } } +// RecordJobAuditDepth updates durable job_audit depth gauges. +func RecordJobAuditDepth(counts map[string]int64) { + for status, n := range counts { + if status == "" { + status = "unknown" + } + jobAuditDepth.WithLabelValues(status).Set(float64(n)) + } +} + +// SetUpstreamBreakerOpen records circuit breaker state for a host (1=open, 0=closed). +func SetUpstreamBreakerOpen(host string, open bool) { + if host == "" { + host = "_" + } + v := 0.0 + if open { + v = 1 + } + upstreamBreakerOpen.WithLabelValues(host).Set(v) +} + // SetBuildInfo sets evobgp_build_info gauge (idempotent labels). func SetBuildInfo(version, gitSHA string) { if version == "" { @@ -292,15 +327,32 @@ func MetricsHandler() http.Handler { return promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{}) } -// HTTPMiddleware records method and status code for all wrapped requests. +// HTTPMiddleware records method, status code, and path group for all wrapped requests. func HTTPMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { sw := &statusRecorder{ResponseWriter: w, status: http.StatusOK} next.ServeHTTP(sw, r) - httpRequests.WithLabelValues(r.Method, strconv.Itoa(sw.status)).Inc() + httpRequests.WithLabelValues(r.Method, strconv.Itoa(sw.status), pathGroup(r.URL.Path)).Inc() }) } +func pathGroup(path string) string { + if path == "" { + return "/" + } + parts := strings.Split(strings.Trim(path, "/"), "/") + if len(parts) == 0 || parts[0] == "" { + return "/" + } + if parts[0] == "v1" && len(parts) >= 2 { + return "/v1/" + parts[1] + } + if parts[0] == "metrics" || parts[0] == "version" { + return "/" + parts[0] + } + return "/" + parts[0] +} + type statusRecorder struct { http.ResponseWriter status int diff --git a/internal/pgmonitor/scheduler.go b/internal/pgmonitor/scheduler.go index a3293ed..5a7ec29 100644 --- a/internal/pgmonitor/scheduler.go +++ b/internal/pgmonitor/scheduler.go @@ -2,7 +2,8 @@ package pgmonitor import ( "context" - "log" + "evobgp/internal/logging" + "fmt" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -23,23 +24,23 @@ func StartScheduler(ctx context.Context, pool *pgxpool.Pool) { c, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() if err := s.RefreshMetricsSnapshot(c); err != nil { - log.Printf("pgmonitor: metrics refresh: %v", err) + logging.Default().Info(fmt.Sprintf("pgmonitor: metrics refresh: %v", err)) } if err := s.DetectAutovacuumLag(c); err != nil { - log.Printf("pgmonitor: autovacuum lag: %v", err) + logging.Default().Info(fmt.Sprintf("pgmonitor: autovacuum lag: %v", err)) } } runHeavy := func() { c, cancel := context.WithTimeout(context.Background(), 3*time.Minute) defer cancel() if err := s.AggregateSlowQueries(c, 30); err != nil { - log.Printf("pgmonitor: slow queries snapshot: %v", err) + logging.Default().Info(fmt.Sprintf("pgmonitor: slow queries snapshot: %v", err)) } if err := s.EstimateTableBloat(c); err != nil { - log.Printf("pgmonitor: bloat: %v", err) + logging.Default().Info(fmt.Sprintf("pgmonitor: bloat: %v", err)) } if err := s.AnalyzeIndexUsage(c); err != nil { - log.Printf("pgmonitor: index usage: %v", err) + logging.Default().Info(fmt.Sprintf("pgmonitor: index usage: %v", err)) } } runLight() @@ -55,5 +56,5 @@ func StartScheduler(ctx context.Context, pool *pgxpool.Pool) { } } }() - log.Printf("pgmonitor: scheduler started (5m light / 15m heavy)") + logging.Default().Info(fmt.Sprintf("pgmonitor: scheduler started (5m light / 15m heavy)")) } diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go index 57258dc..0364ec7 100644 --- a/internal/pipeline/collect_parallel.go +++ b/internal/pipeline/collect_parallel.go @@ -2,8 +2,8 @@ package pipeline import ( "context" + "evobgp/internal/logging" "fmt" - "log" "net/http" "os" "strings" @@ -223,7 +223,7 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client for _, r := range results { if r.err != nil { if cdnPartialOK() { - log.Printf("pipeline: CDN partial skip source error: %v", r.err) + logging.Default().Info(fmt.Sprintf("pipeline: CDN partial skip source error: %v", r.err)) skipped++ continue } diff --git a/internal/pipeline/collect_stale.go b/internal/pipeline/collect_stale.go index e57116b..e2ead58 100644 --- a/internal/pipeline/collect_stale.go +++ b/internal/pipeline/collect_stale.go @@ -1,8 +1,8 @@ package pipeline import ( + "evobgp/internal/logging" "fmt" - "log" "net/netip" "os" "strings" @@ -28,7 +28,7 @@ func cdnPartialOK() bool { } func logStaleUpstream(kind, detail string) { - log.Printf("pipeline: stale upstream fallback (%s): %s", kind, detail) + logging.Default().Info(fmt.Sprintf("pipeline: stale upstream fallback (%s): %s", kind, detail)) } func staleASNPrefixes(st store.Backend, priorSnapshot []store.PrefixRow, asn int64) ([]store.PrefixRow, string, bool) { diff --git a/internal/render/publish.go b/internal/render/publish.go index 2c11f6f..ff9a173 100644 --- a/internal/render/publish.go +++ b/internal/render/publish.go @@ -2,7 +2,8 @@ package render import ( "context" - "log" + "evobgp/internal/logging" + "fmt" "evobgp/internal/store" ) @@ -13,7 +14,7 @@ func PublishLatestTenantRevision(ctx context.Context, st store.Backend) { _ = ctx tenants, err := st.ListTenantIDs() if err != nil { - log.Printf("evobgp-render: list tenants: %v", err) + logging.Default().Info(fmt.Sprintf("evobgp-render: list tenants: %v", err)) return } for _, tid := range tenants { @@ -24,7 +25,7 @@ func PublishLatestTenantRevision(ctx context.Context, st store.Backend) { rid := revs[0].ID for _, sp := range st.ListSpeakersForTenant(tid) { if err := st.PublishRevisionForSpeaker(sp.ID, rid); err != nil { - log.Printf("evobgp-render: publish speaker %s: %v", sp.ID, err) + logging.Default().Info(fmt.Sprintf("evobgp-render: publish speaker %s: %v", sp.ID, err)) } } } diff --git a/internal/render/run.go b/internal/render/run.go index 693d24a..6b92878 100644 --- a/internal/render/run.go +++ b/internal/render/run.go @@ -2,6 +2,8 @@ package render import ( "context" + "evobgp/internal/logging" + "fmt" "log" "os" "strings" @@ -28,14 +30,14 @@ func Run(ctx context.Context, deps *Deps) { defer t.Stop() auto := strings.TrimSpace(os.Getenv("EVOBGP_RENDER_AUTOPUBLISH")) == "1" if auto { - log.Printf("evobgp-render: active (EVOBGP_RENDER_AUTOPUBLISH=1: publish head revision per tenant)") + logging.Default().Info(fmt.Sprintf("evobgp-render: active (EVOBGP_RENDER_AUTOPUBLISH=1: publish head revision per tenant)")) } else { - log.Printf("evobgp-render: active (idle publish; set EVOBGP_RENDER_AUTOPUBLISH=1 to auto-publish head revision)") + logging.Default().Info(fmt.Sprintf("evobgp-render: active (idle publish; set EVOBGP_RENDER_AUTOPUBLISH=1 to auto-publish head revision)")) } for { select { case <-ctx.Done(): - log.Printf("evobgp-render: stopped") + logging.Default().Info(fmt.Sprintf("evobgp-render: stopped")) return case <-t.C: if auto { diff --git a/internal/repository/job_audit_claim.go b/internal/repository/job_audit_claim.go new file mode 100644 index 0000000..9aeceb2 --- /dev/null +++ b/internal/repository/job_audit_claim.go @@ -0,0 +1,130 @@ +package repository + +import ( + "context" + "encoding/json" + "time" + + "github.com/jackc/pgx/v5" +) + +// ClaimedJob is a durable job_audit row claimed for in-process execution (SKIP LOCKED). +type ClaimedJob struct { + ID string + TenantID string + Kind string + IdempotencyKey *string + ModuleID *string + Meta map[string]any + CreatedAt time.Time +} + +// ReclaimStaleRunning resets orphaned running jobs older than staleAfter back to queued. +func (w *JobAuditWriter) ReclaimStaleRunning(ctx context.Context, staleAfter time.Duration) (int64, error) { + if w == nil || w.pool == nil { + return 0, nil + } + if staleAfter <= 0 { + staleAfter = 15 * time.Minute + } + tag, err := w.pool.Exec(ctx, ` + UPDATE job_audit + SET status = 'queued', started_at = NULL, error_message = NULL + WHERE status = 'running' + AND started_at IS NOT NULL + AND started_at < now() - $1::interval`, + staleAfter.String()) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + +// ClaimQueued claims up to limit queued job_audit rows via FOR UPDATE SKIP LOCKED and marks them running. +// Only rows older than grace are claimed so the originating process can own fresh enqueues. +func (w *JobAuditWriter) ClaimQueued(ctx context.Context, limit int, grace time.Duration) ([]ClaimedJob, error) { + if w == nil || w.pool == nil { + return nil, nil + } + if limit <= 0 { + limit = 8 + } + if grace <= 0 { + grace = 30 * time.Second + } + tx, err := w.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return nil, err + } + defer tx.Rollback(ctx) + + rows, err := tx.Query(ctx, ` + WITH cte AS ( + SELECT id + FROM job_audit + WHERE status = 'queued' + AND created_at < now() - $2::interval + ORDER BY created_at ASC + FOR UPDATE SKIP LOCKED + LIMIT $1 + ) + UPDATE job_audit j + SET status = 'running', started_at = COALESCE(j.started_at, now()) + FROM cte + WHERE j.id = cte.id + RETURNING j.id::text, j.tenant_id::text, j.kind, j.idempotency_key, j.module_id::text, j.meta_json, j.created_at`, + limit, grace.String()) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []ClaimedJob + for rows.Next() { + var c ClaimedJob + var idem, mod *string + var metaBytes []byte + if err := rows.Scan(&c.ID, &c.TenantID, &c.Kind, &idem, &mod, &metaBytes, &c.CreatedAt); err != nil { + return nil, err + } + c.IdempotencyKey = idem + if mod != nil && *mod != "" { + c.ModuleID = mod + } + meta := map[string]any{} + if len(metaBytes) > 0 { + _ = json.Unmarshal(metaBytes, &meta) + } + c.Meta = meta + out = append(out, c) + } + if err := rows.Err(); err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, err + } + return out, nil +} + +// CountByStatus returns job_audit row counts grouped by status (best-effort metrics). +func (w *JobAuditWriter) CountByStatus(ctx context.Context) (map[string]int64, error) { + out := map[string]int64{} + if w == nil || w.pool == nil { + return out, nil + } + rows, err := w.pool.Query(ctx, `SELECT status, count(*) FROM job_audit GROUP BY status`) + if err != nil { + return out, err + } + defer rows.Close() + for rows.Next() { + var status string + var n int64 + if err := rows.Scan(&status, &n); err != nil { + return out, err + } + out[status] = n + } + return out, rows.Err() +} diff --git a/internal/scheduler/run.go b/internal/scheduler/run.go index 238c84f..81e1cb6 100644 --- a/internal/scheduler/run.go +++ b/internal/scheduler/run.go @@ -5,6 +5,7 @@ import ( "bytes" "context" "encoding/json" + "evobgp/internal/logging" "fmt" "io" "log" @@ -43,14 +44,14 @@ func Run(ctx context.Context, deps *Deps) { t := time.NewTicker(30 * time.Second) defer t.Stop() if deps.Jobs != nil { - log.Printf("evobgp-scheduler: active (in-process enqueue tenant_refresh)") + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: active (in-process enqueue tenant_refresh)")) } else { - log.Printf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase)) + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: active (HTTP POST .../tenant/refresh → %s)", strings.TrimSpace(deps.APIBase))) } for { select { case <-ctx.Done(): - log.Printf("evobgp-scheduler: stopped") + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: stopped")) return case <-t.C: tick(context.Background(), deps) @@ -61,7 +62,7 @@ func Run(ctx context.Context, deps *Deps) { func tick(ctx context.Context, deps *Deps) { tenants, err := deps.Store.ListTenantIDs() if err != nil { - log.Printf("evobgp-scheduler: list tenants: %v", err) + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: list tenants: %v", err)) return } now := time.Now().UTC() @@ -84,16 +85,16 @@ func tick(ctx context.Context, deps *Deps) { "trigger": "scheduler", }) if err != nil { - log.Printf("evobgp-scheduler: enqueue tenant %s: %v", tid, err) + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: enqueue tenant %s: %v", tid, err)) continue } if created { - log.Printf("evobgp-scheduler: queued tenant refresh for %d module(s) in tenant %s", len(due), tid) + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: queued tenant refresh for %d module(s) in tenant %s", len(due), tid)) } continue } if err := postTenantRefresh(ctx, deps, due, key); err != nil { - log.Printf("evobgp-scheduler: http tenant refresh %s: %v", tid, err) + logging.Default().Info(fmt.Sprintf("evobgp-scheduler: http tenant refresh %s: %v", tid, err)) } } }