// Package checkdefaults derives an entity's default check_defs from the // monitoring kinds its type declares in seeds/ontology.yaml. // // The type says WHAT to watch (`service: [http, process]`); this package // works out HOW — which concrete check_defs rows to write, and what host, // script or URL each needs. Deriving config here rather than in YAML keeps // the ontology declarative and keeps address resolution (which has to walk // the graph) in code. package checkdefaults import ( "context" "encoding/json" "fmt" "log/slog" "strings" "github.com/dtoro/oikos/internal/ontology" "github.com/google/uuid" "github.com/jackc/pgx/v5" ) // Semantic monitoring kinds, as declared on entity types. These are not // check_defs.kind values — one semantic kind can expand to several concrete // checks (`resource` becomes four ssh-script rows). const ( KindPing = "ping" KindResource = "resource" KindUpdates = "updates" KindProcess = "process" KindHTTP = "http" KindCapacity = "capacity" KindBackup = "backup-freshness" KindCertExpiry = "cert-expiry" ) // defaultBackupMaxAge is how long a backup target may go without a new // artifact before it is stale. A day suits the nightly jobs in this lab; // override per target with `backup_max_age_s` in the entity's attributes. const defaultBackupMaxAge = 86400 // Target is the entity default checks are being ensured for. type Target struct { ID uuid.UUID Slug string Type string // Name is the entity's name column, not an attribute. The old code read // attrs["name"], which is never populated — seeds put `name` beside // `attributes`, not inside it — so every service silently produced no // process check. Name string Attrs []byte } // Result reports what Ensure did, so callers can log a type that declared // monitoring but produced nothing instead of failing silently. type Result struct { Created int // Skipped records kinds that were declared but could not be built, with // the reason. A non-empty Skipped on an active entity is a real gap. Skipped []Skip // Undeclared is true when no ancestor of the type declared monitoring — // an ontology gap rather than a fleet gap. Undeclared bool } // Skip is one declared-but-unbuilt check kind. type Skip struct { Kind string Reason string } type checkDef struct { kind string config map[string]any interval int32 } // Ensure writes the default check_defs for one entity, idempotently. // // Returns the number of checks created. An entity whose type declares // monitoring it cannot satisfy comes back with a populated Skipped rather // than an error — a missing address is a modelling gap, not a failure of // this call. func Ensure(ctx context.Context, tx pgx.Tx, tree *ontology.TypeTree, t Target) (Result, error) { var res Result if _, err := tx.Exec(ctx, `INSERT INTO entity_status (entity_id, health, updated_at) VALUES ($1, 'unknown', now()) ON CONFLICT (entity_id) DO NOTHING`, t.ID); err != nil { return res, fmt.Errorf("entity_status %s: %w", t.Slug, err) } mon := tree.Monitoring(t.Type) if !mon.Declared { res.Undeclared = true return res, nil } if mon.None() { return res, nil } var attrs map[string]any if len(t.Attrs) > 0 { _ = json.Unmarshal(t.Attrs, &attrs) } if attrs == nil { attrs = map[string]any{} } // A service has no address of its own — it lives on the container that // provides it. Fall back to the graph before giving up. host := resolveHost(attrs) if host == "" { hostAttrs, err := hostViaGraph(ctx, tx, t.ID) if err != nil { return res, fmt.Errorf("resolve host for %s: %w", t.Slug, err) } host = resolveHost(hostAttrs) if user := resolveSSHUser(hostAttrs); host != "" && user != "root" { attrs["ssh"] = hostAttrs["ssh"] } } user := resolveSSHUser(attrs) port := resolveSSHPort(attrs) var defs []checkDef for _, kind := range mon.Kinds { built, reason := buildKind(kind, t, attrs, host, user, port) if len(built) == 0 { res.Skipped = append(res.Skipped, Skip{Kind: kind, Reason: reason}) continue } defs = append(defs, built...) } for i, def := range defs { created, err := writeCheck(ctx, tx, t, i, def) if err != nil { return res, fmt.Errorf("check %s/%s: %w", t.Slug, def.kind, err) } if created { res.Created++ } } return res, nil } // buildKind turns one declared semantic kind into concrete check_defs, or // returns the reason it could not. func buildKind(kind string, t Target, attrs map[string]any, host, user string, port int) ([]checkDef, string) { ssh := func(script string, args ...string) checkDef { cfg := map[string]any{"script": script, "host": host} if user != "" && user != "root" { cfg["user"] = user } if port != 0 && port != 22 { cfg["port"] = port } if len(args) > 0 && args[0] != "" { cfg["args"] = args[0] } return checkDef{kind: "ssh-script", config: cfg, interval: 60} } switch kind { case KindPing: if host == "" { return nil, "no address on the entity or its host" } return []checkDef{{kind: "ping", config: map[string]any{"host": host}, interval: 30}}, "" case KindResource: if host == "" { return nil, "no address on the entity or its host" } return []checkDef{ ssh("cpu_check.sh"), ssh("memory_check.sh"), ssh("load_check.sh"), ssh("disk_usage_check.sh"), }, "" case KindUpdates: if host == "" { return nil, "no address on the entity or its host" } // 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 == "" { return nil, "no address on the entity or its host" } return []checkDef{ssh("disk_usage_check.sh")}, "" case KindProcess: if host == "" { return nil, "no address on the entity or its host" } if t.Name == "" { return nil, "no name to check a process for" } // process_check.sh takes the unit name as $1 and reports "unknown" // without it. return []checkDef{ssh("process_check.sh", t.Name)}, "" case KindBackup: // A backup target is checked from the machine that writes to it, so it // needs both an address (resolved via the backs-up-to edge) and the // path to look at. path, _ := attrs["path"].(string) if path == "" { return nil, "entity carries no path attribute to check for backups" } if host == "" { return nil, "no address on the entity or whatever backs up to it" } maxAge := defaultBackupMaxAge if v, ok := attrs["backup_max_age_s"].(float64); ok && v > 0 { maxAge = int(v) } cfg := map[string]any{"path": path, "host": host, "max_age_s": maxAge} if user != "" && user != "root" { cfg["user"] = user } if port != 0 && port != 22 { cfg["port"] = port } // 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) if url == "" { return nil, "no url attribute, public_host, or hostname-shaped name" } // max_status rather than an exact expected_status: most services sit // behind Authentik and answer 302/401, which is a working service. return []checkDef{{ kind: "http", config: map[string]any{"url": url, "max_status": 500}, interval: 60, }}, "" case KindCertExpiry: // The host to TLS-dial for the cert. Prefer an explicit `hostname` // attribute, then `cn`, then a dotted name (a certificate's name is // its CN/SAN). Hourly: expiry changes once a day, but a renewal or a // mis-issued cert is worth noticing within the hour. host := certHost(t, attrs) if host == "" { return nil, "no hostname / cn / dotted name to dial for the cert" } return []checkDef{{ kind: "cert-expiry", config: map[string]any{"host": host, "warn_days": 30, "crit_days": 7}, interval: 3600, }}, "" } return nil, "no builder for this kind yet" } // certHost works out the hostname to TLS-dial for a certificate's expiry. func certHost(t Target, attrs map[string]any) string { for _, key := range []string{"hostname", "cn", "san"} { if v, ok := attrs[key].(string); ok && v != "" { return v } } // A dotted name is a hostname (hubris.network, media.hubris.network). if strings.Contains(t.Name, ".") && !strings.Contains(t.Name, " ") { return t.Name } return "" } // writeCheck upserts one check_def and its backing check entity. // // The entity upsert MUST return the row's id. The previous version generated // a fresh uuid, inserted ON CONFLICT (slug) DO NOTHING, then wrote a // check_defs row referencing that uuid. On any re-seed the slug already // existed, the entity insert became a no-op, and the check_defs insert // violated its foreign key — which aborted the whole ingest transaction and // made every subsequent statement fail with 25P02. Because the errors were // discarded, the only visible symptom was an unrelated failure much later. func writeCheck(ctx context.Context, tx pgx.Tx, t Target, idx int, def checkDef) (bool, error) { // The full target slug, not a truncation of it. shortSlug() took the last // 8 characters, so all 21 ingress routes collapsed to ".network" and // generated one identical check slug — they overwrote each other and 20 // of them ended up with no check at all. It also collided service:jellyfin // with lxc:jellyfin. Entity slugs are unique; use them. checkSlug := fmt.Sprintf("check:%s:%s:%d", def.kind, t.Slug, idx) newID, err := uuid.NewV7() if err != nil { newID = uuid.New() } var checkID uuid.UUID err = tx.QueryRow(ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, version, created_at, updated_at) VALUES ($1, $2, 'check', $2, 'active', '{}', 1, now(), now()) ON CONFLICT (slug) DO UPDATE SET updated_at = now() RETURNING id`, newID, checkSlug).Scan(&checkID) if err != nil { return false, fmt.Errorf("upsert check entity %s: %w", checkSlug, err) } configJSON, err := json.Marshal(def.config) if err != nil { return false, err } // 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, target_type, kind, config, interval_s, timeout_s, enabled, last_run_at) VALUES ($1, $2, $6, $3, $4, $5, 30, true, now() - make_interval(secs => random() * $5::int)) ON CONFLICT (entity_id) DO UPDATE SET target_id = EXCLUDED.target_id, target_type = EXCLUDED.target_type, kind = EXCLUDED.kind, config = EXCLUDED.config, interval_s = EXCLUDED.interval_s, updated_at = now()`, checkID, t.ID, def.kind, configJSON, def.interval, t.Type) if err != nil { return false, fmt.Errorf("upsert check_def %s: %w", checkSlug, err) } return tag.RowsAffected() > 0, nil } // httpURL works out what to GET for an http check. // // Ingress routes carry their hostname as the entity name rather than as an // attribute (`name: media.hubris.network`), and most declare no attributes at // all — so the name is the only thing to go on. Requiring a `url` attribute // left all 21 of them unmonitored, which is a shame given an ingress check is // the most end-to-end probe available: it exercises Caddy, DNS, TLS and the // upstream in one request. func httpURL(t Target, attrs map[string]any) string { if url, ok := attrs["url"].(string); ok && url != "" { return url } if h, ok := attrs["public_host"].(string); ok && h != "" { return "https://" + h } // A dotted name is a hostname; a service name like "jellyfin" is not. if strings.Contains(t.Name, ".") && !strings.Contains(t.Name, " ") { return "https://" + t.Name } return "" } // hostViaGraph returns the attributes of the entity that hosts or provides // this one, so a service can inherit its container's address. func hostViaGraph(ctx context.Context, tx pgx.Tx, entityID uuid.UUID) (map[string]any, error) { rows, err := tx.Query(ctx, ` SELECT e.attributes FROM relationships r JOIN entities e ON e.id = r.source_id WHERE r.target_id = $1 AND r.valid_to IS NULL -- backs-up-to points from the thing being backed up TO the target, -- so walking it backwards finds the machine that writes the backups -- — which is the only place a freshness check can run. AND r.type IN ('provides', 'hosts', 'runs-on', 'backs-up-to') ORDER BY CASE r.type WHEN 'provides' THEN 0 WHEN 'runs-on' THEN 1 WHEN 'backs-up-to' THEN 2 ELSE 3 END`, entityID) if err != nil { return nil, err } defer rows.Close() for rows.Next() { var raw []byte if err := rows.Scan(&raw); err != nil { return nil, err } var attrs map[string]any if json.Unmarshal(raw, &attrs) != nil { continue } if resolveHost(attrs) != "" { return attrs, nil } } return nil, rows.Err() } func resolveHost(attrs map[string]any) string { if attrs == nil { return "" } if ip, ok := attrs["lan_ip"].(string); ok && ip != "" { return ip } // public_ipv4 before mesh: the scheduler container has no mesh interface, // so a standalone-server reachable only by mesh IP (netbird-vps) is // unprobeable even though a public IPv4 is available. if ip, ok := attrs["public_ipv4"].(string); ok && ip != "" { return ip } if mesh, ok := attrs["mesh"].(map[string]any); ok { if nb, ok := mesh["netbird"].(map[string]any); ok { if ip, ok := nb["ip"].(string); ok && ip != "" { return ip } // Seeds record the mesh name, not an address — ws:mac-mini // carries only `fqdn`, which is why it resolved to nothing. if fqdn, ok := nb["fqdn"].(string); ok && fqdn != "" { return fqdn } } } if ip, ok := attrs["mesh_ip"].(string); ok && ip != "" { return ip } for _, key := range []string{"host", "address", "public_host"} { if v, ok := attrs[key].(string); ok && v != "" { return v } } return "" } func resolveSSHUser(attrs map[string]any) string { if ssh, ok := attrs["ssh"].(map[string]any); ok { if u, ok := ssh["user"].(string); ok && u != "" { return u } } // Workstations carry their login as a top-level `user` attribute // (mac-mini: user: dtoro) rather than under ssh.user. Take it only when // no explicit ssh.user was set, so a host that genuinely wants root still // gets root. if u, ok := attrs["user"].(string); ok && u != "" { return u } return "root" } func resolveSSHPort(attrs map[string]any) int { if ssh, ok := attrs["ssh"].(map[string]any); ok { switch p := ssh["port"].(type) { case float64: return int(p) case int: return p } } return 22 } // LogResult emits the one line that was missing: a type that asked for // monitoring and did not get it. func LogResult(slug, entityType string, res Result) { switch { case res.Undeclared: slog.Info("checkdefaults: type declares no monitoring", "entity", slug, "type", entityType) case len(res.Skipped) > 0: for _, s := range res.Skipped { slog.Warn("checkdefaults: declared check not created", "entity", slug, "type", entityType, "kind", s.Kind, "reason", s.Reason) } } }