From d7b526a1127b18e4ca8b066a160f3d630b197c73 Mon Sep 17 00:00:00 2001 From: dtoro Date: Tue, 28 Jul 2026 14:03:30 +0200 Subject: [PATCH] fix(scheduler): honour check_defs.interval_s, and renumber migrations off main MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListEnabledCheckDefs selected interval_s but never filtered on it, so every enabled check ran on every 30s pass and the declared per-check intervals were decorative. Invisible at 17 enabled checks; at ~180 it would have meant ~126 SSH connections every 30s (~363k/day) and `apt update` on every machine every 30 seconds — 14,400 mirror hits a day to answer a question that changes daily. - check_defs.last_run_at (migration 026) + a due-ness predicate in the query. A column rather than scheduler memory because this control plane restarts on every deploy, and an in-memory map would re-fire every check on each restart. - runCheck stamps last_run_at before processing the result, so a permanently failing check backs off to its interval instead of re-running every pass. - updates and backup-freshness drop to daily. Both answer questions whose answers change about once a day; 60s was just the shared ssh-script default. - last_run_at is seeded to a random offset within the interval so checks created by the same seed do not stay in lockstep — otherwise ~165 probes land in the same instant each minute instead of spread across it. Deliberately not in the upsert's DO UPDATE: a re-seed must not re-herd them. Steady state becomes ~180k SSH/day (down from ~363k) and 5 apt runs/day (down from 14,400), with each 60s check landing at its own point in the minute. Also renumbers 022→023, 023→024, 024→025: origin/main added its own 022_knowledge_revisions, and prod has already applied version 22. Left colliding, prod would have skipped the monitoring_spec migration entirely and then failed the seed on a missing column. Co-Authored-By: Claude --- internal/checkdefaults/defaults.go | 25 +++++++++++---- internal/db/queries/operations.sql | 10 +++++- internal/db/seed.go | 17 +++++----- internal/db/sqlcgen/models.go | 14 ++++++++ internal/db/sqlcgen/ontology.sql.go | 3 +- internal/db/sqlcgen/operations.sql.go | 17 +++++++++- internal/scheduler/scheduler.go | 14 +++++--- ....sql => 023_entity_type_monitoring.up.sql} | 2 +- ...=> 024_executions_created_at_index.up.sql} | 2 +- ..._logs.up.sql => 025_execution_logs.up.sql} | 2 +- migrations/026_check_defs_last_run.up.sql | 32 +++++++++++++++++++ 11 files changed, 113 insertions(+), 25 deletions(-) rename migrations/{022_entity_type_monitoring.up.sql => 023_entity_type_monitoring.up.sql} (98%) rename migrations/{023_executions_created_at_index.up.sql => 024_executions_created_at_index.up.sql} (95%) rename migrations/{024_execution_logs.up.sql => 025_execution_logs.up.sql} (98%) create mode 100644 migrations/026_check_defs_last_run.up.sql diff --git a/internal/checkdefaults/defaults.go b/internal/checkdefaults/defaults.go index 39b2bb1..2b4dc21 100644 --- a/internal/checkdefaults/defaults.go +++ b/internal/checkdefaults/defaults.go @@ -183,7 +183,13 @@ func buildKind(kind string, t Target, attrs map[string]any, host, user string, p if host == "" { return nil, "no address on the entity or its host" } - return []checkDef{ssh("updates_check.sh")}, "" + // Daily. updates_check.sh runs `apt update` against the distro + // mirrors; the shared 60s ssh-script default would have meant 1,440 + // mirror hits per machine per day to answer a question whose answer + // changes about once a day. + u := ssh("updates_check.sh") + u.interval = 86400 + return []checkDef{u}, "" case KindCapacity: if host == "" { @@ -224,9 +230,9 @@ func buildKind(kind string, t Target, attrs map[string]any, host, user string, p if port != 0 && port != 22 { cfg["port"] = port } - // Hourly: a daily backup does not need a 60s probe, and each one is an - // SSH round trip. - return []checkDef{{kind: "backup-freshness", config: cfg, interval: 3600}}, "" + // Daily. The freshness budget itself is a day, so probing more often + // cannot surface anything sooner — it just costs an SSH round trip. + return []checkDef{{kind: "backup-freshness", config: cfg, interval: 86400}}, "" case KindHTTP: url := httpURL(t, attrs) @@ -286,9 +292,16 @@ func writeCheck(ctx context.Context, tx pgx.Tx, t Target, idx int, def checkDef) // Config is derived from the seed, so the seed wins on re-ingest and // attribute changes propagate. `enabled` is deliberately left alone: it // is operational state an operator may have toggled. + // last_run_at is seeded to a random point inside the interval so checks + // created together do not stay in lockstep. Every check the seed creates + // would otherwise come due in the same instant forever: ~165 probes + // landing at once each minute rather than spread across it. Deliberately + // absent from the DO UPDATE below — a re-seed must not reset the schedule + // and re-herd everything. tag, err := tx.Exec(ctx, - `INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled) - VALUES ($1, $2, $3, $4, $5, 30, true) + `INSERT INTO check_defs (entity_id, target_id, kind, config, interval_s, timeout_s, enabled, last_run_at) + VALUES ($1, $2, $3, $4, $5, 30, true, + now() - make_interval(secs => random() * $5::int)) ON CONFLICT (entity_id) DO UPDATE SET target_id = EXCLUDED.target_id, kind = EXCLUDED.kind, config = EXCLUDED.config, interval_s = EXCLUDED.interval_s, diff --git a/internal/db/queries/operations.sql b/internal/db/queries/operations.sql index 4baf585..0172b30 100644 --- a/internal/db/queries/operations.sql +++ b/internal/db/queries/operations.sql @@ -55,12 +55,20 @@ FROM events WHERE id > $1 ORDER BY id ASC LIMIT $2; -- ===================================================================== -- name: ListEnabledCheckDefs :many +-- Enabled AND due. interval_s used to be selected but never filtered on, so +-- every check ran on every 30s pass and the declared intervals meant nothing. +-- NULL last_run_at = never run = due now. SELECT cd.entity_id, cd.target_id, cd.target_type, cd.kind, cd.config, cd.interval_s, cd.timeout_s, cd.zone, cd.enabled, cd.updated_at, e.slug AS entity_slug FROM check_defs cd JOIN entities e ON e.id = cd.entity_id -WHERE cd.enabled = true; +WHERE cd.enabled = true + AND (cd.last_run_at IS NULL + OR cd.last_run_at <= now() - make_interval(secs => cd.interval_s)); + +-- name: MarkCheckRun :exec +UPDATE check_defs SET last_run_at = now() WHERE entity_id = $1; -- name: GetCheckDef :one SELECT * FROM check_defs WHERE entity_id = $1; diff --git a/internal/db/seed.go b/internal/db/seed.go index 1df7c2a..20a2cbc 100644 --- a/internal/db/seed.go +++ b/internal/db/seed.go @@ -13,15 +13,15 @@ import ( // SeedResult holds counts from a seed ingest operation. type SeedResult struct { - Lifecycles int - EntityTypes int + Lifecycles int + EntityTypes int RelationshipTypes int - Entities int - Relationships int - RiskClasses int - ApprovalRules int - AutonomySettings int - Checks int + Entities int + Relationships int + RiskClasses int + ApprovalRules int + AutonomySettings int + Checks int } // IngestOntologySeed ingests seeds/ontology.yaml into the DB. @@ -476,4 +476,3 @@ func keysOf(m map[string]map[string]any) []string { } return keys } - diff --git a/internal/db/sqlcgen/models.go b/internal/db/sqlcgen/models.go index 7fcd8bf..e584131 100644 --- a/internal/db/sqlcgen/models.go +++ b/internal/db/sqlcgen/models.go @@ -46,6 +46,8 @@ type AgentSession struct { Summary string EntityID *uuid.UUID CompletionNudges int32 + Blocker string + ClosedAt *time.Time } type Approval struct { @@ -110,6 +112,8 @@ type CheckDef struct { Zone *string Enabled bool UpdatedAt time.Time + // When this check last executed. NULL = never, due immediately. Compared against interval_s to decide due-ness. + LastRunAt *time.Time } type Classification struct { @@ -177,6 +181,8 @@ type EntityType struct { Status string CreatedAt time.Time UpdatedAt time.Time + // Check kinds this type warrants, resolved through parent_type. NULL means undeclared (an ontology gap), [] means explicitly unmonitorable, ["http","resource"] means declared kinds. Populated from seeds/ontology.yaml. + MonitoringSpec []byte } type Event struct { @@ -211,6 +217,14 @@ type Execution struct { CreatedAt time.Time } +type ExecutionLog struct { + ExecutionID uuid.UUID + Ts time.Time + Seq int32 + Stream string + Chunk string +} + type Feedback struct { EntityID uuid.UUID ExecutionID uuid.UUID diff --git a/internal/db/sqlcgen/ontology.sql.go b/internal/db/sqlcgen/ontology.sql.go index 5a3acf8..d8091f4 100644 --- a/internal/db/sqlcgen/ontology.sql.go +++ b/internal/db/sqlcgen/ontology.sql.go @@ -30,7 +30,7 @@ func (q *Queries) GetLifecycleForType(ctx context.Context, name string) (Lifecyc } const listEntityTypes = `-- name: ListEntityTypes :many -SELECT name, parent_type, is_abstract, domain, layer, description, lifecycle_id, attribute_schema, schema_version, status, created_at, updated_at FROM entity_types ORDER BY name +SELECT name, parent_type, is_abstract, domain, layer, description, lifecycle_id, attribute_schema, schema_version, status, created_at, updated_at, monitoring_spec FROM entity_types ORDER BY name ` func (q *Queries) ListEntityTypes(ctx context.Context) ([]EntityType, error) { @@ -55,6 +55,7 @@ func (q *Queries) ListEntityTypes(ctx context.Context) ([]EntityType, error) { &i.Status, &i.CreatedAt, &i.UpdatedAt, + &i.MonitoringSpec, ); err != nil { return nil, err } diff --git a/internal/db/sqlcgen/operations.sql.go b/internal/db/sqlcgen/operations.sql.go index 05670d2..a0e52d3 100644 --- a/internal/db/sqlcgen/operations.sql.go +++ b/internal/db/sqlcgen/operations.sql.go @@ -51,7 +51,7 @@ func (q *Queries) GetAutonomySetting(ctx context.Context, key string) (string, e } const getCheckDef = `-- name: GetCheckDef :one -SELECT entity_id, target_id, target_type, kind, config, interval_s, timeout_s, zone, enabled, updated_at FROM check_defs WHERE entity_id = $1 +SELECT entity_id, target_id, target_type, kind, config, interval_s, timeout_s, zone, enabled, updated_at, last_run_at FROM check_defs WHERE entity_id = $1 ` func (q *Queries) GetCheckDef(ctx context.Context, entityID uuid.UUID) (CheckDef, error) { @@ -68,6 +68,7 @@ func (q *Queries) GetCheckDef(ctx context.Context, entityID uuid.UUID) (CheckDef &i.Zone, &i.Enabled, &i.UpdatedAt, + &i.LastRunAt, ) return i, err } @@ -695,6 +696,8 @@ SELECT cd.entity_id, cd.target_id, cd.target_type, cd.kind, cd.config, FROM check_defs cd JOIN entities e ON e.id = cd.entity_id WHERE cd.enabled = true + AND (cd.last_run_at IS NULL + OR cd.last_run_at <= now() - make_interval(secs => cd.interval_s)) ` type ListEnabledCheckDefsRow struct { @@ -714,6 +717,9 @@ type ListEnabledCheckDefsRow struct { // ===================================================================== // Phase 3 queries // ===================================================================== +// Enabled AND due. interval_s used to be selected but never filtered on, so +// every check ran on every 30s pass and the declared intervals meant nothing. +// NULL last_run_at = never run = due now. func (q *Queries) ListEnabledCheckDefs(ctx context.Context) ([]ListEnabledCheckDefsRow, error) { rows, err := q.db.Query(ctx, listEnabledCheckDefs) if err != nil { @@ -1126,6 +1132,15 @@ func (q *Queries) ListSkills(ctx context.Context, status *string) ([]Skill, erro return items, nil } +const markCheckRun = `-- name: MarkCheckRun :exec +UPDATE check_defs SET last_run_at = now() WHERE entity_id = $1 +` + +func (q *Queries) MarkCheckRun(ctx context.Context, entityID uuid.UUID) error { + _, err := q.db.Exec(ctx, markCheckRun, entityID) + return err +} + const putIdempotentResponse = `-- name: PutIdempotentResponse :exec INSERT INTO idempotency_keys (actor, key, request_hash, response_code, response_body) VALUES ($1, $2, $3, $4, $5) diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 90a6127..c8868c3 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -113,6 +113,13 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef result := executeCheck(ctx, cd) + // Stamp the run before processing the result: due-ness must advance even + // when a check fails, or a permanently failing check would be re-run on + // every pass instead of at its declared interval. + if err := q.MarkCheckRun(ctx, cd.EntityID); err != nil { + slog.Error("scheduler: mark check run", "entity", cd.EntitySlug, "error", err) + } + latency := time.Since(start).Milliseconds() if result.metrics == nil { @@ -483,8 +490,8 @@ func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkRes if usedPct > float64(cfg.ThresholdPct) { return checkResult{ health: "degraded", signalKind: "disk", - evidence: fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct), - metrics: metrics, + evidence: fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct), + metrics: metrics, } } @@ -754,7 +761,6 @@ func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Dur return out, nil } - // metricThreshold defines warn/crit thresholds for a single metric. type metricThreshold struct { Warn float64 `json:"warn"` @@ -790,4 +796,4 @@ func evaluateSeverity(kind string, signalKind string, config []byte, metrics map return "warning" } -var _ = uuid.UUID{} // ensure uuid import stays \ No newline at end of file +var _ = uuid.UUID{} // ensure uuid import stays diff --git a/migrations/022_entity_type_monitoring.up.sql b/migrations/023_entity_type_monitoring.up.sql similarity index 98% rename from migrations/022_entity_type_monitoring.up.sql rename to migrations/023_entity_type_monitoring.up.sql index 24e9318..65dde46 100644 --- a/migrations/022_entity_type_monitoring.up.sql +++ b/migrations/023_entity_type_monitoring.up.sql @@ -1,4 +1,4 @@ --- 022_entity_type_monitoring.up.sql +-- 023_entity_type_monitoring.up.sql -- Declare, per entity type, what monitoring that type warrants. -- -- Motivation: only 3 of 89 active entities had an enabled check_def, because diff --git a/migrations/023_executions_created_at_index.up.sql b/migrations/024_executions_created_at_index.up.sql similarity index 95% rename from migrations/023_executions_created_at_index.up.sql rename to migrations/024_executions_created_at_index.up.sql index 7c821f6..7345716 100644 --- a/migrations/023_executions_created_at_index.up.sql +++ b/migrations/024_executions_created_at_index.up.sql @@ -1,4 +1,4 @@ --- 023_executions_created_at_index.up.sql +-- 024_executions_created_at_index.up.sql -- Support newest-first execution history. -- -- ListExecutions previously ordered by the target entity's slug, which is diff --git a/migrations/024_execution_logs.up.sql b/migrations/025_execution_logs.up.sql similarity index 98% rename from migrations/024_execution_logs.up.sql rename to migrations/025_execution_logs.up.sql index 9489cda..0eba15a 100644 --- a/migrations/024_execution_logs.up.sql +++ b/migrations/025_execution_logs.up.sql @@ -1,4 +1,4 @@ --- 024_execution_logs.up.sql +-- 025_execution_logs.up.sql -- Incremental command output for executions. -- -- Until now `executions.result` was a single JSONB blob written once, at the diff --git a/migrations/026_check_defs_last_run.up.sql b/migrations/026_check_defs_last_run.up.sql new file mode 100644 index 0000000..179c109 --- /dev/null +++ b/migrations/026_check_defs_last_run.up.sql @@ -0,0 +1,32 @@ +-- 026_check_defs_last_run.up.sql +-- Make check_defs.interval_s actually mean something. +-- +-- ListEnabledCheckDefs selected interval_s but never filtered on it, and +-- nothing in the scheduler read it except staleSweep. So every enabled check +-- ran on every 30-second pass and the declared per-check intervals were +-- decorative. +-- +-- That went unnoticed at 17 enabled checks (~0.5 SSH/s). Restoring monitoring +-- coverage takes it to ~150, where it would have meant ~126 SSH connections +-- every 30s — roughly 363k/day — and, worst of all, `apt update` on every +-- machine every 30 seconds via updates_check.sh: 14,400 mirror hits a day to +-- answer a question whose answer changes about once a day. +-- +-- last_run_at is a column rather than scheduler memory on purpose: an +-- in-memory map resets on restart, and this control plane restarts on every +-- deploy, so every check would fire at once each time — a thundering herd +-- exactly when the stack is least settled. +-- +-- NULL means "never run", which is due immediately. Existing rows therefore +-- all fire once on the first pass after this migration, then settle into +-- their declared cadence. + +ALTER TABLE check_defs ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMPTZ; + +-- The scheduler's hot query: enabled AND due. Partial on enabled since +-- disabled checks are never considered. +CREATE INDEX IF NOT EXISTS idx_check_defs_due + ON check_defs (last_run_at) + WHERE enabled; + +COMMENT ON COLUMN check_defs.last_run_at IS 'When this check last executed. NULL = never, due immediately. Compared against interval_s to decide due-ness.';