diff --git a/internal/dbcli/dbcli.go b/internal/dbcli/dbcli.go index bac8b23..f7278e7 100644 --- a/internal/dbcli/dbcli.go +++ b/internal/dbcli/dbcli.go @@ -115,22 +115,25 @@ func cmdMaint(args []string, _ string, path string) int { func cmdCleanup(args []string) int { fs := flag.NewFlagSet("cleanup", flag.ExitOnError) - policy := fs.String("policy", "", "cleanup policy name") + policyID := fs.String("policy-id", "", "maintenance policy UUID") dryRun := fs.Bool("dry-run", true, "dry run") - limit := fs.Int("limit", 10000, "max rows") apiURL := fs.String("api-url", "", "control plane base URL") token := fs.String("token", "", "Bearer token (operator)") _ = fs.Parse(args) - if *policy == "" { - fmt.Fprintln(os.Stderr, "cleanup: --policy is required") + if *policyID == "" { + fmt.Fprintln(os.Stderr, "cleanup: --policy-id is required") return 2 } if *apiURL == "" || *token == "" { fmt.Fprintln(os.Stderr, "cleanup: --api-url and --token are required") return 2 } - payload := map[string]any{"policy": *policy, "dry_run": *dryRun, "limit": *limit} - body, err := apiPOST(*apiURL, *token, "/v1/postgres/cleanup", payload) + path := "/v1/maintenance/run" + if *dryRun { + path = "/v1/maintenance/dry-run" + } + payload := map[string]any{"policy_id": *policyID} + body, err := apiPOST(*apiURL, *token, path, payload) if err != nil { fmt.Fprintln(os.Stderr, err) return 1 diff --git a/internal/httpapi/routes_postgres_maintenance.go b/internal/httpapi/routes_postgres_maintenance.go index 797778a..3415a37 100644 --- a/internal/httpapi/routes_postgres_maintenance.go +++ b/internal/httpapi/routes_postgres_maintenance.go @@ -47,11 +47,12 @@ func (s *Server) checkPgMaintRateLimit(tenantID, kind string) bool { } type pgMaintBody struct { - Table string `json:"table"` - DryRun bool `json:"dry_run"` - Index string `json:"index"` - Policy string `json:"policy"` - Limit int `json:"limit"` + Table string `json:"table"` + DryRun bool `json:"dry_run"` + Index string `json:"index"` + Policy string `json:"policy"` + PolicyID string `json:"policy_id"` + Limit int `json:"limit"` } func (s *Server) decodePgMaintBody(r *http.Request) (pgMaintBody, bool) { @@ -168,14 +169,34 @@ func (s *Server) handlePostgresCleanup(w http.ResponseWriter, r *http.Request) { writeProblem(w, http.StatusBadRequest, "Bad Request", invalidInputDetail) return } - if strings.TrimSpace(body.Policy) == "" { - writeProblem(w, http.StatusBadRequest, "Bad Request", "policy is required") + policyID := strings.TrimSpace(body.PolicyID) + if policyID == "" { + policyID = strings.TrimSpace(body.Policy) + } + if policyID == "" { + writeProblem(w, http.StatusBadRequest, "Bad Request", "policy_id is required") return } - s.enqueuePostgresMaint(w, r, a, jobs.KindPostgresCleanup, map[string]any{ - "policy": body.Policy, "dry_run": body.DryRun, "limit": body.Limit, - "job_title": "PostgreSQL cleanup", + if _, err := s.store.GetMaintenancePolicy(policyID); err != nil { + writeStoreErr(w, err) + return + } + idem := strings.TrimSpace(r.Header.Get("Idempotency-Key")) + var idemPtr *string + if idem != "" { + idemPtr = &idem + } + j, _, err := s.jobs.Enqueue(a.TenantID, jobs.KindMaintenancePolicyRun, idemPtr, nil, map[string]any{ + "policy_id": policyID, "dry_run": body.DryRun, "actor_prefix": actorPrefix(a), + "job_title": "PostgreSQL cleanup (deprecated path)", }) + if err != nil { + writeInternalError(w, "postgres_maint_enqueue", err) + return + } + w.Header().Set("Location", "/v1/jobs/"+j.ID) + snap := j.Snapshot() + writeJSON(w, http.StatusAccepted, map[string]any{"job_id": snap["job_id"], "status": snap["status"]}) } func (s *Server) handlePostgresMaintenanceLogs(w http.ResponseWriter, r *http.Request) { diff --git a/internal/httpapi/server.go b/internal/httpapi/server.go index e30c3b7..ac03853 100644 --- a/internal/httpapi/server.go +++ b/internal/httpapi/server.go @@ -8,6 +8,7 @@ import ( "errors" "net/http" "strings" + "time" "evobgp/internal/jobs" "evobgp/internal/maintenance" @@ -107,9 +108,19 @@ func (s *Server) Store() store.Backend { return s.store } // Jobs exposes the in-process async job registry (for scheduler / evobgp-all). func (s *Server) Jobs() *jobs.Registry { return s.jobs } -// StartBackground starts PostgreSQL monitoring scheduler until ctx is cancelled. +// StartBackground starts PostgreSQL monitoring and maintenance schedulers until ctx is cancelled. func (s *Server) StartBackground(ctx context.Context) { if s != nil && s.pgPool != nil { pgmonitor.StartScheduler(ctx, s.pgPool) } + if s != nil && s.maintConfig != nil && s.jobs != nil { + maintenance.StartScheduler(ctx, s.maintConfig, func(policyID string, dryRun bool, idem string) { + key := idem + _, _, _ = s.jobs.Enqueue("", jobs.KindMaintenancePolicyRun, &key, nil, map[string]any{ + "policy_id": policyID, + "dry_run": dryRun, + "trigger": "scheduler", + }) + }, 30*time.Second) + } } diff --git a/internal/ingest/run.go b/internal/ingest/run.go index 9046898..58a1579 100644 --- a/internal/ingest/run.go +++ b/internal/ingest/run.go @@ -17,8 +17,6 @@ type Deps struct { Store store.Backend } -var lastMaintenance time.Time - // Run blocks until ctx is cancelled. func Run(ctx context.Context, deps *Deps) { cfg := config.Load() @@ -36,10 +34,6 @@ func Run(ctx context.Context, deps *Deps) { log.Printf("evobgp-ingest: stopped") return case <-t.C: - if deps.Store != nil && time.Since(lastMaintenance) > time.Hour { - deps.Store.RunPeriodicMaintenance(ctx) - lastMaintenance = time.Now() - } prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second) err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc) cancel() diff --git a/internal/jobs/postgres_worker.go b/internal/jobs/postgres_worker.go index 0183742..40e079b 100644 --- a/internal/jobs/postgres_worker.go +++ b/internal/jobs/postgres_worker.go @@ -2,7 +2,6 @@ package jobs import ( "fmt" - "strings" "evobgp/internal/pgmonitor" ) @@ -118,35 +117,7 @@ func (w *Worker) runPostgresMaint(j *Job, kind string) { } func (w *Worker) runPostgresCleanup(j *Job) { - if w == nil || w.PgPool == nil { - j.Fail("postgresql not configured") - return - } - policy, _ := j.Meta["policy"].(string) - dryRun, _ := j.Meta["dry_run"].(bool) - limit := 0 - if v, ok := j.Meta["limit"].(float64); ok { - limit = int(v) - } - actor, _ := j.Meta["actor_prefix"].(string) - ctx, cancel := j.workContext() - defer cancel() - auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, "cleanup", policy, dryRun) - detail, err := pgmonitor.RunCleanup(ctx, w.PgPool, strings.TrimSpace(policy), dryRun, limit) - var errMsg *string - status := StatusSucceeded - if err != nil { - s := err.Error() - errMsg = &s - status = StatusFailed - j.Fail(s) - } else { - j.mergeMeta(map[string]any{"cleanup": detail, "audit_id": auditID}) - j.Succeed() - } - if auditID != "" { - _ = pgmonitor.FinishMaintenanceAudit(ctx, w.PgPool, auditID, status, detail, errMsg) - } + j.Fail("postgres_cleanup deprecated: configure maintenance_policy in UI and use maintenance_policy_run") } // EnqueuePostgresAnalyzerJobs enqueues periodic analyzer jobs (global tenant id). diff --git a/internal/pgmonitor/maintenance_audit.go b/internal/pgmonitor/maintenance_audit.go index 0f32175..3b0164d 100644 --- a/internal/pgmonitor/maintenance_audit.go +++ b/internal/pgmonitor/maintenance_audit.go @@ -3,81 +3,18 @@ package pgmonitor import ( "context" "encoding/json" - "fmt" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgxpool" ) -// CleanupPolicy names safe retention policies. -type CleanupPolicy string - -const ( - PolicyJobAuditRetention CleanupPolicy = "job_audit_retention" - PolicyASNCacheRetention CleanupPolicy = "asn_cache_retention" -) - -// CleanupRequest for POST /postgres/cleanup. +// CleanupRequest for deprecated POST /postgres/cleanup (use /v1/maintenance/run). type CleanupRequest struct { - Policy string `json:"policy"` - DryRun bool `json:"dry_run"` - Limit int `json:"limit"` -} - -// RunCleanup executes a named retention policy. -func RunCleanup(ctx context.Context, pool *pgxpool.Pool, policy string, dryRun bool, limit int) (map[string]any, error) { - if pool == nil { - return nil, fmt.Errorf("pgmonitor: postgres not configured") - } - if limit <= 0 { - limit = 10000 - } - if limit > 100000 { - limit = 100000 - } - detail := map[string]any{"policy": policy, "dry_run": dryRun, "limit": limit} - switch CleanupPolicy(policy) { - case PolicyJobAuditRetention: - cutoff := time.Now().UTC().Add(-90 * 24 * time.Hour) - if dryRun { - var n int64 - err := pool.QueryRow(ctx, ` - SELECT count(*) FROM job_audit - WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled')`, cutoff).Scan(&n) - detail["would_delete"] = n - return detail, err - } - tag, err := pool.Exec(ctx, ` - DELETE FROM job_audit - WHERE id IN ( - SELECT id FROM job_audit - WHERE created_at < $1 AND status IN ('succeeded', 'failed', 'cancelled') - LIMIT $2 - )`, cutoff, limit) - if err != nil { - return detail, err - } - detail["deleted"] = tag.RowsAffected() - return detail, nil - case PolicyASNCacheRetention: - cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour) - if dryRun { - var n int64 - err := pool.QueryRow(ctx, `SELECT count(*) FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff).Scan(&n) - detail["would_delete"] = n - return detail, err - } - tag, err := pool.Exec(ctx, ` - DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, cutoff) - if err != nil { - return detail, err - } - detail["deleted"] = tag.RowsAffected() - return detail, nil - default: - return nil, fmt.Errorf("pgmonitor: unknown cleanup policy %q", policy) - } + PolicyID string `json:"policy_id"` + Policy string `json:"policy"` + DryRun bool `json:"dry_run"` + Limit int `json:"limit"` } // InsertMaintenanceAudit records an audit row at job start. diff --git a/internal/repository/maintenance.go b/internal/repository/maintenance.go index 808f32d..280baa0 100644 --- a/internal/repository/maintenance.go +++ b/internal/repository/maintenance.go @@ -1,29 +1,8 @@ package repository -import ( - "context" - "time" -) +import "context" -const ( - jobAuditRetentionDays = 90 - asnCacheRetentionDays = 7 -) - -// RunPeriodicMaintenance prunes stale job_audit and asn_prefix_cache rows (PostgreSQL). +// RunPeriodicMaintenance is a no-op; retention is driven by maintenance_policy rows (UI-configured). func (p *Postgres) RunPeriodicMaintenance(ctx context.Context) { - if p == nil || p.pool == nil { - return - } - if ctx == nil { - ctx = context.Background() - } - jobCutoff := time.Now().UTC().Add(-time.Duration(jobAuditRetentionDays) * 24 * time.Hour) - _, _ = p.pool.Exec(ctx, ` - DELETE FROM job_audit - WHERE created_at < $1 - AND status IN ('succeeded', 'failed', 'cancelled')`, jobCutoff) - asnCutoff := time.Now().UTC().Add(-time.Duration(asnCacheRetentionDays) * 24 * time.Hour) - _, _ = p.pool.Exec(ctx, ` - DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, asnCutoff) + _ = ctx }