From 6510a9ca2239f8263ab2a19eca74c321a12d6171 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Fri, 12 Jun 2026 13:21:32 +0700 Subject: [PATCH] feat(maintenance): add policy executor and config provider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConfigProvider, PolicyExecutor, DBStatsProvider, scheduler и safety; зависимость robfig/cron/v3. Co-authored-by: Cursor --- go.mod | 1 + go.sum | 2 + internal/maintenance/config_provider.go | 67 ++++++++++ internal/maintenance/db_stats_provider.go | 62 +++++++++ internal/maintenance/doc.go | 2 + internal/maintenance/policy_executor.go | 150 ++++++++++++++++++++++ internal/maintenance/safety.go | 72 +++++++++++ internal/maintenance/safety_test.go | 46 +++++++ internal/maintenance/scheduler.go | 93 ++++++++++++++ 9 files changed, 495 insertions(+) create mode 100644 internal/maintenance/config_provider.go create mode 100644 internal/maintenance/db_stats_provider.go create mode 100644 internal/maintenance/doc.go create mode 100644 internal/maintenance/policy_executor.go create mode 100644 internal/maintenance/safety.go create mode 100644 internal/maintenance/safety_test.go create mode 100644 internal/maintenance/scheduler.go diff --git a/go.mod b/go.mod index 2181546..f379872 100644 --- a/go.mod +++ b/go.mod @@ -25,6 +25,7 @@ require ( github.com/prometheus/common v0.55.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect golang.org/x/crypto v0.46.0 // indirect golang.org/x/mod v0.31.0 // indirect golang.org/x/net v0.48.0 // indirect diff --git a/go.sum b/go.sum index eae2bcf..2a7a2ec 100644 --- a/go.sum +++ b/go.sum @@ -46,6 +46,8 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= diff --git a/internal/maintenance/config_provider.go b/internal/maintenance/config_provider.go new file mode 100644 index 0000000..18fbbd3 --- /dev/null +++ b/internal/maintenance/config_provider.go @@ -0,0 +1,67 @@ +package maintenance + +import ( + "context" + "sync" + + "evobgp/internal/store" +) + +// ConfigProvider caches maintenance policies from store.Backend with hot reload. +type ConfigProvider struct { + store store.Backend + mu sync.RWMutex + items []*store.MaintenancePolicy +} + +// NewConfigProvider constructs a provider; call Reload before use. +func NewConfigProvider(st store.Backend) *ConfigProvider { + return &ConfigProvider{store: st} +} + +// Reload loads all policies from the database into memory. +func (c *ConfigProvider) Reload(ctx context.Context) error { + if c == nil || c.store == nil { + return nil + } + _ = ctx + items, _, _, err := c.store.ListMaintenancePolicies("", 1000) + if err != nil { + return err + } + cp := make([]*store.MaintenancePolicy, len(items)) + copy(cp, items) + c.mu.Lock() + c.items = cp + c.mu.Unlock() + return nil +} + +// Snapshot returns a copy of cached policies. +func (c *ConfigProvider) Snapshot() []*store.MaintenancePolicy { + if c == nil { + return nil + } + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]*store.MaintenancePolicy, len(c.items)) + copy(out, c.items) + return out +} + +// Get returns one policy by id from cache or store. +func (c *ConfigProvider) Get(ctx context.Context, id string) (*store.MaintenancePolicy, error) { + if c == nil || c.store == nil { + return nil, store.ErrNotFound + } + c.mu.RLock() + for _, p := range c.items { + if p.ID == id { + cp := *p + c.mu.RUnlock() + return &cp, nil + } + } + c.mu.RUnlock() + return c.store.GetMaintenancePolicy(id) +} diff --git a/internal/maintenance/db_stats_provider.go b/internal/maintenance/db_stats_provider.go new file mode 100644 index 0000000..3d16a36 --- /dev/null +++ b/internal/maintenance/db_stats_provider.go @@ -0,0 +1,62 @@ +package maintenance + +import ( + "context" + "fmt" + + "evobgp/internal/pgmonitor" +) + +// TableHints are PostgreSQL statistics hints for UI recommendations. +type TableHints struct { + TableName string `json:"table_name"` + DeadTuples int64 `json:"n_dead_tup"` + BloatRatio float64 `json:"bloat_ratio,omitempty"` + LastAutovacuum string `json:"last_autovacuum,omitempty"` + RecommendVacuum bool `json:"recommend_vacuum"` + Detail string `json:"detail,omitempty"` + Refs []string `json:"refs,omitempty"` +} + +// DBStatsProvider wraps pgmonitor for maintenance policy hints. +type DBStatsProvider struct { + pg *pgmonitor.Service +} + +// NewDBStatsProvider constructs a stats provider. +func NewDBStatsProvider(pg *pgmonitor.Service) *DBStatsProvider { + return &DBStatsProvider{pg: pg} +} + +// Hints returns table-level vacuum/bloat hints. +func (d *DBStatsProvider) Hints(ctx context.Context, tableName string) (TableHints, error) { + out := TableHints{TableName: tableName} + if d == nil || d.pg == nil { + return out, fmt.Errorf("maintenance: postgres monitoring not configured") + } + if err := ValidateTableName(tableName); err != nil { + return out, err + } + tables, err := d.pg.Tables(ctx, 100) + if err != nil { + return out, err + } + for _, t := range tables { + if t.Relname != tableName { + continue + } + out.DeadTuples = t.DeadTuples + out.BloatRatio = t.BloatRatio + if t.LastAutovacuum != nil { + out.LastAutovacuum = t.LastAutovacuum.UTC().Format("2006-01-02T15:04:05Z") + } + if t.BloatRatio > 0.2 && t.DeadTuples > 5000 { + out.RecommendVacuum = true + out.Detail = "Высокая доля n_dead_tup; рекомендуется VACUUM." + out.Refs = []string{t.Relname} + } + return out, nil + } + out.Detail = "Таблица не найдена в pg_stat_user_tables (top by size)." + return out, nil +} diff --git a/internal/maintenance/doc.go b/internal/maintenance/doc.go new file mode 100644 index 0000000..547e6b1 --- /dev/null +++ b/internal/maintenance/doc.go @@ -0,0 +1,2 @@ +// Package maintenance implements PostgreSQL maintenance policies loaded from the database. +package maintenance diff --git a/internal/maintenance/policy_executor.go b/internal/maintenance/policy_executor.go new file mode 100644 index 0000000..63338c2 --- /dev/null +++ b/internal/maintenance/policy_executor.go @@ -0,0 +1,150 @@ +package maintenance + +import ( + "context" + "fmt" + "hash/fnv" + "strings" + "time" + + "evobgp/internal/pgmonitor" + "evobgp/internal/store" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// PolicyExecutor runs maintenance policies against PostgreSQL. +type PolicyExecutor struct { + Store store.Backend + Pool *pgxpool.Pool +} + +// Execute runs cleanup and/or vacuum steps for a policy. +func (e *PolicyExecutor) Execute(ctx context.Context, policy *store.MaintenancePolicy, dryRun bool) (map[string]any, error) { + if e == nil || e.Pool == nil { + return nil, fmt.Errorf("maintenance: postgres not configured") + } + if policy == nil { + return nil, store.ErrInvalidInput + } + if err := ValidateTableName(policy.TableName); err != nil { + return nil, err + } + if err := ValidateCondition(policy.Condition); err != nil { + return nil, err + } + if !store.ValidVacuumStrategy(policy.VacuumStrategy) { + return nil, store.ErrInvalidInput + } + + detail := map[string]any{ + "policy_id": policy.ID, + "table": policy.TableName, + "dry_run": dryRun, + } + + if policy.RetentionPeriodSec != nil || policy.MaxRows != nil { + cleanupDetail, err := e.runCleanup(ctx, policy, dryRun) + for k, v := range cleanupDetail { + detail[k] = v + } + if err != nil { + return detail, err + } + } + + if policy.VacuumStrategy != store.VacuumStrategyNone { + kind := vacuumKind(policy.VacuumStrategy) + vacDetail, err := pgmonitor.ExecMaintenance(ctx, e.Pool, kind, policy.TableName, dryRun) + if vacDetail != nil { + detail["vacuum"] = vacDetail + } + if err != nil { + return detail, err + } + } + + if !dryRun && e.Store != nil { + _ = e.Store.TouchMaintenancePolicyRun(policy.ID, "succeeded", "") + } + + return detail, nil +} + +func (e *PolicyExecutor) runCleanup(ctx context.Context, policy *store.MaintenancePolicy, dryRun bool) (map[string]any, error) { + detail := map[string]any{"cleanup": true} + limit := NormalizeBatchLimit(policy.MaxRows) + qualTable := pgx.Identifier{policy.TableName}.Sanitize() + cond := store.NormalizeMaintenancePolicyCondition(policy.Condition) + + tx, err := e.Pool.Begin(ctx) + if err != nil { + return detail, fmt.Errorf("maintenance: begin tx: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + + lockKey := advisoryKey(policy.ID) + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, lockKey); err != nil { + return detail, fmt.Errorf("maintenance: advisory lock: %w", err) + } + if _, err := tx.Exec(ctx, fmt.Sprintf(`SET LOCAL statement_timeout = '%ds'`, DefaultStatementTimeoutSec)); err != nil { + return detail, fmt.Errorf("maintenance: statement_timeout: %w", err) + } + + var args []any + where := cond + argN := 1 + if policy.RetentionPeriodSec != nil && *policy.RetentionPeriodSec > 0 { + cutoff := time.Now().UTC().Add(-time.Duration(*policy.RetentionPeriodSec) * time.Second) + where = fmt.Sprintf("(%s) AND created_at < $%d", cond, argN) + args = append(args, cutoff) + argN++ + } + + countSQL := fmt.Sprintf(`SELECT count(*) FROM %s WHERE %s`, qualTable, where) + var wouldDelete int64 + if err := tx.QueryRow(ctx, countSQL, args...).Scan(&wouldDelete); err != nil { + return detail, fmt.Errorf("maintenance: count: %w", err) + } + detail["would_delete"] = wouldDelete + if dryRun { + return detail, nil + } + + deleteSQL := fmt.Sprintf(` + DELETE FROM %s WHERE ctid IN ( + SELECT ctid FROM %s WHERE %s LIMIT $%d + )`, qualTable, qualTable, where, argN) + args = append(args, limit) + tag, err := tx.Exec(ctx, deleteSQL, args...) + if err != nil { + return detail, fmt.Errorf("maintenance: delete: %w", err) + } + if err := tx.Commit(ctx); err != nil { + return detail, fmt.Errorf("maintenance: commit: %w", err) + } + detail["deleted"] = tag.RowsAffected() + return detail, nil +} + +func vacuumKind(strategy string) string { + switch strings.TrimSpace(strategy) { + case store.VacuumStrategyVacuum: + return "vacuum" + case store.VacuumStrategyAnalyze: + return "analyze" + case store.VacuumStrategyVacuumAnalyze: + return "vacuum_analyze" + case store.VacuumStrategyReindex: + return "reindex" + default: + return "vacuum" + } +} + +func advisoryKey(policyID string) int64 { + h := fnv.New64a() + _, _ = h.Write([]byte("maint:" + policyID)) + return int64(h.Sum64()) +} diff --git a/internal/maintenance/safety.go b/internal/maintenance/safety.go new file mode 100644 index 0000000..27f7d58 --- /dev/null +++ b/internal/maintenance/safety.go @@ -0,0 +1,72 @@ +package maintenance + +import ( + "fmt" + "regexp" + "strings" +) + +const ( + DefaultBatchRows = 10000 + MaxBatchRows = 100000 + DefaultStatementTimeoutSec = 30 +) + +var ( + blockedTableNames = map[string]struct{}{ + "schema_migrations": {}, + "tenant": {}, + "maintenance_policy": {}, + "maintenance_policy_config_audit": {}, + } + + sqlForbidden = regexp.MustCompile(`(?i)(;|--|/\*|\b(drop|truncate|insert|update|alter|create|grant|revoke|copy)\b)`) +) + +// ValidateTableName ensures table is a safe identifier and not blocked. +func ValidateTableName(name string) error { + name = strings.TrimSpace(name) + if name == "" || !isSafeIdent(name) { + return fmt.Errorf("maintenance: invalid table name") + } + if _, blocked := blockedTableNames[strings.ToLower(name)]; blocked { + return fmt.Errorf("maintenance: table %q is not allowed", name) + } + return nil +} + +// ValidateCondition ensures the WHERE fragment is safe for parameterized cleanup. +func ValidateCondition(condition string) error { + c := strings.TrimSpace(condition) + if c == "" { + return nil + } + if sqlForbidden.MatchString(c) { + return fmt.Errorf("maintenance: unsafe condition") + } + return nil +} + +// NormalizeBatchLimit clamps delete batch size. +func NormalizeBatchLimit(maxRows *int) int { + if maxRows == nil || *maxRows <= 0 { + return DefaultBatchRows + } + if *maxRows > MaxBatchRows { + return MaxBatchRows + } + return *maxRows +} + +func isSafeIdent(name string) bool { + if name == "" { + return false + } + for _, r := range name { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' { + continue + } + return false + } + return true +} diff --git a/internal/maintenance/safety_test.go b/internal/maintenance/safety_test.go new file mode 100644 index 0000000..685e55c --- /dev/null +++ b/internal/maintenance/safety_test.go @@ -0,0 +1,46 @@ +package maintenance + +import "testing" + +func TestValidateCondition(t *testing.T) { + tests := []struct { + cond string + ok bool + }{ + {"true", true}, + {"status IN ('succeeded', 'failed')", true}, + {"1=1; DROP TABLE tenant", false}, + {"x -- comment", false}, + } + for _, tc := range tests { + err := ValidateCondition(tc.cond) + if tc.ok && err != nil { + t.Fatalf("cond %q: want ok, got %v", tc.cond, err) + } + if !tc.ok && err == nil { + t.Fatalf("cond %q: want error", tc.cond) + } + } +} + +func TestValidateTableName(t *testing.T) { + if err := ValidateTableName("job_audit"); err != nil { + t.Fatal(err) + } + if err := ValidateTableName("tenant"); err == nil { + t.Fatal("expected blocked table") + } + if err := ValidateTableName("bad-name"); err == nil { + t.Fatal("expected invalid ident") + } +} + +func TestNormalizeBatchLimit(t *testing.T) { + if got := NormalizeBatchLimit(nil); got != DefaultBatchRows { + t.Fatalf("default=%d got=%d", DefaultBatchRows, got) + } + max := 200000 + if got := NormalizeBatchLimit(&max); got != MaxBatchRows { + t.Fatalf("max=%d got=%d", MaxBatchRows, got) + } +} diff --git a/internal/maintenance/scheduler.go b/internal/maintenance/scheduler.go new file mode 100644 index 0000000..4e4d34d --- /dev/null +++ b/internal/maintenance/scheduler.go @@ -0,0 +1,93 @@ +package maintenance + +import ( + "context" + "fmt" + "log" + "strings" + "sync" + "time" + + "evobgp/internal/jobs" + + "github.com/robfig/cron/v3" +) + +// StartScheduler enqueues maintenance_policy_run jobs when cron schedules match. +func StartScheduler(ctx context.Context, provider *ConfigProvider, reg *jobs.Registry, tick time.Duration) { + if provider == nil || reg == nil { + return + } + if tick <= 0 { + tick = 30 * time.Second + } + go func() { + parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) + var mu sync.Mutex + schedules := map[string]cron.Schedule{} + lastFired := map[string]time.Time{} + + rebuild := func() { + mu.Lock() + defer mu.Unlock() + schedules = map[string]cron.Schedule{} + for _, p := range provider.Snapshot() { + if p == nil || !p.Enabled || strings.TrimSpace(p.Schedule) == "" { + continue + } + sched, err := parser.Parse(p.Schedule) + if err != nil { + log.Printf("maintenance: invalid cron for policy %s: %v", p.ID, err) + continue + } + schedules[p.ID] = sched + } + } + + rebuild() + t := time.NewTicker(tick) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + rebuild() + now := time.Now().UTC() + mu.Lock() + for _, p := range provider.Snapshot() { + if p == nil || !p.Enabled { + continue + } + sched, ok := schedules[p.ID] + if !ok { + continue + } + prev := lastFired[p.ID] + if prev.IsZero() { + prev = now.Add(-time.Minute) + } + next := sched.Next(prev) + if next.After(now) { + continue + } + slot := next.Unix() / 60 + if lf, ok := lastFired[p.ID]; ok && lf.Unix()/60 == slot { + continue + } + lastFired[p.ID] = next + dry := p.DryRunEnabled + idem := fmt.Sprintf("maint-%s-%d", p.ID, slot) + key := idem + _, _, _ = reg.Enqueue("", "maintenance_policy_run", &key, nil, map[string]any{ + "policy_id": p.ID, + "dry_run": dry, + "trigger": "scheduler", + }) + } + mu.Unlock() + } + } + }() + log.Printf("maintenance: policy scheduler started (tick=%s)", tick) +}