denozord/EvoBGP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 131 Wiki Activity

feat(jobs): add durable PG queue reclaim, slog, and richer metrics

Browse Source 
JSON slog в ключевых пакетах; Prometheus path_group, job_audit_depth, upstream breaker; job_audit ClaimQueued/ReclaimStaleRunning + Adopt loop для HA после рестарта.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Denozordec
2026-07-31 12:26:18 +07:00
co-authored by Cursor
parent 13e3d21ce2
commit 8fe74c1d3b
18 changed files with 448 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/evobgp-api/main.go
+10 -9
View File
@@ -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 {
internal/broker/connect.go
+3 -2
View File
@@ -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
}
internal/deploy/run.go
+7 -5
View File
@@ -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))
}
}
}
internal/httpapi/server.go
+4
View File
@@ -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) {
internal/httpclient/circuit.go
+18
View File
@@ -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{}
internal/httpclient/httpclient.go
+9 -2
View File
@@ -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
}
internal/ingest/run.go
+5 -3
View File
@@ -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))
}
}
}
internal/jobs/durable.go
+137
View File
@@ -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()
}
}
}()
}
internal/logging/logger.go
+38
View File
@@ -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...)
}
internal/maintenance/scheduler.go
+3 -3
View File
@@ -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))
}
internal/observability/metrics.go
+55 -3
View File
@@ -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
internal/pgmonitor/scheduler.go
+8 -7
View File
@@ -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)"))
}
internal/pipeline/collect_parallel.go
+2 -2
View File
@@ -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
}
internal/pipeline/collect_stale.go
+2 -2
View File
@@ -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) {
internal/render/publish.go
+4 -3
View File
@@ -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))
}
}
}
internal/render/run.go
+5 -3
View File
@@ -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 {
internal/repository/job_audit_claim.go
+130
View File
@@ -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()
}
internal/scheduler/run.go
+8 -7
View File
@@ -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))
}
}
}