fix(scheduler): honour check_defs.interval_s, and renumber migrations off main
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -476,4 +476,3 @@ func keysOf(m map[string]map[string]any) []string {
|
||||
}
|
||||
return keys
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
@@ -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"`
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
32
migrations/026_check_defs_last_run.up.sql
Normal file
32
migrations/026_check_defs_last_run.up.sql
Normal file
@@ -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.';
|
||||
Reference in New Issue
Block a user