feat(observability): restore monitoring coverage, make gaps visible, stream executions
Monitoring coverage was 3 of 89 active entities. Three bugs, each hidden by discarded errors in checkdefaults: - writeCheck generated a fresh uuid, inserted the check entity ON CONFLICT (slug) DO NOTHING, then wrote a check_defs row referencing it. On any re-seed the slug already existed, the entity insert no-oped, and the FK violated — aborting the ingest transaction and surfacing as an unrelated failure several entities later. Re-seeding has been broken since; prod's coverage was frozen at its first successful seed. This is what TestSeedIngestIdempotentAndNoDuplicateEdges had been reporting. - shortSlug truncated to the last 8 chars, so all 21 ingress routes collapsed to ".network" and overwrote each other; service:jellyfin collided with lxc:jellyfin. - The ssh-script checker never read the `args` config checkdefaults wrote, so process_check.sh always ran without its unit name and returned "unknown". Coverage is now 75/89. Monitoring is declared per entity type in seeds/ontology.yaml and resolved through the is-a hierarchy, so a type can say it warrants nothing (site, lan, mesh, cluster) and never be reported as a gap. coverageSweep raises an `unmonitored` signal only where a type declares monitoring it lacks — 8 real gaps, no false positives. Also: - entity_types.attribute_schema was never ingested: the seed loader read "attribute_schema" but the YAML says "attributes", so all 60 types stored JSON null. - ListExecutions ignored its declared target/action/correlation_id filters and paginated on a non-unique target slug, dropping and repeating rows. - started_at was captured but only written at terminal state, so a running execution reported NULL for its whole life. The three MCP auto-run copies wrote no timing at all; they are now one autoRun helper. - SSH output was buffered to completion and discarded entirely on timeout. Both sshExec copies now stream through a shared execlog sink into execution_logs, and keep partial output when a command is cancelled. - executions.correlation_id was a random per-execution uuid that correlated nothing; it is now the chat session id, which is what lets the chat tail live output. - reversible_low had no auto-run branch despite policy declaring it unattended. Since computeCommandRisk never returns it, the class only arises when an agent declares it over a read_only command — so gating it penalised candor without adding safety. - backup-target gains a backup-freshness checker (portable find -mmin, since the first target is on macOS), resolving its host by walking backs-up-to backwards. The pre-deploy pg_dump is now a tracked backup target. UI: an Executions section on entity detail with live output tailing, and streamed output under a running `run` call in the chat timeline. Migrations 022-024. Ops.svelte and context.ts exclude execution.output from their refetch triggers, which would otherwise fire once a second per command. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
@@ -9,10 +10,12 @@ import (
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/dtoro/oikos/internal/db"
|
||||
"github.com/dtoro/oikos/internal/db/sqlcgen"
|
||||
"github.com/dtoro/oikos/internal/execlog"
|
||||
"github.com/dtoro/oikos/internal/observability"
|
||||
"github.com/google/uuid"
|
||||
"golang.org/x/crypto/ssh"
|
||||
@@ -71,7 +74,35 @@ func initSSH() {
|
||||
// report. Generous enough for a real apt/docker install; not infinite.
|
||||
const sshExecTimeout = 10 * time.Minute
|
||||
|
||||
// streamWriter buffers everything it is given while forwarding each write to a
|
||||
// sink. One on session.Stdout and another sharing the same buffer on
|
||||
// session.Stderr reproduces CombinedOutput's interleaving in the order the
|
||||
// remote end produced it. Mirrors the twin in internal/mcp/server.go.
|
||||
type streamWriter struct {
|
||||
mu *sync.Mutex
|
||||
buf *bytes.Buffer
|
||||
stream string
|
||||
sink execlog.Sink
|
||||
}
|
||||
|
||||
func (w *streamWriter) Write(p []byte) (int, error) {
|
||||
w.mu.Lock()
|
||||
w.buf.Write(p)
|
||||
w.mu.Unlock()
|
||||
if w.sink != nil {
|
||||
// Copy: the ssh library reuses p once Write returns.
|
||||
w.sink(w.stream, append([]byte(nil), p...))
|
||||
}
|
||||
return len(p), nil
|
||||
}
|
||||
|
||||
func sshExec(ctx context.Context, host, user, command string) (string, error) {
|
||||
return sshExecStream(ctx, host, user, command, nil)
|
||||
}
|
||||
|
||||
// sshExecStream runs a command and reports its combined output, forwarding
|
||||
// each chunk to sink as it arrives. A nil sink behaves exactly as before.
|
||||
func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (string, error) {
|
||||
initSSH()
|
||||
if len(_sshKey) == 0 {
|
||||
return "", fmt.Errorf("no SSH key available")
|
||||
@@ -105,50 +136,62 @@ func sshExec(ctx context.Context, host, user, command string) (string, error) {
|
||||
}
|
||||
defer session.Close()
|
||||
|
||||
type result struct {
|
||||
out []byte
|
||||
err error
|
||||
var (
|
||||
mu sync.Mutex
|
||||
buf bytes.Buffer
|
||||
)
|
||||
session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink}
|
||||
session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink}
|
||||
|
||||
collected := func() string {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return strings.TrimSpace(buf.String())
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
// See internal/mcp/server.go's sshExec for why this recovers rather
|
||||
// than letting a rare SSH-library panic crash the whole api process.
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)}
|
||||
done <- fmt.Errorf("panic in ssh exec: %v", r)
|
||||
}
|
||||
}()
|
||||
out, err := session.CombinedOutput(command)
|
||||
done <- result{out, err}
|
||||
// Run rather than CombinedOutput so the assigned writers are used;
|
||||
// Run returns only after both streams are fully drained.
|
||||
done <- session.Run(command)
|
||||
}()
|
||||
|
||||
select {
|
||||
case r := <-done:
|
||||
text := strings.TrimSpace(string(r.out))
|
||||
case err := <-done:
|
||||
text := collected()
|
||||
// A non-zero exit MUST surface as an error. The previous guard only
|
||||
// errored when there was no output, so a `pct create` that printed
|
||||
// "CT 132 already exists" and exited non-zero was reported as
|
||||
// success — the execution was marked completed though nothing was
|
||||
// provisioned.
|
||||
if r.err != nil {
|
||||
if err != nil {
|
||||
if text != "" {
|
||||
return text, fmt.Errorf("%w: %s", r.err, text)
|
||||
return text, fmt.Errorf("%w: %s", err, text)
|
||||
}
|
||||
return text, fmt.Errorf("exec: %w", r.err)
|
||||
return text, fmt.Errorf("exec: %w", err)
|
||||
}
|
||||
return text, nil
|
||||
case <-time.After(sshExecTimeout):
|
||||
// Close the session/client to hang up the remote side; the
|
||||
// goroutine above will eventually exit once that unblocks
|
||||
// CombinedOutput, but we don't wait for it — the caller needs an
|
||||
// answer now, not an indefinite hang.
|
||||
// goroutine above will eventually exit once that unblocks Run, but we
|
||||
// don't wait for it — the caller needs an answer now, not an
|
||||
// indefinite hang.
|
||||
session.Close()
|
||||
client.Close()
|
||||
return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
|
||||
// Return what arrived before it hung, rather than "". A provisioning
|
||||
// command that stalls halfway is precisely when its output matters.
|
||||
return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
|
||||
case <-ctx.Done():
|
||||
session.Close()
|
||||
client.Close()
|
||||
return "", ctx.Err()
|
||||
return collected(), ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -231,7 +274,16 @@ func emitExecutionEvent(ctx context.Context, pool *db.Pool, execID uuid.UUID, st
|
||||
if status == "failed" {
|
||||
severity = "warning"
|
||||
}
|
||||
_ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", "", detail)
|
||||
// The correlation id was hardcoded to "", so execution events could not be
|
||||
// tied back to the session that caused them — the one join you want when
|
||||
// asking "what did this agent turn actually do?". It is already on the
|
||||
// execution row; read it rather than threading it through eleven callers.
|
||||
var correlationID string
|
||||
if err := pool.QueryRow(ctx,
|
||||
`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); err != nil {
|
||||
correlationID = ""
|
||||
}
|
||||
_ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", correlationID, detail)
|
||||
if status == "completed" || status == "failed" || status == "cancelled" {
|
||||
closePlanStepForExecution(ctx, pool, execID, status)
|
||||
}
|
||||
@@ -280,6 +332,29 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID,
|
||||
action, params := actionStr[:idx], actionStr[idx+1:]
|
||||
|
||||
startedAt := time.Now()
|
||||
// Persist started_at now, not at the end. It was captured here but only
|
||||
// written in the terminal UPDATE, so a running execution reported
|
||||
// started_at = NULL for its entire life — the UI could not show how long
|
||||
// anything had been going, which is exactly when you want to know.
|
||||
if _, err := pool.Exec(ctx,
|
||||
`UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`,
|
||||
execID, startedAt); err != nil {
|
||||
slog.Error("httpapi: mark execution running", "error", err, "execution_id", execID)
|
||||
}
|
||||
|
||||
// Stream output for the actions whose output an operator actually watches:
|
||||
// a long apt upgrade, a pct create, an arbitrary approved `run`. The small
|
||||
// internal lookups further down (listing template cache, pvesh nextid) stay
|
||||
// unstreamed — they are plumbing, and logging them would bury the command
|
||||
// the operator approved.
|
||||
var correlationID string
|
||||
if qerr := pool.QueryRow(ctx,
|
||||
`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); qerr != nil {
|
||||
correlationID = ""
|
||||
}
|
||||
sink, flushLogs := execlog.New(ctx, pool, execID, correlationID)
|
||||
defer flushLogs()
|
||||
|
||||
var output, cmd string
|
||||
|
||||
switch action {
|
||||
@@ -295,30 +370,30 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID,
|
||||
default:
|
||||
cmd = fmt.Sprintf("systemctl %s %s 2>&1", params, svc)
|
||||
}
|
||||
output, err = sshExec(ctx, host, user, cmd)
|
||||
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
||||
|
||||
case "apt_upgrade":
|
||||
svc := strings.TrimPrefix(targetSlug, "lxc:")
|
||||
cmd = fmt.Sprintf("apt update -qq 2>&1 >/dev/null && apt upgrade -y -qq 2>&1; echo '---'; systemctl is-active %s || true", svc)
|
||||
output, err = sshExec(ctx, host, user, cmd)
|
||||
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
||||
|
||||
case "pct_create":
|
||||
var cfg struct {
|
||||
VMID int `json:"vmid"`
|
||||
Hostname string `json:"hostname"`
|
||||
Cores int `json:"cores"`
|
||||
Memory int `json:"memory"`
|
||||
DiskGB int `json:"disk_gb"`
|
||||
IP string `json:"ip"`
|
||||
GW string `json:"gw"`
|
||||
Bridge string `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template"`
|
||||
Privileged flexBool `json:"privileged"`
|
||||
Nesting flexBool `json:"nesting"`
|
||||
Mounts []string `json:"mounts"`
|
||||
Nameserver string `json:"nameserver"`
|
||||
Searchdomain string `json:"searchdomain"`
|
||||
VMID int `json:"vmid"`
|
||||
Hostname string `json:"hostname"`
|
||||
Cores int `json:"cores"`
|
||||
Memory int `json:"memory"`
|
||||
DiskGB int `json:"disk_gb"`
|
||||
IP string `json:"ip"`
|
||||
GW string `json:"gw"`
|
||||
Bridge string `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0
|
||||
Storage string `json:"storage"`
|
||||
Template string `json:"template"`
|
||||
Privileged flexBool `json:"privileged"`
|
||||
Nesting flexBool `json:"nesting"`
|
||||
Mounts []string `json:"mounts"`
|
||||
Nameserver string `json:"nameserver"`
|
||||
Searchdomain string `json:"searchdomain"`
|
||||
// No services/post_install here anymore — pct_create is atomic
|
||||
// (create + start + register only). Installing packages and
|
||||
// running setup scripts is the agent's job via follow-up `run`
|
||||
@@ -504,7 +579,7 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID,
|
||||
}
|
||||
|
||||
slog.Info("httpapi: pct_create running", "vmid", cfg.VMID, "hostname", cfg.Hostname, "cmd", createCmd)
|
||||
output, err = sshExec(ctx, host, user, createCmd)
|
||||
output, err = sshExecStream(ctx, host, user, createCmd, sink)
|
||||
|
||||
// pct_create is now DELIBERATELY ATOMIC: create + start + register,
|
||||
// nothing else. It used to also run apt installs and a post_install
|
||||
@@ -579,7 +654,7 @@ func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID,
|
||||
return
|
||||
}
|
||||
cmd = wrap(cfg.Command)
|
||||
output, err = sshExec(ctx, host, user, cmd)
|
||||
output, err = sshExecStream(ctx, host, user, cmd, sink)
|
||||
|
||||
default:
|
||||
slog.Error("httpapi: unknown gated action for approved execution", "action", action, "execution_id", execID)
|
||||
|
||||
Reference in New Issue
Block a user