oikos/internal/checkdefaults/defaults.go

// 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"
		}
		// A service's name is a logical label, not usually its systemd unit
		// or container name (matrix = matrix-synapse.service + containers).
		// Prefer an explicit probe target when declared; process_check.sh also
		// matches a unit prefix or a docker container as a fallback.
		unit := ""
		for _, key := range []string{"probe_unit", "systemd_unit", "container"} {
			if v, _ := attrs[key].(string); v != "" {
				unit = v
				break
			}
		}
		// Ontology intent: "http when it has a url, else a process check." A
		// url-fronted service is already liveness-probed via http (the real
		// endpoint, through the TLS terminator); the process check is redundant
		// and fragile (needs host access + the exact unit/container name), and
		// under worst-of aggregation it lets a broken supplementary probe veto
		// a working service. Emit it only for services WITHOUT a url, or when
		// an explicit probe_unit opts into binary-level depth.
		if unit == "" {
			if httpURL(t, attrs) != "" {
				return nil, "url present and no probe_unit; http check covers liveness"
			}
			unit = t.Name
		}
		if unit == "" {
			return nil, "no name to check a process for"
		}
		// process_check.sh takes the unit/container name as $1 and reports
		// "unknown" without it.
		return []checkDef{ssh("process_check.sh", unit)}, ""

	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 whose cert to read (SNI / cert CN). Prefer an explicit
		// `hostname` attribute, then `cn`, then a dotted name. Hourly: expiry
		// changes once a day, but a renewal or 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"
		}
		// `dial` is the TLS terminator's address to connect to (Caddy's lab
		// IP), used when the hostname doesn't resolve/reach from the scheduler.
		// Without it the probe can't reach *.hubris.network from a container
		// with no mesh / split-horizon DNS.
		dial, _ := attrs["dial"].(string)
		config := map[string]any{"host": host, "warn_days": 30, "crit_days": 7}
		if dial != "" {
			config["dial"] = dial
		}
		return []checkDef{{
			kind:     "cert-expiry",
			config:   config,
			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)
		}
	}
}