diff --git a/cmd/evobgp-all/main.go b/cmd/evobgp-all/main.go index 8576144..5f7c969 100644 --- a/cmd/evobgp-all/main.go +++ b/cmd/evobgp-all/main.go @@ -29,6 +29,9 @@ func main() { os.Exit(dbcli.Run(os.Args[2:])) } cfg := config.Load() + if err := config.ValidateProductionEnforce(); err != nil { + log.Fatal(err) + } opts := httpapi.Options{ APIKeys: os.Getenv("EVOBGP_API_KEYS"), DatabaseURL: cfg.DatabaseURL, @@ -86,6 +89,13 @@ func main() { <-ctx.Done() shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() + n := srv.Jobs().RequestCancelAll() + if n > 0 { + log.Printf("draining %d job(s)…", n) + if err := srv.Jobs().Drain(shutdownCtx); err != nil { + log.Printf("job drain: %v", err) + } + } if err := httpSrv.Shutdown(shutdownCtx); err != nil { log.Printf("HTTP shutdown: %v", err) } diff --git a/cmd/evobgp-api/main.go b/cmd/evobgp-api/main.go index 6258950..37b1811 100644 --- a/cmd/evobgp-api/main.go +++ b/cmd/evobgp-api/main.go @@ -24,6 +24,9 @@ func main() { os.Exit(dbcli.Run(os.Args[2:])) } cfg := config.Load() + if err := config.ValidateProductionEnforce(); err != nil { + log.Fatal(err) + } seedDemo := os.Getenv("EVOBGP_SEED_DEMO") != "0" opts := httpapi.Options{ APIKeys: os.Getenv("EVOBGP_API_KEYS"), @@ -78,6 +81,13 @@ func main() { <-ctx.Done() shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() + n := srv.Jobs().RequestCancelAll() + if n > 0 { + log.Printf("draining %d job(s)…", n) + if err := srv.Jobs().Drain(shutdownCtx); err != nil { + log.Printf("job drain: %v", err) + } + } if err := httpSrv.Shutdown(shutdownCtx); err != nil { log.Printf("HTTP shutdown: %v", err) } diff --git a/docs/production-checklist.md b/docs/production-checklist.md index fefa3f6..23edd89 100644 --- a/docs/production-checklist.md +++ b/docs/production-checklist.md @@ -4,12 +4,14 @@ ## Обязательно +- `EVOBGP_PRODUCTION=1` (или `EVOBGP_ENV=production`) — при старте API/`evobgp-all` процесс **откажется** стартовать, если нарушены жёсткие требования ниже (`config.ValidateProductionEnforce`). - `EVOBGP_SEED_DEMO=0` — отключить demo-tenant и токен `Bearer dev`. - `EVOBGP_DEV_INSECURE` не задавать или `0` — не использовать lab-флаги в prod. - `EVOBGP_BUNDLE_SEED_HEX` — задать стабильный hex-ключ подписи бандлов; сохранить pubkey для нод. - PostgreSQL с TLS (`sslmode` не `disable`) при доступе вне private network. - `EVOBGP_CORS_ORIGINS` — явный whitelist origin веб-панели. - `EVOBGP_STALE_ON_UPSTREAM_ERROR=1` (по умолчанию) — stale snapshot при сбоях CDN/ASN/DoH. +- Опционально `EVOBGP_CDN_PARTIAL_OK=1` — при сбое одного CDN source без stale cache продолжать refresh остальных (иначе fail модуля). ## Рекомендуется diff --git a/internal/config/production.go b/internal/config/production.go new file mode 100644 index 0000000..c1fcfc6 --- /dev/null +++ b/internal/config/production.go @@ -0,0 +1,38 @@ +package config + +import ( + "fmt" + "os" + "strings" +) + +// ProductionMode reports whether EVOBGP_PRODUCTION / EVOBGP_ENV=production is set. +func ProductionMode() bool { + if v := strings.TrimSpace(os.Getenv("EVOBGP_PRODUCTION")); v == "1" || strings.EqualFold(v, "true") { + return true + } + env := strings.ToLower(strings.TrimSpace(os.Getenv("EVOBGP_ENV"))) + return env == "production" || env == "prod" +} + +// ValidateProductionEnforce enforces docs/production-checklist.md hard requirements +// when ProductionMode() is true. Returns an error that should abort process start. +func ValidateProductionEnforce() error { + if !ProductionMode() { + return nil + } + seedDemo := os.Getenv("EVOBGP_SEED_DEMO") + if seedDemo != "0" { + return fmt.Errorf("config: production requires EVOBGP_SEED_DEMO=0 (got %q)", seedDemo) + } + if strings.TrimSpace(os.Getenv("EVOBGP_DEV_INSECURE")) == "1" { + return fmt.Errorf("config: production forbids EVOBGP_DEV_INSECURE=1") + } + if strings.TrimSpace(os.Getenv("EVOBGP_BUNDLE_SEED_HEX")) == "" { + return fmt.Errorf("config: production requires EVOBGP_BUNDLE_SEED_HEX") + } + if strings.TrimSpace(os.Getenv("EVOBGP_CDN_ALLOW_PRIVATE")) == "1" { + return fmt.Errorf("config: production forbids EVOBGP_CDN_ALLOW_PRIVATE=1") + } + return nil +} diff --git a/internal/config/production_test.go b/internal/config/production_test.go new file mode 100644 index 0000000..6a0cf15 --- /dev/null +++ b/internal/config/production_test.go @@ -0,0 +1,31 @@ +package config + +import "testing" + +func TestValidateProductionEnforce(t *testing.T) { + t.Setenv("EVOBGP_PRODUCTION", "") + t.Setenv("EVOBGP_ENV", "") + if err := ValidateProductionEnforce(); err != nil { + t.Fatalf("non-production: %v", err) + } + + t.Setenv("EVOBGP_PRODUCTION", "1") + t.Setenv("EVOBGP_SEED_DEMO", "1") + t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "abcd") + t.Setenv("EVOBGP_DEV_INSECURE", "") + t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "") + if err := ValidateProductionEnforce(); err == nil { + t.Fatal("expected error for SEED_DEMO!=0") + } + + t.Setenv("EVOBGP_SEED_DEMO", "0") + t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "") + if err := ValidateProductionEnforce(); err == nil { + t.Fatal("expected error for missing BUNDLE_SEED_HEX") + } + + t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "bd8fbcd31545aacfdd228203beca8e945ab9a752f2ce5624cf42d9f316389a9d") + if err := ValidateProductionEnforce(); err != nil { + t.Fatalf("valid production: %v", err) + } +} diff --git a/internal/httpapi/routes_crud.go b/internal/httpapi/routes_crud.go index 1b96c73..c5b3acd 100644 --- a/internal/httpapi/routes_crud.go +++ b/internal/httpapi/routes_crud.go @@ -314,7 +314,7 @@ func (s *Server) handlePreviewCDNSource(w http.ResponseWriter, r *http.Request) writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "invalid url") return } - resp, err := s.cdnHTTP.Do(req) + resp, err := pipeline.UpstreamHTTPDo(r.Context(), s.cdnHTTP, req) if err != nil { writeBadGateway(w, "cdn preview fetch", err) return diff --git a/internal/jobs/job.go b/internal/jobs/job.go index 2f91b13..5e69d25 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -1,6 +1,7 @@ package jobs import ( + "context" "fmt" "os" "sort" @@ -111,6 +112,42 @@ func (j *Job) mergeMeta(kv map[string]any) { } } +// metaString returns a string meta field under lock (safe vs concurrent mergeMeta). +func (j *Job) metaString(key string) string { + j.mu.Lock() + defer j.mu.Unlock() + if j.Meta == nil { + return "" + } + s, _ := j.Meta[key].(string) + return s +} + +// metaBool returns a bool meta field under lock. +func (j *Job) metaBool(key string) bool { + j.mu.Lock() + defer j.mu.Unlock() + if j.Meta == nil { + return false + } + b, _ := j.Meta[key].(bool) + return b +} + +// metaCopy returns a shallow copy of job meta under lock. +func (j *Job) metaCopy() map[string]any { + j.mu.Lock() + defer j.mu.Unlock() + if j.Meta == nil { + return nil + } + out := make(map[string]any, len(j.Meta)) + for k, v := range j.Meta { + out[k] = v + } + return out +} + // statusLocked is used by the worker defer for metrics (any stable terminal or in-flight status). func (j *Job) statusLocked() string { j.mu.Lock() @@ -528,3 +565,65 @@ func (r *Registry) RequestCancel(tenantID, jobID string) (*Job, error) { j.RequestCancel() return j, nil } + +// RequestCancelAll requests cancellation of all non-terminal jobs (all tenants). +func (r *Registry) RequestCancelAll() int { + if r == nil { + return 0 + } + r.mu.RLock() + defer r.mu.RUnlock() + n := 0 + for _, j := range r.byID { + if j == nil { + continue + } + st := j.statusLocked() + if st == StatusSucceeded || st == StatusFailed || st == StatusCancelled { + continue + } + j.RequestCancel() + n++ + } + return n +} + +// ActiveCount returns the number of queued or running jobs. +func (r *Registry) ActiveCount() int { + if r == nil { + return 0 + } + r.mu.RLock() + defer r.mu.RUnlock() + n := 0 + for _, j := range r.byID { + if j == nil { + continue + } + st := j.statusLocked() + if st == StatusQueued || st == StatusRunning { + n++ + } + } + return n +} + +// Drain waits until no queued/running jobs remain or ctx is done. +// Call RequestCancelAll first for a cooperative shutdown. +func (r *Registry) Drain(ctx context.Context) error { + if r == nil { + return nil + } + ticker := time.NewTicker(50 * time.Millisecond) + defer ticker.Stop() + for { + if r.ActiveCount() == 0 { + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + } +} diff --git a/internal/jobs/maintenance_worker.go b/internal/jobs/maintenance_worker.go index ae8aa1a..44e395d 100644 --- a/internal/jobs/maintenance_worker.go +++ b/internal/jobs/maintenance_worker.go @@ -21,14 +21,13 @@ func (w *Worker) runMaintenancePolicy(j *Job) { j.Fail("postgresql not configured") return } - policyID, _ := j.Meta["policy_id"].(string) - policyID = strings.TrimSpace(policyID) + policyID := strings.TrimSpace(j.metaString("policy_id")) if policyID == "" { j.Fail("missing policy_id in job meta") return } - dryRun, _ := j.Meta["dry_run"].(bool) - actor, _ := j.Meta["actor_prefix"].(string) + dryRun := j.metaBool("dry_run") + actor := j.metaString("actor_prefix") ctx, cancel := j.workContext() defer cancel() diff --git a/internal/jobs/postgres_worker.go b/internal/jobs/postgres_worker.go index 40e079b..573dbfd 100644 --- a/internal/jobs/postgres_worker.go +++ b/internal/jobs/postgres_worker.go @@ -93,9 +93,9 @@ func (w *Worker) runPostgresMaint(j *Job, kind string) { j.Fail("postgresql not configured") return } - table, _ := j.Meta["table"].(string) - dryRun, _ := j.Meta["dry_run"].(bool) - actor, _ := j.Meta["actor_prefix"].(string) + table := j.metaString("table") + dryRun := j.metaBool("dry_run") + actor := j.metaString("actor_prefix") ctx, cancel := j.workContext() defer cancel() auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, kind, table, dryRun) diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index ccd20ac..11aadda 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -291,8 +291,8 @@ func (w *Worker) runModuleRefresh(j *Job) { } }() - mid, _ := j.Meta["module_id"].(string) - if strings.TrimSpace(mid) == "" { + mid := strings.TrimSpace(j.metaString("module_id")) + if mid == "" { j.Fail("missing module_id in job meta") return } @@ -325,7 +325,7 @@ func (w *Worker) runTenantRefresh(j *Job) { } }() - moduleIDs := moduleIDsFromJobMeta(j.Meta) + moduleIDs := moduleIDsFromJobMeta(j.metaCopy()) if len(moduleIDs) == 0 { j.Fail("missing module_ids in job meta") return @@ -449,8 +449,9 @@ func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) { } func (w *Worker) runDeployApply(j *Job) { - revID, _ := j.Meta["revision_id"].(string) - spk, hasSpeaker := j.Meta["speaker_id"].(string) + 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 @@ -571,7 +572,7 @@ func (w *Worker) runDeployApply(j *Job) { } func (w *Worker) runRollback(j *Job) { - src, _ := j.Meta["source_revision_id"].(string) + src := strings.TrimSpace(j.metaString("source_revision_id")) if src == "" { j.Fail("missing source_revision_id in job meta") return diff --git a/internal/pipeline/collect_parallel.go b/internal/pipeline/collect_parallel.go index e89d759..57258dc 100644 --- a/internal/pipeline/collect_parallel.go +++ b/internal/pipeline/collect_parallel.go @@ -3,6 +3,7 @@ package pipeline import ( "context" "fmt" + "log" "net/http" "os" "strings" @@ -218,12 +219,21 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client wg.Wait() var out []store.PrefixRow + var skipped int for _, r := range results { if r.err != nil { + if cdnPartialOK() { + log.Printf("pipeline: CDN partial skip source error: %v", r.err) + skipped++ + continue + } return nil, r.err } out = append(out, r.rows...) } + if skipped > 0 && len(out) == 0 && len(valid) > 0 { + return nil, fmt.Errorf("cdn: all %d source(s) failed (partial ok)", len(valid)) + } if len(valid) > 0 { if err := mergeAllCDNSourcesIntoModuleSnapshot(st, tenantID, mod, priorSnapshot, out); err != nil { return nil, err diff --git a/internal/pipeline/collect_stale.go b/internal/pipeline/collect_stale.go index 2871be6..e57116b 100644 --- a/internal/pipeline/collect_stale.go +++ b/internal/pipeline/collect_stale.go @@ -20,6 +20,13 @@ func staleOnUpstreamError() bool { return false } +// cdnPartialOK reports whether a failed CDN source without stale cache should be skipped +// instead of failing the whole module refresh. Opt-in: EVOBGP_CDN_PARTIAL_OK=1. +func cdnPartialOK() bool { + v := strings.TrimSpace(os.Getenv("EVOBGP_CDN_PARTIAL_OK")) + return v == "1" || strings.EqualFold(v, "true") +} + func logStaleUpstream(kind, detail string) { log.Printf("pipeline: stale upstream fallback (%s): %s", kind, detail) } diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 69957a4..7586dc4 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -282,7 +282,7 @@ func resolveDomainWithDOHMessage(ctx context.Context, hc *http.Client, baseURL, return nil, err } req.Header.Set("Accept", "application/dns-message") - resp, err := hc.Do(req) + resp, err := httpclient.DoWithRetry(ctx, hc, req, 3) if err != nil { return nil, err } @@ -354,7 +354,7 @@ func resolveDomainWithDOHJSON(ctx context.Context, hc *http.Client, baseURL, hos } req.Header.Set("Accept", "application/dns-json") - resp, err := hc.Do(req) + resp, err := httpclient.DoWithRetry(ctx, hc, req, 3) if err != nil { return nil, err } diff --git a/internal/pipeline/upstream_http.go b/internal/pipeline/upstream_http.go index 2345487..994d3f1 100644 --- a/internal/pipeline/upstream_http.go +++ b/internal/pipeline/upstream_http.go @@ -8,6 +8,11 @@ import ( "evobgp/internal/httpclient" ) +// UpstreamHTTPDo performs an outbound GET/POST with circuit breaker + retries (CDN, previews). +func UpstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) { + return upstreamHTTPDo(ctx, hc, req) +} + func upstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) { if hc == nil { hc = httpclient.New(httpclient.DefaultTimeout)