dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

fix: scheduler wrote health/metrics/events to probe entities, not targets

Browse Source 
Problem: every host/service/lxc/etc. entity_status row was permanently
stuck at 'unknown' since creation. Verified against the live DB:
metric_samples had 17,559 rows, 100% attached to type='check' probe
entities and 0% to any real monitored entity; only 25 check entities
ever had real health written. check_defs.entity_id (the probe's own
bookkeeping entity) and check_defs.target_id (the host/service actually
being observed) were both real fields, but the scheduler wrote
UpsertEntityStatus/InsertMetricSample/emitSchedulerEvent keyed by
entity_id instead of target_id — so every check ran and every result
was real, it just landed on the wrong row. This is the mechanism behind
observed drift: the agent's dashboard/health tools reported the
internal probes' state, never the actual fleet.

Change:
- scheduler.go: runCheck/resolveSignal now resolve targetID from
  cd.TargetID (falling back to the check's own id if unset) and write
  status/metrics/events there. Signals stay keyed by the check entity,
  unchanged, matching their existing resolution logic.
- Added a staleness sweep to housekeeping(): an entity whose last
  observation is older than 3x its fastest enabled check's interval
  (floor 5m) is marked 'stale' and emits health.stale, so a stalled
  scheduler or disabled check_def can no longer look like current data
  forever.
- migrations/016: deletes the now-orphaned check-entity entity_status
  rows so dashboard/fleet-health rollups stop double-counting probes as
  monitored entities. Historical metric_samples on check entities are
  left as-is (time-series data, not safe to reattribute).
- openapi.yaml + regenerated gen code: Entity gains health/last_check_at;
  'stale' added to the health enum everywhere it's used.
- dashboard.go / GetFleetHealth / nomos's get_health_summary MCP tool:
  exclude type='check' entities from rollups.
- nomos/agent.go: replay prior turns' tool_use/tool_result pairs into
  the conversation instead of dropping them (previously only final text
  was replayed, forcing the agent to re-derive fleet state every turn),
  and inject a compact live fleet-health snapshot into the system prompt
  each turn so it starts oriented instead of spending an iteration on
  discovery.

Risk: config_mutation (schema-adjacent — new migration, no destructive
DDL, additive DELETE only on orphaned rows). No behavior change until
oikos-api/oikos-scheduler/nomos are rebuilt and redeployed.

Verification: go build/vet clean across the repo. Ran this worktree's
own API binary against the live dev Postgres on an alternate port
(read-only from the live containers' perspective) and confirmed
/api/v1/entities now returns health/last_check_at, and the dashboard
health rollup dropped from double-counting to an honest 168 unmonitored
entities (matches reality pre-deploy — the live scheduler hasn't run
the fixed code yet). Confirmed check_defs.target_id correctly maps
multiple checks to host:hubris via direct psql query.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
authentik Default Admin
2026-07-09 00:26:04 +02:00
parent a39e67b6e9
commit 279549c8c9
10 changed files with 798 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
internal/scheduler/scheduler.go
View File

@@ -94,6 +94,13 @@ func runCheckPass(ctx context.Context, pool *db.Pool) {
}
// runCheck executes a single check and processes the result.
//
// check_defs.entity_id identifies the *check* (probe) entity itself;
// check_defs.target_id identifies the entity actually being observed (the
// host/service/etc). Health, metrics, and events must attach to the target
// so the observed entity's own record reflects reality — not the internal
// probe. Signals stay keyed by the check entity (cd.EntityID), matching how
// they are created below and resolved elsewhere.
func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) {
q := sqlcgen.New(pool)
start := time.Now()
@@ -107,9 +114,14 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef
}
result.metrics["probe_latency_ms"] = float64(latency)
targetID := cd.EntityID
if cd.TargetID != nil {
targetID = *cd.TargetID
}
for metric, value := range result.metrics {
_ = q.InsertMetricSample(ctx, sqlcgen.InsertMetricSampleParams{
EntityID: cd.EntityID,
EntityID: targetID,
Metric: metric,
Value: value,
Tags: []byte(`{}`),
@@ -121,18 +133,18 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef
"entity", cd.EntitySlug, "kind", cd.Kind, "error", result.err)
}
prevHealth := currentHealth(ctx, pool, cd.EntityID)
prevHealth := currentHealth(ctx, pool, targetID)
if result.signalKind == "" || result.health == "healthy" {
resolveSignal(ctx, pool, cd.EntityID, cd.EntitySlug)
resolveSignal(ctx, pool, cd.EntityID, targetID, cd.EntitySlug)
_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
EntityID: cd.EntityID,
EntityID: targetID,
Health: "healthy",
LastCheckAt: &[]time.Time{time.Now()}[0],
Details: []byte(`{}`),
})
if prevHealth != "" && prevHealth != "healthy" {
emitSchedulerEvent(ctx, pool, "health.changed", cd.EntityID, "info",
emitSchedulerEvent(ctx, pool, "health.changed", targetID, "info",
map[string]any{"slug": cd.EntitySlug, "from": prevHealth, "to": "healthy"})
}
return
@@ -156,7 +168,7 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef
}
_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
EntityID: cd.EntityID,
EntityID: targetID,
Health: result.health,
LastCheckAt: &[]time.Time{time.Now()}[0],
Details: []byte(`{}`),
@@ -164,11 +176,11 @@ func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDef
_ = sig
if prevHealth == "" || prevHealth == "healthy" {
emitSchedulerEvent(ctx, pool, "signal.raised", cd.EntityID, severity,
emitSchedulerEvent(ctx, pool, "signal.raised", targetID, severity,
map[string]any{"slug": cd.EntitySlug, "kind": result.signalKind, "evidence": result.evidence})
}
if prevHealth != result.health {
emitSchedulerEvent(ctx, pool, "health.changed", cd.EntityID, severity,
emitSchedulerEvent(ctx, pool, "health.changed", targetID, severity,
map[string]any{"slug": cd.EntitySlug, "from": prevHealth, "to": result.health})
}
}
@@ -188,21 +200,23 @@ func emitSchedulerEvent(ctx context.Context, pool *db.Pool, eventType string, en
_ = observability.Event(ctx, sqlcgen.New(pool), eventType, &entityID, severity, "scheduler", "", data)
}
// resolveSignal resolves any open signal for the given check entity.
func resolveSignal(ctx context.Context, pool *db.Pool, entityID uuid.UUID, slug string) {
// resolveSignal resolves any open signal raised by the given check entity.
// checkID matches how signals are keyed (UpsertSignal uses the check's own
// entity id); targetID is the observed entity whose status this affects.
func resolveSignal(ctx context.Context, pool *db.Pool, checkID, targetID uuid.UUID, slug string) {
q := sqlcgen.New(pool)
// Check if there's an open signal on this entity
tag, err := pool.Exec(ctx, `UPDATE signals SET state = 'resolved', updated_at = now()
WHERE entity_id = $1 AND state = 'raised'`, entityID)
WHERE entity_id = $1 AND state = 'raised'`, checkID)
if err != nil {
return
}
if tag.RowsAffected() > 0 {
emitSchedulerEvent(ctx, pool, "signal.resolved", entityID, "info",
emitSchedulerEvent(ctx, pool, "signal.resolved", targetID, "info",
map[string]any{"slug": slug})
}
_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
EntityID: entityID,
EntityID: targetID,
Health: "healthy",
LastCheckAt: &[]time.Time{time.Now()}[0],
Details: []byte(`{}`),
@@ -250,10 +264,74 @@ func housekeeping(ctx context.Context, pool *db.Pool) {
slog.Error("scheduler: prune idempotency keys", "error", err)
}
staleSweep(ctx, pool)
// Log housekeeping completion
slog.Debug("scheduler: housekeeping done", "pruned_idempotency_before", cutoff.Format(time.RFC3339))
}
// staleMultiplier and staleFloor bound how long an entity can go unobserved
// before its last-known health is no longer trusted. An entity is stale once
// it has gone longer than staleMultiplier times its fastest enabled check's
// interval (or staleFloor, whichever is larger) without a fresh observation —
// covering both a stalled scheduler and a disabled/broken check_def.
const (
staleMultiplier = 3
staleFloor = 5 * time.Minute
)
// staleSweep marks entities whose last observation has aged past their
// check's expected cadence as 'stale', so the system never reports an old
// health value as if it were current. Runs once per housekeeping pass.
func staleSweep(ctx context.Context, pool *db.Pool) {
rows, err := pool.Query(ctx, `
SELECT e.id, e.slug, st.health
FROM entity_status st
JOIN entities e ON e.id = st.entity_id
JOIN (
SELECT target_id, MIN(interval_s) AS min_interval
FROM check_defs
WHERE enabled AND target_id IS NOT NULL
GROUP BY target_id
) iv ON iv.target_id = st.entity_id
WHERE st.health <> 'stale'
AND (st.last_check_at IS NULL
OR st.last_check_at < now() - make_interval(secs => GREATEST(iv.min_interval * $1, $2)))`,
staleMultiplier, int(staleFloor.Seconds()))
if err != nil {
slog.Error("scheduler: stale sweep query", "error", err)
return
}
defer rows.Close()
type staleEntity struct {
id uuid.UUID
slug string
health string
}
var stale []staleEntity
for rows.Next() {
var se staleEntity
if err := rows.Scan(&se.id, &se.slug, &se.health); err != nil {
continue
}
stale = append(stale, se)
}
rows.Close()
for _, se := range stale {
_, err := pool.Exec(ctx,
`UPDATE entity_status SET health = 'stale', updated_at = now() WHERE entity_id = $1`, se.id)
if err != nil {
slog.Error("scheduler: mark stale", "entity", se.slug, "error", err)
continue
}
slog.Warn("scheduler: entity stale", "entity", se.slug, "prev_health", se.health)
emitSchedulerEvent(ctx, pool, "health.stale", se.id, "warning",
map[string]any{"slug": se.slug, "from": se.health, "to": "stale"})
}
}
// checkHTTP performs an HTTP health check.
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
cfg := struct {