diff --git a/eventstream/store.go b/eventstream/store.go index 176527d9..035ff067 100644 --- a/eventstream/store.go +++ b/eventstream/store.go @@ -48,10 +48,16 @@ func HighWater(s Store) (int64, error) { return lastNanos, nil } -func Insert(s Store, ev Event, n *notifier.Notifier) error { +type Inserter func(Store, Event) error + +// callers open their transaction inside this so every event writer takes the locks in the same order +func WithClock(fn func(Inserter) error) error { clockMu.Lock() defer clockMu.Unlock() + return fn(insertLocked) +} +func insertLocked(s Store, ev Event) error { if ev.Created == 0 { now := time.Now().UnixNano() if now <= lastNanos { @@ -63,17 +69,26 @@ func Insert(s Store, ev Event, n *notifier.Notifier) error { lastNanos = ev.Created } - if _, err := s.Exec( + _, err := s.Exec( `insert into events (rkey, nsid, event, created) values (?, ?, ?, ?)`, ev.Rkey, ev.Nsid, []byte(ev.EventJson), ev.Created, - ); err != nil { - return err - } - n.NotifyAll() - return nil + ) + return err +} + +func Insert(s Store, ev Event, n *notifier.Notifier) error { + return WithClock(func(insert Inserter) error { + if err := insert(s, ev); err != nil { + return err + } + if n != nil { + n.NotifyAll() + } + return nil + }) } func List(s Store, cursor int64, limit int) ([]Event, error) { diff --git a/spindle/db/db.go b/spindle/db/db.go index 82b09950..1d21279c 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -115,7 +115,8 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { tpl text not null, traceparent text not null default '', tracestate text not null default '', - created_at integer not null default (strftime('%s', 'now')) + created_at integer not null default (strftime('%s', 'now')), + created_at_ns integer not null default 0 ); create table if not exists workflows ( @@ -154,7 +155,8 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { state text not null, quota_reservation_id text, owner_did text, - repo_did text + repo_did text, + mill_records_terminal_metrics integer not null default 0 ); create table if not exists mill_executor_cursors ( @@ -196,7 +198,10 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { error text not null default '', exit_code integer not null default 0, ref text not null, - hash text not null + hash text not null, + failure_class text not null default '', + failure_reason text not null default '', + mill_records_terminal_metrics integer not null default 0 ); create table if not exists migrations ( @@ -626,6 +631,107 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "jobs-created-at", func(tx *sql.Tx) error { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('jobs') where name = 'created_at'`, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + return nil + } + _, err := tx.Exec(`alter table jobs add column created_at integer not null default 0`) + return err + }); err != nil { + return err + } + + if err := orm.RunMigration(conn, logger, "jobs-created-at-ns", func(tx *sql.Tx) error { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('jobs') where name = 'created_at_ns'`, + ).Scan(&present); err != nil { + return err + } + if present == 0 { + if _, err := tx.Exec(`alter table jobs add column created_at_ns integer not null default 0`); err != nil { + return err + } + } + _, err := tx.Exec(` + update jobs + set created_at_ns = created_at * 1000000000 + where created_at_ns = 0 and created_at between 1 and 9999999999 + `) + return err + }); err != nil { + return err + } + + if err := orm.RunMigration(conn, logger, "pending-artifact-failure-attribution", func(tx *sql.Tx) error { + for _, column := range []string{"failure_class", "failure_reason"} { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('executor_pending_artifacts') where name = ?`, + column, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + continue + } + if _, err := tx.Exec( + `alter table executor_pending_artifacts add column ` + column + ` text not null default ''`, + ); err != nil { + return err + } + } + return nil + }); err != nil { + return err + } + + if err := orm.RunMigration(conn, logger, "pending-artifact-terminal-metric-authority", func(tx *sql.Tx) error { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('executor_pending_artifacts') + where name = 'mill_records_terminal_metrics'`, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + return nil + } + _, err := tx.Exec(` + alter table executor_pending_artifacts + add column mill_records_terminal_metrics integer not null default 0 + `) + return err + }); err != nil { + return err + } + + if err := orm.RunMigration(conn, logger, "mill-lease-terminal-metric-authority", func(tx *sql.Tx) error { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('mill_leases') + where name = 'mill_records_terminal_metrics'`, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + return nil + } + _, err := tx.Exec(` + alter table mill_leases + add column mill_records_terminal_metrics integer not null default 0 + `) + return err + }); err != nil { + return err + } + return nil } diff --git a/spindle/db/events.go b/spindle/db/events.go index cbd8c891..2166c8cf 100644 --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -1,6 +1,8 @@ package db import ( + "context" + "database/sql" "encoding/json" "fmt" "tangled.org/core/api/tangled" @@ -142,6 +144,56 @@ func (d *DB) GetStatus(workflowId models.WorkflowId) (*tangled.PipelineStatus, e return &status, nil } +type statusQueryer interface { + QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row +} + +func workflowStartupDelay(ctx context.Context, q statusQueryer, pipelineAtURI, workflow string) (time.Duration, bool, error) { + var pending, running sql.NullInt64 + err := q.QueryRowContext(ctx, ` + select + min(case when json_extract(event, '$.status') = 'pending' then created end), + min(case when json_extract(event, '$.status') = 'running' then created end) + from events + where nsid = 'sh.tangled.pipeline.status' + and json_extract(event, '$.pipeline') = ? + and json_extract(event, '$.workflow') = ? + `, pipelineAtURI, workflow).Scan(&pending, &running) + if err != nil { + return 0, false, err + } + if !pending.Valid || !running.Valid { + return 0, false, nil + } + delay := time.Duration(running.Int64 - pending.Int64) + if delay < 0 { + delay = 0 + } + return delay, true, nil +} + +func (d *DB) WorkflowStartupDelay(ctx context.Context, workflowID models.WorkflowId) (time.Duration, bool, error) { + return workflowStartupDelay(ctx, d, string(workflowID.PipelineId.AtUri()), workflowID.Name) +} + +func (tx *EventBatchTx) WorkflowStartupDelay(ctx context.Context, pipelineAtURI, workflow string) (time.Duration, bool, error) { + return workflowStartupDelay(ctx, tx.tx, pipelineAtURI, workflow) +} + +func (tx *EventBatchTx) HasWorkflowStatus(ctx context.Context, pipelineAtURI, workflow, status string) (bool, error) { + var present bool + err := tx.tx.QueryRowContext(ctx, ` + select exists( + select 1 from events + where nsid = 'sh.tangled.pipeline.status' + and json_extract(event, '$.pipeline') = ? + and json_extract(event, '$.workflow') = ? + and json_extract(event, '$.status') = ? + ) + `, pipelineAtURI, workflow, status).Scan(&present) + return present, err +} + func (d *DB) StatusPending(workflowId models.WorkflowId, n *notifier.Notifier) error { return d.createStatusEvent(workflowId, models.StatusKindPending, nil, nil, n) } diff --git a/spindle/db/events_metrics_test.go b/spindle/db/events_metrics_test.go new file mode 100644 index 00000000..4df51445 --- /dev/null +++ b/spindle/db/events_metrics_test.go @@ -0,0 +1,88 @@ +package db + +import ( + "context" + "path/filepath" + "strings" + "testing" + "time" + + "tangled.org/core/notifier" + "tangled.org/core/spindle/models" +) + +func TestWorkflowStartupDelay(t *testing.T) { + database, err := Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + + n := notifier.New() + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.example.com", Rkey: "pipeline"}, + Name: "build", + } + if err := database.StatusPending(wid, &n); err != nil { + t.Fatal(err) + } + pendingAt := time.Now().Add(-5 * time.Second).UnixNano() + if _, err := database.Exec(`update events set created = ? where json_extract(event, '$.status') = 'pending'`, pendingAt); err != nil { + t.Fatal(err) + } + if err := database.StatusRunning(wid, &n); err != nil { + t.Fatal(err) + } + + delay, ok, err := database.WorkflowStartupDelay(context.Background(), wid) + if err != nil { + t.Fatal(err) + } + if !ok { + t.Fatal("startup delay missing") + } + if delay < 4*time.Second || delay > 6*time.Second { + t.Fatalf("startup delay = %v, want about 5s", delay) + } +} + +func TestWorkflowStartupDelayUsesStatusIndex(t *testing.T) { + database, err := Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + + rows, err := database.Query(` + explain query plan + select + min(case when json_extract(event, '$.status') = 'pending' then created end), + min(case when json_extract(event, '$.status') = 'running' then created end) + from events + where nsid = 'sh.tangled.pipeline.status' + and json_extract(event, '$.pipeline') = ? + and json_extract(event, '$.workflow') = ? + `, "at://knot.example.com/sh.tangled.pipeline/pipeline", "build") + if err != nil { + t.Fatal(err) + } + defer rows.Close() + + usedIndex := false + for rows.Next() { + var id, parent, unused int + var detail string + if err := rows.Scan(&id, &parent, &unused, &detail); err != nil { + t.Fatal(err) + } + if strings.Contains(detail, "idx_events_pipeline_status") { + usedIndex = true + } + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + if !usedIndex { + t.Fatal("workflow startup query did not use idx_events_pipeline_status") + } +} diff --git a/spindle/db/jobs.go b/spindle/db/jobs.go index 9951d1f2..fa54132f 100644 --- a/spindle/db/jobs.go +++ b/spindle/db/jobs.go @@ -5,8 +5,11 @@ import ( "database/sql" "encoding/json" "errors" + "time" "tangled.org/core/api/tangled" + "tangled.org/core/eventstream" + "tangled.org/core/notifier" "tangled.org/core/spindle/models" ) @@ -19,46 +22,124 @@ type JobRow struct { Tpl tangled.Pipeline Traceparent string Tracestate string + CreatedAtNs int64 } // the atomic check rejects a banned job before admission var ErrBannedSubject = errors.New("subject is banned") -func (d *DB) EnqueueJob(ctx context.Context, repoDid string, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo, tpl tangled.Pipeline, traceparent, tracestate string) error { - tplJson, err := json.Marshal(tpl) - if err != nil { - return err - } - tx, err := d.BeginTx(ctx, nil) +func (d *DB) EnqueueJob( + ctx context.Context, + repoDid string, + pipelineID models.PipelineId, + sourceRepo *tangled.Pipeline_TriggerRepo, + tpl tangled.Pipeline, + traceparent, tracestate string, +) error { + return d.enqueueJob(ctx, repoDid, pipelineID, sourceRepo, tpl, traceparent, tracestate, nil, false) +} + +func (d *DB) EnqueueJobWithPending( + ctx context.Context, + repoDid string, + pipelineID models.PipelineId, + sourceRepo *tangled.Pipeline_TriggerRepo, + tpl tangled.Pipeline, + traceparent, tracestate string, + n *notifier.Notifier, +) error { + return d.enqueueJob(ctx, repoDid, pipelineID, sourceRepo, tpl, traceparent, tracestate, n, true) +} + +func (d *DB) enqueueJob( + ctx context.Context, + repoDid string, + pipelineID models.PipelineId, + sourceRepo *tangled.Pipeline_TriggerRepo, + tpl tangled.Pipeline, + traceparent, tracestate string, + n *notifier.Notifier, + markPending bool, +) error { + tplJSON, err := json.Marshal(tpl) if err != nil { return err } - defer tx.Rollback() - var owner sql.NullString - if err := tx.QueryRowContext(ctx, `select owner from repos where repo_did = ?`, repoDid).Scan(&owner); err != nil && !errors.Is(err, sql.ErrNoRows) { - return err - } - var banned bool - if err := tx.QueryRowContext(ctx, ` - select exists( + enqueue := func(insert eventstream.Inserter) error { + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + var owner sql.NullString + if err := tx.QueryRowContext(ctx, `select owner from repos where repo_did = ?`, repoDid).Scan(&owner); err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + var banned bool + if err := tx.QueryRowContext(ctx, ` + select exists( select 1 from bans where subject_did in (?, ?) - )`, repoDid, owner.String).Scan(&banned); err != nil { - return err - } - if banned { - return ErrBannedSubject + )`, repoDid, owner.String).Scan(&banned); err != nil { + return err + } + if banned { + return ErrBannedSubject + } + + createdAtNS := time.Now().UnixNano() + var sourceRepoValue any + if sourceRepoJSON := sourceRepoJson(sourceRepo); sourceRepoJSON != nil { + sourceRepoValue = string(sourceRepoJSON) + } + if _, err = tx.ExecContext(ctx, ` + insert into jobs ( + repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl, + traceparent, tracestate, created_at, created_at_ns + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?) + `, repoDid, pipelineID.Knot, pipelineID.Rkey, sourceRepoValue, string(tplJSON), + traceparent, tracestate, time.Unix(0, createdAtNS).Unix(), createdAtNS, + ); err != nil { + return err + } + + if markPending { + for _, workflow := range tpl.Workflows { + if workflow == nil { + continue + } + event, err := statusEvent( + string(pipelineID.AtUri()), + workflow.Name, + string(models.StatusKindPending), + nil, + nil, + ) + if err != nil { + return err + } + if err := insert(tx, event); err != nil { + return err + } + } + } + return tx.Commit() } - _, err = tx.ExecContext(ctx, ` - insert into jobs (repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl, traceparent, tracestate) - values (?, ?, ?, ?, ?, ?, ?) - `, repoDid, pipelineId.Knot, pipelineId.Rkey, string(sourceRepoJson(sourceRepo)), string(tplJson), traceparent, tracestate) + if markPending { + err = eventstream.WithClock(enqueue) + } else { + err = enqueue(nil) + } if err != nil { return err } - return tx.Commit() + if markPending && n != nil { + n.NotifyAll() + } + return nil } func (d *DB) DequeueJob(ctx context.Context) (*JobRow, error) { var row JobRow @@ -71,8 +152,8 @@ func (d *DB) DequeueJob(ctx context.Context) (*JobRow, error) { order by id asc limit 1 ) - returning id, repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl, traceparent, tracestate - `).Scan(&row.Id, &row.RepoDid, &row.PipelineIdKnot, &row.PipelineIdRkey, &sourceRepoStr, &tplJson, &row.Traceparent, &row.Tracestate) + returning id, repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl, traceparent, tracestate, created_at_ns + `).Scan(&row.Id, &row.RepoDid, &row.PipelineIdKnot, &row.PipelineIdRkey, &sourceRepoStr, &tplJson, &row.Traceparent, &row.Tracestate, &row.CreatedAtNs) if err != nil { if err == sql.ErrNoRows { return nil, nil diff --git a/spindle/db/jobs_test.go b/spindle/db/jobs_test.go index 94a6dced..a0775dda 100644 --- a/spindle/db/jobs_test.go +++ b/spindle/db/jobs_test.go @@ -2,10 +2,13 @@ package db import ( "context" + "database/sql" "path/filepath" "testing" + "time" "tangled.org/core/api/tangled" + "tangled.org/core/notifier" "tangled.org/core/spindle/models" ) @@ -42,4 +45,95 @@ func TestJobTraceContextSurvivesEnqueue(t *testing.T) { if job.Traceparent != traceparent || job.Tracestate != tracestate { t.Fatalf("trace context = (%q, %q), want (%q, %q)", job.Traceparent, job.Tracestate, traceparent, tracestate) } + if created := time.Unix(0, job.CreatedAtNs); created.Before(time.Now().Add(-5*time.Second)) || created.After(time.Now().Add(time.Second)) { + t.Fatalf("created_at_ns = %v, want current enqueue time", created) + } +} + +func TestEnqueueJobWithPendingCommitsJobAndStatusesTogether(t *testing.T) { + ctx := context.Background() + database, err := Make(ctx, filepath.Join(t.TempDir(), "spindle.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + + n := notifier.New() + pipelineID := models.PipelineId{Knot: "knot.example.com", Rkey: "pipeline"} + tpl := tangled.Pipeline{Workflows: []*tangled.Pipeline_Workflow{{Name: "build"}}} + if err := database.EnqueueJobWithPending( + ctx, + "did:web:repo.example.com", + pipelineID, + nil, + tpl, + "", + "", + &n, + ); err != nil { + t.Fatal(err) + } + + status, err := database.GetStatus(models.WorkflowId{PipelineId: pipelineID, Name: "build"}) + if err != nil { + t.Fatal(err) + } + if status.Status != string(models.StatusKindPending) { + t.Fatalf("status = %q, want pending", status.Status) + } + job, err := database.DequeueJob(ctx) + if err != nil { + t.Fatal(err) + } + if job == nil || job.CreatedAtNs == 0 { + t.Fatalf("dequeued job = %+v, want nanosecond admission timestamp", job) + } +} + +func TestMakeMigratesLegacyJobTimestampToNanoseconds(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "spindle.db") + raw, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatal(err) + } + _, err = raw.Exec(` + create table jobs ( + id integer primary key autoincrement, + repo_did text not null, + pipeline_id_knot text not null, + pipeline_id_rkey text not null, + source_repo text, + tpl text not null, + traceparent text not null default '', + tracestate text not null default '', + created_at integer not null + ); + create table migrations ( + id integer primary key autoincrement, + name text unique + ); + insert into jobs ( + repo_did, pipeline_id_knot, pipeline_id_rkey, tpl, created_at + ) values ('did:web:repo.example.com', 'knot.example.com', 'pipeline', '{}', 1700000000); + `) + if err != nil { + t.Fatal(err) + } + if err := raw.Close(); err != nil { + t.Fatal(err) + } + + database, err := Make(ctx, path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + job, err := database.DequeueJob(ctx) + if err != nil { + t.Fatal(err) + } + if job == nil || job.CreatedAtNs != 1700000000*int64(time.Second) { + t.Fatalf("created_at_ns = %v, want converted legacy seconds", job) + } } diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 8d04cfdc..26c67ed0 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -10,17 +10,18 @@ import ( // recovery uses persisted identity, fencing and quota state type MillLease struct { - LeaseID string - NodeID string - Epoch string - Engine string - Knot string - Rkey string - Workflow string - State string - QuotaReservationID string - OwnerDID string - RepoDID string + LeaseID string + NodeID string + Epoch string + Engine string + Knot string + Rkey string + Workflow string + State string + QuotaReservationID string + OwnerDID string + RepoDID string + MillRecordsTerminalMetrics bool } type ExecutorCursor struct { @@ -45,15 +46,16 @@ func (d *DB) SaveMillLease(l MillLease) error { _, err := d.Exec( `insert into mill_leases ( lease_id, node_id, epoch, engine, knot, rkey, workflow, state, - quota_reservation_id, owner_did, repo_did - ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + quota_reservation_id, owner_did, repo_did, mill_records_terminal_metrics + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict(lease_id) do update set state = excluded.state, quota_reservation_id = excluded.quota_reservation_id, owner_did = excluded.owner_did, - repo_did = excluded.repo_did`, + repo_did = excluded.repo_did, + mill_records_terminal_metrics = excluded.mill_records_terminal_metrics`, l.LeaseID, l.NodeID, l.Epoch, l.Engine, l.Knot, l.Rkey, l.Workflow, l.State, - l.QuotaReservationID, l.OwnerDID, l.RepoDID, + l.QuotaReservationID, l.OwnerDID, l.RepoDID, l.MillRecordsTerminalMetrics, ) return err } @@ -66,7 +68,8 @@ func (d *DB) DeleteMillLease(leaseID string) error { func (d *DB) ListMillLeases() ([]MillLease, error) { rows, err := d.Query(` select lease_id, node_id, epoch, engine, knot, rkey, workflow, state, - coalesce(quota_reservation_id, ''), coalesce(owner_did, ''), coalesce(repo_did, '') + coalesce(quota_reservation_id, ''), coalesce(owner_did, ''), coalesce(repo_did, ''), + mill_records_terminal_metrics from mill_leases `) if err != nil { @@ -79,7 +82,7 @@ func (d *DB) ListMillLeases() ([]MillLease, error) { var l MillLease if err := rows.Scan( &l.LeaseID, &l.NodeID, &l.Epoch, &l.Engine, &l.Knot, &l.Rkey, &l.Workflow, &l.State, - &l.QuotaReservationID, &l.OwnerDID, &l.RepoDID, + &l.QuotaReservationID, &l.OwnerDID, &l.RepoDID, &l.MillRecordsTerminalMetrics, ); err != nil { return nil, err } @@ -259,8 +262,9 @@ func (d *DB) GetOutboxState() (string, uint64, error) { } type EventBatchTx struct { - tx *sql.Tx - db *DB + tx *sql.Tx + db *DB + insert eventstream.Inserter } func (tx *EventBatchTx) InsertStatusEvent(pipelineAtUri, workflow, status string, workflowError *string, exitCode *int64) error { @@ -268,7 +272,7 @@ func (tx *EventBatchTx) InsertStatusEvent(pipelineAtUri, workflow, status string if err != nil { return err } - return eventstream.Insert(tx.tx, event, nil) + return tx.insert(tx.tx, event) } func (tx *EventBatchTx) DeleteLease(leaseID string) error { @@ -335,27 +339,41 @@ func (tx *EventBatchTx) InsertArtifactRef(leaseID, repoDid, workflow, ref, hash } type PendingArtifact struct { - LeaseID string - Workflow string - Status string - Error string - ExitCode int64 - Ref string - Hash string + LeaseID string + Workflow string + Status string + Error string + ExitCode int64 + Ref string + Hash string + FailureClass string + FailureReason string + MillRecordsTerminalMetrics bool } -func (d *DB) SavePendingArtifact(leaseID, workflow, status, errStr string, exitCode int64, ref, hash string) error { +func (d *DB) SavePendingArtifact( + leaseID, workflow, status, errStr string, + exitCode int64, + ref, hash, failureClass, failureReason string, + millRecordsTerminalMetrics bool, +) error { _, err := d.Exec( - `insert into executor_pending_artifacts (lease_id, workflow, status, error, exit_code, ref, hash) - values (?, ?, ?, ?, ?, ?, ?) + `insert into executor_pending_artifacts ( + lease_id, workflow, status, error, exit_code, ref, hash, + failure_class, failure_reason, mill_records_terminal_metrics + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict(lease_id) do update set workflow = excluded.workflow, status = excluded.status, error = excluded.error, exit_code = excluded.exit_code, ref = excluded.ref, - hash = excluded.hash`, + hash = excluded.hash, + failure_class = excluded.failure_class, + failure_reason = excluded.failure_reason, + mill_records_terminal_metrics = excluded.mill_records_terminal_metrics`, leaseID, workflow, status, errStr, exitCode, ref, hash, + failureClass, failureReason, millRecordsTerminalMetrics, ) return err } @@ -366,7 +384,11 @@ func (d *DB) RemovePendingArtifact(leaseID string) error { } func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { - rows, err := d.Query(`select lease_id, workflow, status, error, exit_code, ref, hash from executor_pending_artifacts`) + rows, err := d.Query(` + select lease_id, workflow, status, error, exit_code, ref, hash, + failure_class, failure_reason, mill_records_terminal_metrics + from executor_pending_artifacts + `) if err != nil { return nil, err } @@ -375,7 +397,10 @@ func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { var res []PendingArtifact for rows.Next() { var p PendingArtifact - if err := rows.Scan(&p.LeaseID, &p.Workflow, &p.Status, &p.Error, &p.ExitCode, &p.Ref, &p.Hash); err != nil { + if err := rows.Scan( + &p.LeaseID, &p.Workflow, &p.Status, &p.Error, &p.ExitCode, &p.Ref, &p.Hash, + &p.FailureClass, &p.FailureReason, &p.MillRecordsTerminalMetrics, + ); err != nil { return nil, err } res = append(res, p) @@ -384,25 +409,26 @@ func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { } func (d *DB) ApplyEventBatch(n *notifier.Notifier, fn func(tx *EventBatchTx) error) error { - tx, err := d.Begin() - if err != nil { - return err - } - defer tx.Rollback() - - batchTx := &EventBatchTx{ - tx: tx, - db: d, - } - - if err := fn(batchTx); err != nil { - return err - } + err := eventstream.WithClock(func(insert eventstream.Inserter) error { + tx, err := d.Begin() + if err != nil { + return err + } + defer tx.Rollback() - if err := tx.Commit(); err != nil { + batchTx := &EventBatchTx{ + tx: tx, + db: d, + insert: insert, + } + if err := fn(batchTx); err != nil { + return err + } + return tx.Commit() + }) + if err != nil { return err } - if n != nil { n.NotifyAll() } diff --git a/spindle/db/mill_state_test.go b/spindle/db/mill_state_test.go index 9902c78c..d11339f6 100644 --- a/spindle/db/mill_state_test.go +++ b/spindle/db/mill_state_test.go @@ -14,14 +14,15 @@ func TestMillLeaseRoundTrip(t *testing.T) { d := newTestDB(t) lease := MillLease{ - LeaseID: "lease-1", - NodeID: "node-1", - Epoch: "inc-1", - Engine: "dummy", - Knot: "knot.example", - Rkey: "rkey1", - Workflow: "build", - State: "reserved", + LeaseID: "lease-1", + NodeID: "node-1", + Epoch: "inc-1", + Engine: "dummy", + Knot: "knot.example", + Rkey: "rkey1", + Workflow: "build", + State: "reserved", + MillRecordsTerminalMetrics: true, } if err := d.SaveMillLease(lease); err != nil { t.Fatalf("SaveMillLease: %v", err) @@ -39,7 +40,9 @@ func TestMillLeaseRoundTrip(t *testing.T) { if len(leases) != 1 { t.Fatalf("ListMillLeases returned %d leases, want 1 (state transition must replace, not duplicate)", len(leases)) } - if leases[0].LeaseID != lease.LeaseID || leases[0].State != "running" { + if leases[0].LeaseID != lease.LeaseID || + leases[0].State != "running" || + !leases[0].MillRecordsTerminalMetrics { t.Fatalf("ListMillLeases[0] = %+v, want %+v", leases[0], lease) } @@ -488,9 +491,16 @@ func TestExecutorCursorResetHelpers(t *testing.T) { func TestClearPendingArtifacts(t *testing.T) { d := newTestDB(t) - if err := d.SavePendingArtifact("lease-1", "build", "success", "", 0, "ref", "sha256:x"); err != nil { + if err := d.SavePendingArtifact("lease-1", "build", "success", "", 0, "ref", "sha256:x", "none", "success", true); err != nil { t.Fatalf("SavePendingArtifact: %v", err) } + rows, err := d.ListPendingArtifacts() + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].FailureClass != "none" || rows[0].FailureReason != "success" || !rows[0].MillRecordsTerminalMetrics { + t.Fatalf("pending artifact attribution = %+v", rows) + } if err := d.ClearPendingArtifacts(); err != nil { t.Fatalf("ClearPendingArtifacts: %v", err) } diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go index 035201b0..fbd2a408 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -32,6 +32,80 @@ var ( ErrWorkflowCanceled = errors.New("workflow canceled") ) +type FailureClass string + +const ( + FailureClassNone FailureClass = "none" + FailureClassUser FailureClass = "user" + FailureClassInfrastructure FailureClass = "infrastructure" + FailureClassPolicy FailureClass = "policy" +) + +type FailureReason string + +const ( + FailureReasonSuccess FailureReason = "success" + FailureReasonCommandFailed FailureReason = "command_failed" + FailureReasonResourceExhausted FailureReason = "resource_exhausted" + FailureReasonConfigurationFailed FailureReason = "configuration_failed" + FailureReasonWorkflowInvalid FailureReason = "workflow_invalid" + FailureReasonSetupFailed FailureReason = "setup_failed" + FailureReasonRuntimeFailed FailureReason = "runtime_failed" + FailureReasonCapacityUnavailable FailureReason = "capacity_unavailable" + FailureReasonQuotaDenied FailureReason = "quota_denied" + FailureReasonExecutorLost FailureReason = "executor_lost" + FailureReasonTimeout FailureReason = "timeout" + FailureReasonCancelled FailureReason = "cancelled" +) + +type WorkflowFailure struct { + Class FailureClass + Reason FailureReason + Err error +} + +func (e *WorkflowFailure) Error() string { + if e.Err == nil { + return string(e.Reason) + } + return e.Err.Error() +} + +func (e *WorkflowFailure) Unwrap() error { + return e.Err +} + +func ClassifiedFailure(class FailureClass, reason FailureReason, err error) error { + if err == nil { + return nil + } + var classified *WorkflowFailure + if errors.As(err, &classified) { + return err + } + return &WorkflowFailure{Class: class, Reason: reason, Err: err} +} + +func FailureAttribution(result string, err error) (string, string) { + switch result { + case "success": + return string(FailureClassNone), string(FailureReasonSuccess) + case "timeout": + return string(FailureClassUser), string(FailureReasonTimeout) + case "cancelled": + return string(FailureClassUser), string(FailureReasonCancelled) + } + + var failure *WorkflowFailure + if errors.As(err, &failure) { + return string(failure.Class), string(failure.Reason) + } + if errors.Is(err, ErrNoWorkflowSlots) { + return string(FailureClassInfrastructure), string(FailureReasonCapacityUnavailable) + } + return string(FailureClassInfrastructure), string(FailureReasonRuntimeFailed) +} + var ( activeMu sync.Mutex activeCancels = make(map[models.WorkflowId]context.CancelCauseFunc) @@ -197,6 +271,12 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q if dbErr != nil { l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) } + observability.GetMetrics(ctx).RecordWorkflowTerminal( + engineName(eng), + "failure", + string(FailureClassUser), + string(FailureReasonWorkflowInvalid), + ) continue } @@ -263,6 +343,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q } startTime := time.Now() result := "success" + remoteExecutionStarted := false defer func() { if err != nil { @@ -334,11 +415,19 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q ) } + failureClass, failureReason := FailureAttribution(resStr, err) if remoteStatus { + if !remoteExecutionStarted { + metrics.RecordWorkflowTerminal(engName, resStr, failureClass, failureReason) + } return } - metrics.RecordWorkflowEnd(wfCtx, engName, resStr, duration) + if observability.NotifyWorkflowTerminal(wfCtx, engName, resStr, failureClass, failureReason) { + metrics.RecordWorkflowExecutionEnd(wfCtx, engName, resStr, duration) + } else { + metrics.RecordWorkflowEnd(wfCtx, engName, resStr, failureClass, failureReason, duration) + } }() var wfLogger models.WorkflowLogger @@ -422,6 +511,9 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q } quotaLease, err = qm.Acquire(wfCtx, req) if err != nil { + if errors.Is(err, quota.ErrDenied) { + err = ClassifiedFailure(FailureClassPolicy, FailureReasonQuotaDenied, err) + } setTerminalError("acquiring quota", err) return } @@ -436,6 +528,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q return } slotAcquired = true + remoteExecutionStarted = true } if !remoteStatus { @@ -444,6 +537,11 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q wl.ErrorContext(wfCtx, "failed to set workflow status to running", "wid", wid, "err", err) return } + if delay, ok, delayErr := db.WorkflowStartupDelay(wfCtx, wid); delayErr != nil { + wl.WarnContext(wfCtx, "failed to measure workflow startup delay", "err", delayErr) + } else if ok { + metrics.RecordWorkflowStartupDelay(wfCtx, engName, "local", delay) + } } wl.InfoContext(wfCtx, "workflow started", @@ -457,6 +555,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q err = eng.SetupWorkflow(wfCtx, wid, &w, wfLogger) if err != nil { + err = ClassifiedFailure(FailureClassInfrastructure, FailureReasonSetupFailed, err) destroyWorkflow = !isCanceled(wfCtx) if !remoteStatus { setTerminalError("setting up workflow", err) diff --git a/spindle/engine/failure_metrics_test.go b/spindle/engine/failure_metrics_test.go new file mode 100644 index 00000000..461dd7e2 --- /dev/null +++ b/spindle/engine/failure_metrics_test.go @@ -0,0 +1,65 @@ +package engine + +import ( + "errors" + "testing" +) + +func TestFailureAttribution(t *testing.T) { + base := errors.New("detail") + tests := []struct { + name string + result string + err error + wantClass string + wantReason string + }{ + {name: "success", result: "success", wantClass: "none", wantReason: "success"}, + {name: "timeout", result: "timeout", err: ErrTimedOut, wantClass: "user", wantReason: "timeout"}, + {name: "cancelled", result: "cancelled", err: ErrWorkflowCanceled, wantClass: "user", wantReason: "cancelled"}, + { + name: "user command", + result: "failure", + err: ClassifiedFailure(FailureClassUser, FailureReasonCommandFailed, base), + wantClass: "user", + wantReason: "command_failed", + }, + { + name: "infrastructure", + result: "failure", + err: ClassifiedFailure(FailureClassInfrastructure, FailureReasonRuntimeFailed, base), + wantClass: "infrastructure", + wantReason: "runtime_failed", + }, + { + name: "policy", + result: "failure", + err: ClassifiedFailure(FailureClassPolicy, FailureReasonQuotaDenied, base), + wantClass: "policy", + wantReason: "quota_denied", + }, + {name: "capacity", result: "failure", err: ErrNoWorkflowSlots, wantClass: "infrastructure", wantReason: "capacity_unavailable"}, + {name: "unclassified", result: "failure", err: base, wantClass: "infrastructure", wantReason: "runtime_failed"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotClass, gotReason := FailureAttribution(tt.result, tt.err) + if gotClass != tt.wantClass || gotReason != tt.wantReason { + t.Fatalf("FailureAttribution() = (%q, %q), want (%q, %q)", gotClass, gotReason, tt.wantClass, tt.wantReason) + } + }) + } +} + +func TestClassifiedFailurePreservesCauseAndFirstClassification(t *testing.T) { + cause := errors.New("cause") + first := ClassifiedFailure(FailureClassUser, FailureReasonCommandFailed, cause) + second := ClassifiedFailure(FailureClassInfrastructure, FailureReasonRuntimeFailed, first) + if second != first { + t.Fatal("nested classification replaced the original leaf classification") + } + if !errors.Is(second, cause) { + t.Fatal("classified failure does not preserve errors.Is") + } +} diff --git a/spindle/engines/microvm/agent.go b/spindle/engines/microvm/agent.go index 30c20946..23db3e51 100644 --- a/spindle/engines/microvm/agent.go +++ b/spindle/engines/microvm/agent.go @@ -22,6 +22,20 @@ const guestWorkflowUser = "spindle-workflow" var errGuestTimedOut = errors.New("guest reported step timed out") +type activationGuestError struct { + err error +} + +func (e *activationGuestError) Error() string { return e.err.Error() } +func (e *activationGuestError) Unwrap() error { return e.err } + +type activationTransportError struct { + err error +} + +func (e *activationTransportError) Error() string { return e.err.Error() } +func (e *activationTransportError) Unwrap() error { return e.err } + type agentHub struct { l *slog.Logger ln *vsock.Listener @@ -209,7 +223,7 @@ func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agent defer s.mu.Unlock() if id == "" { - return nil, fmt.Errorf("empty ID passed to ActivateConfig") + return nil, &activationTransportError{err: fmt.Errorf("empty ID passed to ActivateConfig")} } if req.TimeoutSeconds == 0 { req.TimeoutSeconds = timeoutSeconds(ctx, guestTimeoutGrace) @@ -218,13 +232,13 @@ func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agent Id: id, ActivateConfig: req, }); err != nil { - return nil, fmt.Errorf("send activate_config: %w", err) + return nil, &activationTransportError{err: fmt.Errorf("send activate_config: %w", err)} } for { msg, err := s.decode(ctx) if err != nil { - return nil, err + return nil, &activationTransportError{err: err} } if msg.BuiltPaths == nil && msg.Id != id { continue @@ -242,10 +256,10 @@ func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agent } } else if p := msg.ActivateConfigResult; p != nil { if p.Error != "" { - return nil, errors.New(p.Error) + return nil, &activationGuestError{err: errors.New(p.Error)} } if p.Toplevel == "" { - return nil, fmt.Errorf("activate config returned empty toplevel") + return nil, &activationGuestError{err: fmt.Errorf("activate config returned empty toplevel")} } return p, nil } diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go index e72d83d0..501f0f78 100644 --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -158,10 +158,10 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin var dwf manifestWorkflow if err := engine.DescribeManifestError(twf.Raw, manifestWorkflow{}); err != nil { - return nil, err + return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, err) } if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil { - return nil, err + return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, err) } for _, dstep := range dwf.Steps { @@ -194,9 +194,13 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin } if config.Enabled() { if !imageSpec.SupportsConfigActivation() { - return nil, fmt.Errorf( - "microVM image %q is not a NixOS image: services, virtualisation, dependencies and registry workflow options require a NixOS image", - imageName, + return nil, engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonConfigurationFailed, + fmt.Errorf( + "microVM image %q is not a NixOS image: services, virtualisation, dependencies and registry workflow options require a NixOS image", + imageName, + ), ) } // the cached toplevel is guest-asserted, so the cache key has to @@ -213,7 +217,11 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin var err error configKey, err = buildConfigKey(imageSpec, config, repoDid) if err != nil { - return nil, fmt.Errorf("build config key: %w", err) + return nil, engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonConfigurationFailed, + fmt.Errorf("build config key: %w", err), + ) } activationStep := Step{ name: "NixOS config activation", @@ -235,7 +243,7 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin cacheURLs, cacheKeys, err := workflowCaches(dwf.Caches) if err != nil { - return nil, err + return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonConfigurationFailed, err) } var ownerDID, repoDID string @@ -247,10 +255,18 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin } } if ownerDID == "" { - return nil, fmt.Errorf("missing owner DID in pipeline trigger metadata") + return nil, engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("missing owner DID in pipeline trigger metadata"), + ) } if repoDID == "" { - return nil, fmt.Errorf("missing repository DID in pipeline trigger metadata") + return nil, engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("missing repository DID in pipeline trigger metadata"), + ) } swf.OwnerDID = ownerDID swf.RepoDID = repoDID @@ -475,7 +491,20 @@ func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.W step := w.Steps[idx] if s, ok := step.(Step); ok && s.action == activationStepAction { err := e.activateConfig(execCtx, wid, state, s, wfLogger.DataWriter(idx, "stdout")) - return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "Failed to activate config", err) + failureClass := engine.FailureClassInfrastructure + failureReason := engine.FailureReasonRuntimeFailed + var guestError *activationGuestError + if errors.As(err, &guestError) { + failureClass = engine.FailureClassUser + failureReason = engine.FailureReasonConfigurationFailed + } + return e.classifyStepError( + ctx, wid, step, state, stderr, vmExited, + "Failed to activate config", + failureClass, + failureReason, + err, + ) } env := stepEnvironment(w, step, secrets) @@ -494,7 +523,13 @@ func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.W Stderr: stderr, }) if err != nil { - return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "User step error", err) + return e.classifyStepError( + ctx, wid, step, state, stderr, vmExited, + "User step error", + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + err, + ) } if exitCode != 0 { @@ -512,14 +547,29 @@ func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.W }) e.writeDebugHint(wid, len(w.Steps), wfLogger) } - return fmt.Errorf("User step error: exited with code %d", exitCode) + return engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonCommandFailed, + fmt.Errorf("User step error: exited with code %d", exitCode), + ) } return nil } // reads the vm serial logs so we report the tail of that as an error instead of // just "guest agent connection lost: EOF" -func (e *Engine) classifyStepError(ctx context.Context, wid models.WorkflowId, step models.Step, state *workflowState, stderr io.Writer, vmExited *atomic.Bool, category string, err error) error { +func (e *Engine) classifyStepError( + ctx context.Context, + wid models.WorkflowId, + step models.Step, + state *workflowState, + stderr io.Writer, + vmExited *atomic.Bool, + category string, + fallbackClass engine.FailureClass, + fallbackReason engine.FailureReason, + err error, +) error { if err == nil { return nil } @@ -538,7 +588,13 @@ func (e *Engine) classifyStepError(ctx context.Context, wid models.WorkflowId, s fmt.Fprintln(stderr, reason) l.Error(reason, "oom", oom) } - return fmt.Errorf("%s:\n%w", category, errors.New(reason+"; see workflow logs for serial output")) + class := engine.FailureClassInfrastructure + failureReason := engine.FailureReasonRuntimeFailed + if oom { + class = engine.FailureClassUser + failureReason = engine.FailureReasonResourceExhausted + } + return engine.ClassifiedFailure(class, failureReason, fmt.Errorf("%s:\n%w", category, errors.New(reason+"; see workflow logs for serial output"))) } if errors.Is(err, errGuestTimedOut) || ctx.Err() != nil { @@ -566,7 +622,11 @@ func (e *Engine) classifyStepError(ctx context.Context, wid models.WorkflowId, s l.Error("step failed", "error", err) crashErr = err } - return fmt.Errorf("%s:\n%w", category, crashErr) + return engine.ClassifiedFailure( + fallbackClass, + fallbackReason, + fmt.Errorf("%s:\n%w", category, crashErr), + ) } func (e *Engine) activateConfig(ctx context.Context, wid models.WorkflowId, state *workflowState, step Step, out io.Writer) error { diff --git a/spindle/engines/microvm/image.go b/spindle/engines/microvm/image.go index c2de0de0..4390f8aa 100644 --- a/spindle/engines/microvm/image.go +++ b/spindle/engines/microvm/image.go @@ -11,6 +11,7 @@ import ( "strings" "tangled.org/core/spindle/config" + "tangled.org/core/spindle/engine" ) const imageSpecFileName = "spec.json" @@ -191,39 +192,68 @@ func resolveImageSpecPath(base, path string) string { } func ResolveImageSpec(pipelinesCfg config.MicroVMPipelines, name string) (ImageSpec, string, string, error) { + explicitImage := strings.TrimSpace(name) != "" name = strings.TrimSpace(name) if name == "" { name = strings.TrimSpace(pipelinesCfg.DefaultImage) } if name == "" { - return ImageSpec{}, "", "", fmt.Errorf("no image specified in workflow and SPINDLE_MICROVM_PIPELINES_DEFAULT_IMAGE is not set") + return ImageSpec{}, "", "", engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("no image specified in workflow and SPINDLE_MICROVM_PIPELINES_DEFAULT_IMAGE is not set"), + ) } if !isPlainImageName(name) { - return ImageSpec{}, "", "", fmt.Errorf("invalid microVM image name %q: must be a plain name, not a path", name) + return ImageSpec{}, "", "", engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonConfigurationFailed, + fmt.Errorf("invalid microVM image name %q: must be a plain name, not a path", name), + ) } imageDir := strings.TrimSpace(pipelinesCfg.ImageDir) if imageDir == "" { - return ImageSpec{}, "", "", fmt.Errorf("microVM workflows require SPINDLE_MICROVM_PIPELINES_IMAGE_DIR") + return ImageSpec{}, "", "", engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("microVM workflows require SPINDLE_MICROVM_PIPELINES_IMAGE_DIR"), + ) } candidates := imageCandidates(imageDir, name) for _, candidate := range candidates { path, ok, err := imageSpecPath(candidate) if err != nil { - return ImageSpec{}, "", "", err + return ImageSpec{}, "", "", engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + err, + ) } if !ok { continue } imageSpec, err := LoadImageSpec(path) if err != nil { - return ImageSpec{}, "", "", err + return ImageSpec{}, "", "", engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + err, + ) } return imageSpec, path, name, nil } - return ImageSpec{}, "", "", fmt.Errorf("microVM image %q was not found; looked in: %s", name, strings.Join(candidates, ", ")) + class := engine.FailureClassInfrastructure + if explicitImage { + class = engine.FailureClassUser + } + return ImageSpec{}, "", "", engine.ClassifiedFailure( + class, + engine.FailureReasonConfigurationFailed, + fmt.Errorf("microVM image %q was not found; looked in: %s", name, strings.Join(candidates, ", ")), + ) } func (e *Engine) resolveImage(name string) (ImageSpec, string, string, error) { diff --git a/spindle/engines/nixery/engine.go b/spindle/engines/nixery/engine.go index b89013f2..22a5aa56 100644 --- a/spindle/engines/nixery/engine.go +++ b/spindle/engines/nixery/engine.go @@ -94,10 +94,10 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin Environment map[string]string `yaml:"environment"` }{} if err := engine.DescribeManifestError(twf.Raw, dwf); err != nil { - return nil, err + return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, err) } if err := yaml.Unmarshal([]byte(twf.Raw), &dwf); err != nil { - return nil, err + return nil, engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonWorkflowInvalid, err) } for _, dstep := range dwf.Steps { @@ -117,10 +117,18 @@ func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipelin } } if ownerDID == "" { - return nil, fmt.Errorf("missing owner DID in pipeline trigger metadata") + return nil, engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("missing owner DID in pipeline trigger metadata"), + ) } if repoDID == "" { - return nil, fmt.Errorf("missing repository DID in pipeline trigger metadata") + return nil, engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("missing repository DID in pipeline trigger metadata"), + ) } swf.OwnerDID = ownerDID swf.RepoDID = repoDID @@ -442,7 +450,11 @@ func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.W Env: envs, }) if err != nil { - return fmt.Errorf("User step error:\ncreating exec: %w", err) + return engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("creating exec: %w", err), + ) } // start tailing logs in background @@ -474,21 +486,33 @@ func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.W execInspectResp, err := e.docker.ContainerExecInspect(ctx, mkExecResp.ID) if err != nil { - return fmt.Errorf("User step error:\n%w", err) + return engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("inspecting exec: %w", err), + ) } if execInspectResp.ExitCode != 0 { inspectResp, err := e.docker.ContainerInspect(ctx, addl.container) if err != nil { - return fmt.Errorf("User step error:\n%w", err) + return engine.ClassifiedFailure( + engine.FailureClassInfrastructure, + engine.FailureReasonRuntimeFailed, + fmt.Errorf("inspecting workflow container: %w", err), + ) } e.l.Error("workflow failed!", "workflow_id", wid.String(), "exit_code", execInspectResp.ExitCode, "oom_killed", inspectResp.State.OOMKilled) if inspectResp.State.OOMKilled { - return fmt.Errorf("User step error:\n%w", ErrOOMKilled) + return engine.ClassifiedFailure(engine.FailureClassUser, engine.FailureReasonResourceExhausted, fmt.Errorf("User step error:\n%w", ErrOOMKilled)) } - return fmt.Errorf("User step error: exited with code %d", execInspectResp.ExitCode) + return engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonCommandFailed, + fmt.Errorf("User step error: exited with code %d", execInspectResp.ExitCode), + ) } return nil diff --git a/spindle/mill/auth_test.go b/spindle/mill/auth_test.go index 4e06a057..6464d1df 100644 --- a/spindle/mill/auth_test.go +++ b/spindle/mill/auth_test.go @@ -15,6 +15,7 @@ import ( "tangled.org/core/notifier" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" + "tangled.org/core/spindle/observability" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" @@ -209,6 +210,8 @@ func TestOnAttemptResultDeliversOwnedLease(t *testing.T) { m := New(l, Config{ReconnectGrace: time.Minute}) m.Attach(bdb, &n, testQuotaManager(t, bdb)) + metrics := observability.NewMetrics() + m.RegisterMetrics(metrics) owned := newLease("lease-owned", "node-b", "inc-b", "dummy") m.mu.Lock() @@ -242,6 +245,16 @@ func TestOnAttemptResultDeliversOwnedLease(t *testing.T) { if owned.getState() != leaseDone { t.Fatal("owned lease was not sealed after its terminal was delivered") } + + families, err := metrics.Registry().Gather() + if err != nil { + t.Fatal(err) + } + for _, family := range families { + if family.GetName() == "spindle_workflows_total" { + t.Fatal("mill counted a terminal whose executor retained metric authority") + } + } } func TestOnStatusEventOwnership(t *testing.T) { diff --git a/spindle/mill/executor/capability_test.go b/spindle/mill/executor/capability_test.go index a9b4cde7..1aa4001c 100644 --- a/spindle/mill/executor/capability_test.go +++ b/spindle/mill/executor/capability_test.go @@ -9,6 +9,7 @@ import ( "testing" "tangled.org/core/api/tangled" + "tangled.org/core/spindle/engine" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models" ) @@ -117,3 +118,35 @@ func TestHandleReserveValidatesPlacementBeforeAcquiringSlot(t *testing.T) { t.Fatalf("active reservations = %d, want 0", len(e.active)) } } + +func TestHandleReservePreservesTypedInitFailure(t *testing.T) { + enc := newCaptureEncoder() + eng := &fakeEngine{ + initErr: engine.ClassifiedFailure( + engine.FailureClassUser, + engine.FailureReasonWorkflowInvalid, + errors.New("invalid workflow"), + ), + } + e := &Executor{ + l: slog.New(slog.NewTextHandler(io.Discard, nil)), + enc: enc, + seats: 1, + engines: map[string]models.Engine{"microvm": eng}, + active: make(map[string]*reservation), + } + + e.handleReserve(context.Background(), testReserveSeat(t, "lease-1", "microvm")) + result := (<-enc.messages).GetReserveResult() + if result == nil || result.GetAccepted() { + t.Fatalf("reserve result = %+v, want rejection", result) + } + if result.GetFailureClass() != string(engine.FailureClassUser) || + result.GetFailureReason() != string(engine.FailureReasonWorkflowInvalid) { + t.Fatalf( + "failure attribution = %q/%q, want user/workflow_invalid", + result.GetFailureClass(), + result.GetFailureReason(), + ) + } +} diff --git a/spindle/mill/executor/executor.go b/spindle/mill/executor/executor.go index 16255818..e120479b 100644 --- a/spindle/mill/executor/executor.go +++ b/spindle/mill/executor/executor.go @@ -92,13 +92,16 @@ type reservation struct { repoDid syntax.DID vault *memVault - committed bool - cancelled bool - cancel context.CancelFunc - ttlTimer *time.Timer - stopTail func() - runDone chan struct{} - traceParent trace.SpanContext + committed bool + cancelled bool + cancel context.CancelFunc + ttlTimer *time.Timer + stopTail func() + runDone chan struct{} + traceParent trace.SpanContext + failureClass string + failureReason string + millRecordsTerminalMetrics bool } type messageEncoder interface { @@ -333,11 +336,21 @@ func (e *Executor) sendQuota(msg *millproto.Message) error { } func (e *Executor) sendReject(leaseID string, reason string, class millv1.RejectClass) { + e.sendRejectWithAttribution(leaseID, reason, class, "", "") +} + +func (e *Executor) sendRejectWithAttribution( + leaseID, reason string, + class millv1.RejectClass, + failureClass, failureReason string, +) { e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ - LeaseId: leaseID, - Accepted: false, - RejectReason: reason, - RejectClass: class, + LeaseId: leaseID, + Accepted: false, + RejectReason: reason, + RejectClass: class, + FailureClass: failureClass, + FailureReason: failureReason, }}) } @@ -385,14 +398,29 @@ func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { reject := func(reason string, class millv1.RejectClass) { e.sendReject(rs.GetLeaseId(), reason, class) } + rejectError := func(prefix string, err error, class millv1.RejectClass) { + failureClass, failureReason := engine.FailureAttribution("failure", err) + e.sendRejectWithAttribution( + rs.GetLeaseId(), + prefix+err.Error(), + class, + failureClass, + failureReason, + ) + } e.mu.Lock() draining := e.draining + atCapacity := len(e.active) >= e.seatCapacity() e.mu.Unlock() if draining { reject("draining", millv1.RejectClass_REJECT_CLASS_TRANSIENT) return } + if atCapacity { + reject("no executor seats available", millv1.RejectClass_REJECT_CLASS_TRANSIENT) + return + } realEngine, ok := e.engines[rs.GetTargetEngine()] if !ok { @@ -426,20 +454,20 @@ func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { } wf, err := realEngine.InitWorkflow(twf, tpl) if err != nil { - reject("init workflow: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) + rejectError("init workflow: ", err, millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) return } if e.quotaClient != nil { if binder, ok := realEngine.(engine.WorkflowQuotaStoreBinder); ok { if err := binder.BindWorkflowQuotaStore(wf, e.quotaClient.ForLease(rs.GetLeaseId())); err != nil { - reject("bind workflow quota store: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) + rejectError("bind workflow quota store: ", err, millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) return } } } if validator, ok := realEngine.(engine.WorkflowPlacementValidator); ok { if err := validator.ValidateWorkflowPlacement(wf); err != nil { - reject("validate workflow placement: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) + rejectError("validate workflow placement: ", err, millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) return } } @@ -487,6 +515,13 @@ func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { reject("draining", millv1.RejectClass_REJECT_CLASS_TRANSIENT) return } + if len(e.active) >= e.seatCapacity() { + e.mu.Unlock() + e.snapshotMu.Unlock() + slot.Release() + reject("no executor seats available", millv1.RejectClass_REJECT_CLASS_TRANSIENT) + return + } if len(e.active) == 0 && e.cleanupPending == 0 { e.idleCh = make(chan struct{}) } @@ -495,9 +530,10 @@ func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { e.mu.Unlock() e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ - LeaseId: rs.GetLeaseId(), - Accepted: true, - QuotaResources: reportedResources(realEngine, wf), + LeaseId: rs.GetLeaseId(), + Accepted: true, + QuotaResources: reportedResources(realEngine, wf), + SupportsMillTerminalMetrics: true, }}) accepted = true e.pushSnapshotLocked() @@ -533,6 +569,7 @@ func (e *Executor) handleCommit(ctx context.Context, cl *millv1.CommitLease) { return } res.committed = true + res.millRecordsTerminalMetrics = cl.GetMillRecordsTerminalMetrics() if res.ttlTimer != nil { res.ttlTimer.Stop() } @@ -566,6 +603,17 @@ func (e *Executor) handleCommit(ctx context.Context, cl *millv1.CommitLease) { } jobCtx, cancel := context.WithCancel(runCtx) + if cl.GetMillRecordsTerminalMetrics() { + jobCtx = observability.WithWorkflowTerminalObserver(jobCtx, func(_, _, failureClass, reason string) { + e.mu.Lock() + defer e.mu.Unlock() + if e.active[res.leaseID] != res { + return + } + res.failureClass = failureClass + res.failureReason = reason + }) + } res.cancel = cancel res.runDone = make(chan struct{}) e.mu.Unlock() @@ -773,6 +821,13 @@ func waitForDrainStage(ctx context.Context, done <-chan struct{}) error { } } +func (e *Executor) seatCapacity() int { + if e.seats > 0 { + return e.seats + } + return defaultSeats +} + func ttlDuration(secs uint32) time.Duration { if secs == 0 { return defaultReservationTTL @@ -804,6 +859,9 @@ func (e *Executor) RegisterMetrics(metrics *observability.Metrics) { return } metrics.RegisterExecutorGauges( + func() float64 { + return float64(e.seatCapacity()) + }, func() float64 { e.mu.Lock() defer e.mu.Unlock() diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go index 8d7dc989..b4f047f0 100644 --- a/spindle/mill/executor/observe.go +++ b/spindle/mill/executor/observe.go @@ -13,6 +13,7 @@ import ( "github.com/hpcloud/tail" "tangled.org/core/api/tangled" + "tangled.org/core/spindle/engine" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models" @@ -80,6 +81,11 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error if runDone != nil { <-runDone } + e.mu.Lock() + failureClass := res.failureClass + failureReason := res.failureReason + millRecordsTerminalMetrics := res.millRecordsTerminalMetrics + e.mu.Unlock() // the log tail finalizes first so all log lines precede the terminal event if res.stopTail != nil { @@ -90,6 +96,13 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error if cancelled { terminalStatus = string(models.StatusKindCancelled) } + if cancelled { + failureClass = string(engine.FailureClassUser) + failureReason = string(engine.FailureReasonCancelled) + } + if failureClass == "" || failureReason == "" { + failureClass, failureReason = defaultTerminalAttribution(terminalStatus) + } var logDir string if e.cfg != nil { logDir = e.cfg.Server.LogDir @@ -120,7 +133,10 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error // persist pending artifact state so a restart can retry the upload if e.db != nil { - _ = e.db.SavePendingArtifact(res.leaseID, res.wid.Name, terminalStatus, errStr, exitCode, ref, hash) + _ = e.db.SavePendingArtifact( + res.leaseID, res.wid.Name, terminalStatus, errStr, exitCode, ref, hash, + failureClass, failureReason, millRecordsTerminalMetrics, + ) } // upload with a context that survives job cancellation @@ -138,7 +154,10 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error } // append the terminal event with the LogArtifact ref - if err := e.appendTerminalWithArtifact(res.leaseID, terminalStatus, st, ref, hash); err != nil { + if err := e.appendTerminalWithArtifact( + res.leaseID, terminalStatus, st, ref, hash, + failureClass, failureReason, millRecordsTerminalMetrics, + ); err != nil { return err } diff --git a/spindle/mill/executor/outbox.go b/spindle/mill/executor/outbox.go index e210e1de..3de4396f 100644 --- a/spindle/mill/executor/outbox.go +++ b/spindle/mill/executor/outbox.go @@ -12,6 +12,7 @@ import ( "google.golang.org/protobuf/proto" "tangled.org/core/api/tangled" "tangled.org/core/spindle/db" + "tangled.org/core/spindle/engine" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models" @@ -123,10 +124,29 @@ func (e *Executor) appendStatus(leaseID string, st *tangled.PipelineStatus) erro } func (e *Executor) appendTerminal(leaseID, status string, st *tangled.PipelineStatus) error { - return e.appendTerminalWithArtifact(leaseID, status, st, "", "") + failureClass, failureReason := defaultTerminalAttribution(status) + return e.appendTerminalWithArtifact(leaseID, status, st, "", "", failureClass, failureReason, true) } -func (e *Executor) appendTerminalWithArtifact(leaseID, status string, st *tangled.PipelineStatus, ref, hash string) error { +func defaultTerminalAttribution(status string) (string, string) { + switch status { + case string(models.StatusKindSuccess): + return string(engine.FailureClassNone), string(engine.FailureReasonSuccess) + case string(models.StatusKindTimeout): + return string(engine.FailureClassUser), string(engine.FailureReasonTimeout) + case string(models.StatusKindCancelled): + return string(engine.FailureClassUser), string(engine.FailureReasonCancelled) + default: + return string(engine.FailureClassInfrastructure), string(engine.FailureReasonRuntimeFailed) + } +} + +func (e *Executor) appendTerminalWithArtifact( + leaseID, status string, + st *tangled.PipelineStatus, + ref, hash, failureClass, failureReason string, + millRecordsTerminalMetrics bool, +) error { var terminalStatus millv1.TerminalStatus switch status { case string(models.StatusKindSuccess): @@ -150,10 +170,13 @@ func (e *Executor) appendTerminalWithArtifact(leaseID, status string, st *tangle } } payload := &millv1.Event_AttemptResult{AttemptResult: &millv1.AttemptResult{ - Status: terminalStatus, - Error: errStr, - ExitCode: exit, - LogArtifact: logArtifact, + Status: terminalStatus, + Error: errStr, + ExitCode: exit, + LogArtifact: logArtifact, + FailureClass: failureClass, + FailureReason: failureReason, + MillRecordsTerminalMetrics: millRecordsTerminalMetrics, }} return e.appendAndSend(leaseID, payload, true) } @@ -195,7 +218,10 @@ func (e *Executor) recoverPendingArtifacts() error { Error: &p.Error, ExitCode: &p.ExitCode, } - if err := e.appendTerminalWithArtifact(p.LeaseID, p.Status, st, p.Ref, p.Hash); err == nil { + if err := e.appendTerminalWithArtifact( + p.LeaseID, p.Status, st, p.Ref, p.Hash, p.FailureClass, p.FailureReason, + p.MillRecordsTerminalMetrics, + ); err == nil { _ = e.db.RemovePendingArtifact(p.LeaseID) } } diff --git a/spindle/mill/executor/reserved_test.go b/spindle/mill/executor/reserved_test.go index da8b5438..9476fc4f 100644 --- a/spindle/mill/executor/reserved_test.go +++ b/spindle/mill/executor/reserved_test.go @@ -77,9 +77,13 @@ type fakeEngine struct { slot engine.WorkflowSlot secrets chan []secrets.UnlockedSecret done chan struct{} + initErr error } func (e *fakeEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { + if e.initErr != nil { + return nil, e.initErr + } return &models.Workflow{Name: twf.Name}, nil } func (e *fakeEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, l models.WorkflowLogger) error { @@ -407,6 +411,41 @@ func testExecutor(t *testing.T) *Executor { return e } +func TestReserveHonorsExecutorSeatCapacity(t *testing.T) { + enc := newCaptureEncoder() + e := testExecutor(t) + e.enc = enc + e.seats = 1 + e.engines = map[string]models.Engine{"dummy": &fakeEngine{}} + + e.handleReserve(context.Background(), testReserveSeat(t, "lease-1", "dummy")) + first := (<-enc.messages).GetReserveResult() + if first == nil || !first.GetAccepted() { + t.Fatalf("first reserve result = %+v, want accepted", first) + } + if snapshot := (<-enc.messages).GetNodeSnapshot(); snapshot == nil { + t.Fatal("accepted reservation did not publish a snapshot") + } + + e.handleReserve(context.Background(), testReserveSeat(t, "lease-2", "dummy")) + second := (<-enc.messages).GetReserveResult() + if second == nil || second.GetAccepted() { + t.Fatalf("second reserve result = %+v, want transient rejection", second) + } + if second.GetRejectClass() != millv1.RejectClass_REJECT_CLASS_TRANSIENT { + t.Fatalf("second reject class = %v, want transient", second.GetRejectClass()) + } + if second.GetRejectReason() != "no executor seats available" { + t.Fatalf("second reject reason = %q", second.GetRejectReason()) + } + + cleanup, ok := e.takeUncommittedReservation("lease-1", true) + if !ok { + t.Fatal("first reservation missing during cleanup") + } + cleanup() +} + func testDB(t *testing.T) *db.DB { d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) if err != nil { diff --git a/spindle/mill/integration_test.go b/spindle/mill/integration_test.go index 6af984f9..fa780c5c 100644 --- a/spindle/mill/integration_test.go +++ b/spindle/mill/integration_test.go @@ -20,6 +20,7 @@ import ( "tangled.org/core/spindle/engines/dummy" "tangled.org/core/spindle/mill/executor" "tangled.org/core/spindle/models" + "tangled.org/core/spindle/observability" ) func TestEndToEndDummyJob(t *testing.T) { @@ -35,6 +36,8 @@ func TestEndToEndDummyJob(t *testing.T) { bn := notifier.New() mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second}) mill.Attach(bdb, &bn, testQuotaManager(t, bdb)) + millMetrics := observability.NewMetrics() + mill.RegisterMetrics(millMetrics) registerTestExecutor(t, bdb, "exec-1", HashToken("test-token"), nil) srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn)) @@ -62,7 +65,9 @@ func TestEndToEndDummyJob(t *testing.T) { if err != nil { t.Fatalf("executor.New: %v", err) } - go exec.Connect(ctx) + execMetrics := observability.NewMetrics() + exec.RegisterMetrics(execMetrics) + go exec.Connect(observability.WithMetrics(ctx, execMetrics)) be := NewEngine("dummy", mill) twf := tangled.Pipeline_Workflow{ @@ -74,6 +79,9 @@ func TestEndToEndDummyJob(t *testing.T) { t.Fatalf("InitWorkflow: %v", err) } wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "rkey1"}, Name: "build"} + if err := bdb.StatusPending(wid, &bn); err != nil { + t.Fatal(err) + } placeCtx, placeCancel := context.WithTimeout(ctx, 10*time.Second) defer placeCancel() @@ -126,6 +134,55 @@ func TestEndToEndDummyJob(t *testing.T) { t.Logf("mill events after completion: %+v", events) t.Fatal("mill never saw streamed running status") } + + millFamilies, err := millMetrics.Registry().Gather() + if err != nil { + t.Fatal(err) + } + var millTotal, millClassified, millStartupSamples float64 + for _, family := range millFamilies { + switch family.GetName() { + case "spindle_workflows_total": + for _, metric := range family.GetMetric() { + millTotal += metric.GetCounter().GetValue() + } + case "spindle_workflow_terminations_total": + for _, metric := range family.GetMetric() { + millClassified += metric.GetCounter().GetValue() + } + case "spindle_workflow_startup_delay_seconds": + for _, metric := range family.GetMetric() { + millStartupSamples += float64(metric.GetHistogram().GetSampleCount()) + } + } + } + if millTotal != 1 || millClassified != 1 || millStartupSamples != 1 { + t.Fatalf( + "mill metrics = total %v classified %v startup samples %v, want exactly one of each", + millTotal, millClassified, millStartupSamples, + ) + } + + executorFamilies, err := execMetrics.Registry().Gather() + if err != nil { + t.Fatal(err) + } + var executorTotal, executorDurations float64 + for _, family := range executorFamilies { + switch family.GetName() { + case "spindle_workflows_total": + for _, metric := range family.GetMetric() { + executorTotal += metric.GetCounter().GetValue() + } + case "spindle_workflow_duration_seconds": + for _, metric := range family.GetMetric() { + executorDurations += float64(metric.GetHistogram().GetSampleCount()) + } + } + } + if executorTotal != 0 || executorDurations != 1 { + t.Fatalf("executor metrics = terminal %v duration samples %v, want 0 and 1", executorTotal, executorDurations) + } } func TestExecutorConfiguredLabelsAreStoredOnSession(t *testing.T) { diff --git a/spindle/mill/lease.go b/spindle/mill/lease.go index 4f211a8b..af8be700 100644 --- a/spindle/mill/lease.go +++ b/spindle/mill/lease.go @@ -49,7 +49,8 @@ type RemoteLease struct { ownerDID string repoDID string // keep the winning resources if quota waiting causes a re-bid - resources quota.Resources + resources quota.Resources + millRecordsTerminalMetrics bool // restored after a mill restart. no RunStep waits on it, so terminals // and death are authored directly. set before publication, never mutated orphaned bool diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go index 930c794c..91266ada 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -10,6 +10,7 @@ import ( "os" "path/filepath" "slices" + "sort" "strings" "sync" "time" @@ -351,11 +352,18 @@ func (m *Mill) failLeasesAfterGrace(sess *millSession) { } func (m *Mill) place(ctx context.Context, engineName string, wid models.WorkflowId, wf *models.Workflow) (engine.WorkflowSlot, error) { + placementStart := time.Now() + placementResult := "error" + defer func() { + m.metrics.RecordMillPlacementResult(ctx, engineName, placementResult, time.Since(placementStart)) + }() + m.mu.Lock() if m.cfg.MaxPending > 0 && m.pending >= m.cfg.MaxPending { max := m.cfg.MaxPending cur := m.pending m.mu.Unlock() + placementResult = "rejected" m.metrics.RecordMillPlacementAdmission(false, "max_pending_reached") return nil, fmt.Errorf("%w: mill has %d pending jobs (max %d)", engine.ErrNoWorkflowSlots, cur, max) } @@ -370,11 +378,10 @@ func (m *Mill) place(ctx context.Context, engineName string, wid models.Workflow for { if err := ctx.Err(); err != nil { - result := "cancelled" + placementResult = "cancelled" if errors.Is(err, context.DeadlineExceeded) { - result = "timeout" + placementResult = "timeout" } - m.metrics.RecordMillPlacementResult(result) return nil, err } @@ -384,17 +391,19 @@ func (m *Mill) place(ctx context.Context, engineName string, wid models.Workflow lease, err := m.bid(ctx, engineName, wid, wf) if err != nil { - m.metrics.RecordMillPlacementResult("error") return nil, err } if lease != nil { slot, retry, err := m.admit(ctx, engineName, wid, wf, lease) if err != nil { - m.metrics.RecordMillPlacementResult("error") + var failure *engine.WorkflowFailure + if errors.As(err, &failure) && failure.Class == engine.FailureClassPolicy { + placementResult = "rejected" + } return nil, err } if slot != nil { - m.metrics.RecordMillPlacementResult("success") + placementResult = "success" return slot, nil } // quota churn handed the seat back, bid again without waiting @@ -407,11 +416,10 @@ func (m *Mill) place(ctx context.Context, engineName string, wid models.Workflow // no executor available. wait for a change or ctx select { case <-ctx.Done(): - result := "cancelled" + placementResult = "cancelled" if errors.Is(ctx.Err(), context.DeadlineExceeded) { - result = "timeout" + placementResult = "timeout" } - m.metrics.RecordMillPlacementResult(result) return nil, ctx.Err() case <-ch: } @@ -435,7 +443,11 @@ func (m *Mill) admit(ctx context.Context, engineName string, wid models.Workflow case !res.Temporary: m.releaseSeat(lease) - return nil, false, fmt.Errorf("%w: workflow quota denied: %s", engine.ErrWorkflowFailed, res.Reason) + return nil, false, engine.ClassifiedFailure( + engine.FailureClassPolicy, + engine.FailureReasonQuotaDenied, + fmt.Errorf("%w: workflow quota denied: %s", engine.ErrWorkflowFailed, res.Reason), + ) } // don't hold fleet capacity while waiting for quota @@ -443,6 +455,9 @@ func (m *Mill) admit(ctx context.Context, engineName string, wid models.Workflow qlease, err = m.qm.Acquire(ctx, req) if err != nil { + if errors.Is(err, quota.ErrDenied) { + err = engine.ClassifiedFailure(engine.FailureClassPolicy, engine.FailureReasonQuotaDenied, err) + } return nil, false, fmt.Errorf("wait for workflow quota: %w", err) } @@ -574,11 +589,13 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId } type bidResult struct { - sess *millSession - lease *RemoteLease - rank int - incompatible bool - reason string + sess *millSession + lease *RemoteLease + rank int + incompatible bool + reason string + failureClass string + failureReason string } limit := m.cfg.TopK if limit <= 0 { @@ -661,7 +678,14 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId if !rr.GetAccepted() { m.dropReservation(leaseID) if rr.GetRejectClass() == millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE { - results <- bidResult{sess: sess, rank: rank, incompatible: true, reason: rr.GetRejectReason()} + results <- bidResult{ + sess: sess, + rank: rank, + incompatible: true, + reason: rr.GetRejectReason(), + failureClass: rr.GetFailureClass(), + failureReason: rr.GetFailureReason(), + } return } results <- bidResult{} @@ -676,6 +700,7 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId return } lease.resources = maps.Clone(rr.GetQuotaResources()) + lease.millRecordsTerminalMetrics = rr.GetSupportsMillTerminalMetrics() accepted = true results <- bidResult{sess: sess, lease: lease, rank: rank} } @@ -690,6 +715,9 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId var winner *bidResult var losers []*RemoteLease var incompatible []string + var incompatibleClass, incompatibleReason string + attributionConsistent := true + attributionSet := false // any soft reject (transient or timeout) means the fleet was just // busy, so an all-incompatible outcome isn't a hard placement error softReject := false @@ -702,6 +730,15 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId if r.reason != "" { incompatible = append(incompatible, r.reason) } + if r.failureClass == "" || r.failureReason == "" { + attributionConsistent = false + } else if !attributionSet { + incompatibleClass = r.failureClass + incompatibleReason = r.failureReason + attributionSet = true + } else if incompatibleClass != r.failureClass || incompatibleReason != r.failureReason { + attributionConsistent = false + } } else if r.lease == nil { softReject = true } @@ -734,7 +771,15 @@ func (m *Mill) bid(ctx context.Context, engineName string, wid models.WorkflowId if winner == nil { if len(incompatible) > 0 && !softReject { - return nil, fmt.Errorf("no compatible executor for %s: %s", engineName, strings.Join(incompatible, "; ")) + err := fmt.Errorf("no compatible executor for %s: %s", engineName, strings.Join(incompatible, "; ")) + if attributionSet && attributionConsistent { + return nil, engine.ClassifiedFailure( + engine.FailureClass(incompatibleClass), + engine.FailureReason(incompatibleReason), + err, + ) + } + return nil, err } return nil, nil } @@ -863,10 +908,11 @@ func (m *Mill) commitAndWait(ctx context.Context, wf *models.Workflow, unlocked traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) commit := &millproto.Message{CommitLease: &millv1.CommitLease{ - LeaseId: lease.id, - Secrets: pbSecrets, - Traceparent: traceparent, - Tracestate: tracestate, + LeaseId: lease.id, + Secrets: pbSecrets, + Traceparent: traceparent, + Tracestate: tracestate, + MillRecordsTerminalMetrics: lease.millRecordsTerminalMetrics, }} // commit retries ride reconnects, a reservation outlives one @@ -1210,6 +1256,39 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { var pendingTerminals []pendingTerminal var artifactLeases []*RemoteLease finishedInBatch := make(map[string]struct{}) + type startupObservation struct { + engine string + delay time.Duration + } + var startupObservations []startupObservation + leaseSet := make(map[*RemoteLease]struct{}) + for _, entry := range newEntries { + m.mu.Lock() + lease := m.leases[entry.LeaseId] + m.mu.Unlock() + if lease == nil || lease.nodeID != sess.nodeID { + continue + } + if lease.epoch != "" && lease.epoch != sess.epoch { + return protoErrf("lease %q epoch %q does not match session %q", lease.id, lease.epoch, sess.epoch) + } + leaseSet[lease] = struct{}{} + } + lockedLeases := make([]*RemoteLease, 0, len(leaseSet)) + for lease := range leaseSet { + lockedLeases = append(lockedLeases, lease) + } + sort.Slice(lockedLeases, func(i, j int) bool { + return lockedLeases[i].id < lockedLeases[j].id + }) + for _, lease := range lockedLeases { + lease.finishMu.Lock() + } + unlockLeases := func() { + for i := len(lockedLeases) - 1; i >= 0; i-- { + lockedLeases[i].finishMu.Unlock() + } + } var highestSeqno uint64 = current @@ -1259,9 +1338,25 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { } pipelineAtUri := string(lease.wid.PipelineId.AtUri()) if tx != nil { + alreadyRunning, err := tx.HasWorkflowStatus(context.Background(), pipelineAtUri, lease.wid.Name, statusStr) + if err != nil { + return err + } if err := tx.InsertStatusEvent(pipelineAtUri, lease.wid.Name, statusStr, errMsg, exitCode); err != nil { return err } + if !alreadyRunning { + delay, ok, err := tx.WorkflowStartupDelay(context.Background(), pipelineAtUri, lease.wid.Name) + if err != nil { + return err + } + if ok { + startupObservations = append(startupObservations, startupObservation{ + engine: lease.engine, + delay: delay, + }) + } + } } case entry.GetAttemptResult() != nil: @@ -1340,9 +1435,14 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { } if err != nil { + unlockLeases() return err } + for _, observation := range startupObservations { + m.metrics.RecordWorkflowStartupDelay(context.Background(), observation.engine, "mill", observation.delay) + } + if m.cfg.LogDir != "" { for _, lease := range artifactLeases { path := models.LogFilePath(m.cfg.LogDir, lease.wid) @@ -1359,20 +1459,44 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { m.mu.Unlock() for _, pt := range pendingTerminals { + if pt.lease.millRecordsTerminalMetrics && + pt.ar.GetMillRecordsTerminalMetrics() && + pt.ar.GetFailureClass() != "" && + pt.ar.GetFailureReason() != "" { + m.metrics.RecordWorkflowTerminal( + pt.lease.engine, + terminalMetricResult(pt.ar.GetStatus()), + pt.ar.GetFailureClass(), + pt.ar.GetFailureReason(), + ) + } // orphans have no waiting RunStep, just mark and clean up if pt.lease.orphaned { pt.lease.markDone() - _ = m.cleanupLease(pt.lease) + _ = m.cleanupLeaseLocked(pt.lease) continue } pt.lease.deliverTerminal(pt.ar) if pt.lease.cleanupReady() { - _ = m.cleanupLease(pt.lease) + _ = m.cleanupLeaseLocked(pt.lease) } } + unlockLeases() return m.sendAck(sess, highestSeqno) } +func terminalMetricResult(status millv1.TerminalStatus) string { + switch status { + case millv1.TerminalStatus_TERMINAL_STATUS_SUCCESS: + return "success" + case millv1.TerminalStatus_TERMINAL_STATUS_TIMEOUT: + return "timeout" + case millv1.TerminalStatus_TERMINAL_STATUS_CANCELLED: + return "cancelled" + default: + return "failure" + } +} func (m *Mill) sendAck(sess *millSession, seqno uint64) error { msg := &millproto.Message{Ack: &millv1.Ack{ diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go index 4e4034da..f27b3821 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -107,9 +107,11 @@ func sameStringMultiset(a, b []string) bool { } type reserveReply struct { - accepted bool - rejectClass millv1.RejectClass - reason string + accepted bool + rejectClass millv1.RejectClass + reason string + failureClass string + failureReason string } func addReplyingCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, asked chan<- string, reply reserveReply) *millSession { @@ -124,10 +126,12 @@ func addReplyingCandidateSession(t *testing.T, m *Mill, nodeID string, labels [] asked <- nodeID } sess.deliver(rs.GetLeaseId(), &millproto.Message{ReserveResult: &millv1.ReserveResult{ - LeaseId: rs.GetLeaseId(), - Accepted: reply.accepted, - RejectReason: reply.reason, - RejectClass: reply.rejectClass, + LeaseId: rs.GetLeaseId(), + Accepted: reply.accepted, + RejectReason: reply.reason, + RejectClass: reply.rejectClass, + FailureClass: reply.failureClass, + FailureReason: reply.failureReason, }}) return nil })) @@ -944,6 +948,32 @@ func TestCleanupRetry(t *testing.T) { } } +func TestBidPreservesConsistentTypedIncompatibility(t *testing.T) { + m := New(discardLogger(), Config{TopK: 1, BidTimeout: time.Second}) + addReplyingCandidateSession(t, m, "node-1", nil, 0, nil, reserveReply{ + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, + reason: "invalid workflow", + failureClass: string(engine.FailureClassUser), + failureReason: string(engine.FailureReasonWorkflowInvalid), + }) + + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "pipeline"}, + Name: "build", + } + _, err := m.bid(context.Background(), "dummy", wid, testWorkflow("build")) + if err == nil { + t.Fatal("bid succeeded despite incompatible executor") + } + var failure *engine.WorkflowFailure + if !errors.As(err, &failure) { + t.Fatalf("bid error %T did not preserve typed attribution", err) + } + if failure.Class != engine.FailureClassUser || failure.Reason != engine.FailureReasonWorkflowInvalid { + t.Fatalf("failure attribution = %s/%s", failure.Class, failure.Reason) + } +} + func TestBoundedBidding(t *testing.T) { m := New(discardLogger(), Config{TopK: 2, BidTimeout: 10 * time.Millisecond}) diff --git a/spindle/mill/proto/gen/mill.pb.go b/spindle/mill/proto/gen/mill.pb.go index e8aa399e..30904ff6 100644 --- a/spindle/mill/proto/gen/mill.pb.go +++ b/spindle/mill/proto/gen/mill.pb.go @@ -578,14 +578,17 @@ func (x *ReserveSeat) GetTracestate() string { } type ReserveResult struct { - state protoimpl.MessageState `protogen:"open.v1"` - LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` - Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"` - RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"` - RejectClass RejectClass `protobuf:"varint,4,opt,name=reject_class,json=rejectClass,proto3,enum=spindle.mill.v1.RejectClass" json:"reject_class,omitempty"` - QuotaResources map[string]int64 `protobuf:"bytes,5,rep,name=quota_resources,json=quotaResources,proto3" json:"quota_resources,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` + Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"` + RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"` + RejectClass RejectClass `protobuf:"varint,4,opt,name=reject_class,json=rejectClass,proto3,enum=spindle.mill.v1.RejectClass" json:"reject_class,omitempty"` + QuotaResources map[string]int64 `protobuf:"bytes,5,rep,name=quota_resources,json=quotaResources,proto3" json:"quota_resources,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` + FailureClass string `protobuf:"bytes,6,opt,name=failure_class,json=failureClass,proto3" json:"failure_class,omitempty"` + FailureReason string `protobuf:"bytes,7,opt,name=failure_reason,json=failureReason,proto3" json:"failure_reason,omitempty"` + SupportsMillTerminalMetrics bool `protobuf:"varint,8,opt,name=supports_mill_terminal_metrics,json=supportsMillTerminalMetrics,proto3" json:"supports_mill_terminal_metrics,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ReserveResult) Reset() { @@ -653,6 +656,27 @@ func (x *ReserveResult) GetQuotaResources() map[string]int64 { return nil } +func (x *ReserveResult) GetFailureClass() string { + if x != nil { + return x.FailureClass + } + return "" +} + +func (x *ReserveResult) GetFailureReason() string { + if x != nil { + return x.FailureReason + } + return "" +} + +func (x *ReserveResult) GetSupportsMillTerminalMetrics() bool { + if x != nil { + return x.SupportsMillTerminalMetrics + } + return false +} + // a single unlocked secret type Secret struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -708,13 +732,14 @@ func (x *Secret) GetValue() string { // promotes a reservation to a running job and hands over the secrets type CommitLease struct { - state protoimpl.MessageState `protogen:"open.v1"` - LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` - Secrets []*Secret `protobuf:"bytes,2,rep,name=secrets,proto3" json:"secrets,omitempty"` - Traceparent string `protobuf:"bytes,3,opt,name=traceparent,proto3" json:"traceparent,omitempty"` - Tracestate string `protobuf:"bytes,4,opt,name=tracestate,proto3" json:"tracestate,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` + Secrets []*Secret `protobuf:"bytes,2,rep,name=secrets,proto3" json:"secrets,omitempty"` + Traceparent string `protobuf:"bytes,3,opt,name=traceparent,proto3" json:"traceparent,omitempty"` + Tracestate string `protobuf:"bytes,4,opt,name=tracestate,proto3" json:"tracestate,omitempty"` + MillRecordsTerminalMetrics bool `protobuf:"varint,5,opt,name=mill_records_terminal_metrics,json=millRecordsTerminalMetrics,proto3" json:"mill_records_terminal_metrics,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *CommitLease) Reset() { @@ -775,6 +800,13 @@ func (x *CommitLease) GetTracestate() string { return "" } +func (x *CommitLease) GetMillRecordsTerminalMetrics() bool { + if x != nil { + return x.MillRecordsTerminalMetrics + } + return false +} + type Committed struct { state protoimpl.MessageState `protogen:"open.v1"` LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` @@ -1075,13 +1107,16 @@ func (x *LogArtifact) GetHash() string { // terminal outcome of an attempt type AttemptResult struct { - state protoimpl.MessageState `protogen:"open.v1"` - Status TerminalStatus `protobuf:"varint,1,opt,name=status,proto3,enum=spindle.mill.v1.TerminalStatus" json:"status,omitempty"` - Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` - ExitCode int64 `protobuf:"varint,3,opt,name=exit_code,json=exitCode,proto3" json:"exit_code,omitempty"` - LogArtifact *LogArtifact `protobuf:"bytes,4,opt,name=log_artifact,json=logArtifact,proto3" json:"log_artifact,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Status TerminalStatus `protobuf:"varint,1,opt,name=status,proto3,enum=spindle.mill.v1.TerminalStatus" json:"status,omitempty"` + Error string `protobuf:"bytes,2,opt,name=error,proto3" json:"error,omitempty"` + ExitCode int64 `protobuf:"varint,3,opt,name=exit_code,json=exitCode,proto3" json:"exit_code,omitempty"` + LogArtifact *LogArtifact `protobuf:"bytes,4,opt,name=log_artifact,json=logArtifact,proto3" json:"log_artifact,omitempty"` + FailureClass string `protobuf:"bytes,5,opt,name=failure_class,json=failureClass,proto3" json:"failure_class,omitempty"` + FailureReason string `protobuf:"bytes,6,opt,name=failure_reason,json=failureReason,proto3" json:"failure_reason,omitempty"` + MillRecordsTerminalMetrics bool `protobuf:"varint,7,opt,name=mill_records_terminal_metrics,json=millRecordsTerminalMetrics,proto3" json:"mill_records_terminal_metrics,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *AttemptResult) Reset() { @@ -1142,6 +1177,27 @@ func (x *AttemptResult) GetLogArtifact() *LogArtifact { return nil } +func (x *AttemptResult) GetFailureClass() string { + if x != nil { + return x.FailureClass + } + return "" +} + +func (x *AttemptResult) GetFailureReason() string { + if x != nil { + return x.FailureReason + } + return "" +} + +func (x *AttemptResult) GetMillRecordsTerminalMetrics() bool { + if x != nil { + return x.MillRecordsTerminalMetrics + } + return false +} + // live non-replay log frame type LiveLog struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -1556,8 +1612,6 @@ func (x *Message) GetQuotaResp() *QuotaResponse { return nil } -// asks the mill to perform one reservation lifecycle operation -// the mill resolves identity from lease_id so the executor cannot choose who pays type QuotaRequest struct { state protoimpl.MessageState `protogen:"open.v1"` RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` @@ -1657,10 +1711,8 @@ type QuotaResponse struct { Allowed bool `protobuf:"varint,3,opt,name=allowed,proto3" json:"allowed,omitempty"` Reason string `protobuf:"bytes,4,opt,name=reason,proto3" json:"reason,omitempty"` Error string `protobuf:"bytes,5,opt,name=error,proto3" json:"error,omitempty"` - // a denial the fair queue may still grant later - // callers must not treat it as a permanent refusal - Temporary bool `protobuf:"varint,6,opt,name=temporary,proto3" json:"temporary,omitempty"` - Resource string `protobuf:"bytes,7,opt,name=resource,proto3" json:"resource,omitempty"` + Temporary bool `protobuf:"varint,6,opt,name=temporary,proto3" json:"temporary,omitempty"` + Resource string `protobuf:"bytes,7,opt,name=resource,proto3" json:"resource,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1782,26 +1834,30 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "\vtraceparent\x18\b \x01(\tR\vtraceparent\x12\x1e\n" + "\n" + "tracestate\x18\t \x01(\tR\n" + - "tracestate\"\x81\x03\n" + + "tracestate\"\xa4\x04\n" + "\rReserveResult\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x1a\n" + "\baccepted\x18\x02 \x01(\bR\baccepted\x12#\n" + "\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12I\n" + "\freject_class\x18\x04 \x01(\x0e2\x1c.spindle.mill.v1.RejectClassB\b\xbaH\x05\x82\x01\x02\x10\x01R\vrejectClass\x12}\n" + - "\x0fquota_resources\x18\x05 \x03(\v22.spindle.mill.v1.ReserveResult.QuotaResourcesEntryB \xbaH\x1d\x9a\x01\x1a\x10\x10\"\x06r\x04 \x01(@*\x0e\"\f\x18\x80\x80\x80\x80\x80\x80\x80\x80@(\x00R\x0equotaResources\x1aA\n" + + "\x0fquota_resources\x18\x05 \x03(\v22.spindle.mill.v1.ReserveResult.QuotaResourcesEntryB \xbaH\x1d\x9a\x01\x1a\x10\x10\"\x06r\x04 \x01(@*\x0e\"\f\x18\x80\x80\x80\x80\x80\x80\x80\x80@(\x00R\x0equotaResources\x12,\n" + + "\rfailure_class\x18\x06 \x01(\tB\a\xbaH\x04r\x02( R\ffailureClass\x12.\n" + + "\x0efailure_reason\x18\a \x01(\tB\a\xbaH\x04r\x02(@R\rfailureReason\x12C\n" + + "\x1esupports_mill_terminal_metrics\x18\b \x01(\bR\x1bsupportsMillTerminalMetrics\x1aA\n" + "\x13QuotaResourcesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01\"0\n" + "\x06Secret\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value\"\xa6\x01\n" + + "\x05value\x18\x02 \x01(\tR\x05value\"\xe9\x01\n" + "\vCommitLease\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x121\n" + "\asecrets\x18\x02 \x03(\v2\x17.spindle.mill.v1.SecretR\asecrets\x12 \n" + "\vtraceparent\x18\x03 \x01(\tR\vtraceparent\x12\x1e\n" + "\n" + "tracestate\x18\x04 \x01(\tR\n" + - "tracestate\"/\n" + + "tracestate\x12A\n" + + "\x1dmill_records_terminal_metrics\x18\x05 \x01(\bR\x1amillRecordsTerminalMetrics\"/\n" + "\tCommitted\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\"2\n" + "\fReleaseLease\x12\"\n" + @@ -1818,13 +1874,16 @@ const file_spindle_mill_v1_mill_proto_rawDesc = "" + "\texit_code\x18\x03 \x01(\x03R\bexitCode\"E\n" + "\vLogArtifact\x12\x19\n" + "\x03ref\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x03ref\x12\x1b\n" + - "\x04hash\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x04hash\"\xd3\x01\n" + + "\x04hash\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x04hash\"\xf4\x02\n" + "\rAttemptResult\x12C\n" + "\x06status\x18\x01 \x01(\x0e2\x1f.spindle.mill.v1.TerminalStatusB\n" + "\xbaH\a\x82\x01\x04\x10\x01 \x00R\x06status\x12\x1f\n" + "\x05error\x18\x02 \x01(\tB\t\xbaH\x06r\x04(\x80\x80\x04R\x05error\x12\x1b\n" + "\texit_code\x18\x03 \x01(\x03R\bexitCode\x12?\n" + - "\flog_artifact\x18\x04 \x01(\v2\x1c.spindle.mill.v1.LogArtifactR\vlogArtifact\"Q\n" + + "\flog_artifact\x18\x04 \x01(\v2\x1c.spindle.mill.v1.LogArtifactR\vlogArtifact\x12,\n" + + "\rfailure_class\x18\x05 \x01(\tB\a\xbaH\x04r\x02( R\ffailureClass\x12.\n" + + "\x0efailure_reason\x18\x06 \x01(\tB\a\xbaH\x04r\x02(@R\rfailureReason\x12A\n" + + "\x1dmill_records_terminal_metrics\x18\a \x01(\bR\x1amillRecordsTerminalMetrics\"Q\n" + "\aLiveLog\x12\"\n" + "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\"\n" + "\braw_json\x18\x02 \x01(\fB\a\xbaH\x04z\x02\x10\x01R\arawJson\"\xe8\x01\n" + diff --git a/spindle/mill/proto/spindle/mill/v1/mill.proto b/spindle/mill/proto/spindle/mill/v1/mill.proto index ed93751b..0ff2e217 100644 --- a/spindle/mill/proto/spindle/mill/v1/mill.proto +++ b/spindle/mill/proto/spindle/mill/v1/mill.proto @@ -67,6 +67,9 @@ message ReserveResult { keys: {string: {min_bytes: 1, max_bytes: 64}}, values: {int64: {gte: 0, lte: 4611686018427387904}} }]; + string failure_class = 6 [(buf.validate.field).string.max_bytes = 32]; + string failure_reason = 7 [(buf.validate.field).string.max_bytes = 64]; + bool supports_mill_terminal_metrics = 8; } // a single unlocked secret @@ -81,6 +84,7 @@ message CommitLease { repeated Secret secrets = 2; string traceparent = 3; string tracestate = 4; + bool mill_records_terminal_metrics = 5; } message Committed { @@ -138,6 +142,9 @@ message AttemptResult { string error = 2 [(buf.validate.field).string.max_bytes = 65536]; int64 exit_code = 3; LogArtifact log_artifact = 4; + string failure_class = 5 [(buf.validate.field).string.max_bytes = 32]; + string failure_reason = 6 [(buf.validate.field).string.max_bytes = 64]; + bool mill_records_terminal_metrics = 7; } // live non-replay log frame diff --git a/spindle/mill/quota_admission_test.go b/spindle/mill/quota_admission_test.go index fa9df45a..093757f0 100644 --- a/spindle/mill/quota_admission_test.go +++ b/spindle/mill/quota_admission_test.go @@ -14,6 +14,7 @@ import ( "tangled.org/core/notifier" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" + "tangled.org/core/spindle/observability" "tangled.org/core/spindle/quota" millproto "tangled.org/core/spindle/mill/proto" @@ -254,6 +255,8 @@ func TestPlaceFailsAndReleasesSeatOnPermanentDenial(t *testing.T) { {Allowed: false, Temporary: false, Reason: "request_exceeds_limit", Resource: "memory_mib"}, }} m, _ := quotaMill(t, store) + metrics := observability.NewMetrics() + m.RegisterMetrics(metrics) cand := addQuotaCandidate(t, m, "node-a", fixedReport(2, 2048)) wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "rkey1"}, Name: "build"} @@ -276,6 +279,30 @@ func TestPlaceFailsAndReleasesSeatOnPermanentDenial(t *testing.T) { if live != 0 { t.Errorf("mill kept %d lease records after a permanent denial", live) } + families, err := metrics.Registry().Gather() + if err != nil { + t.Fatal(err) + } + var rejectedResult, admissionTotal float64 + for _, family := range families { + for _, metric := range family.GetMetric() { + labels := map[string]string{} + for _, label := range metric.GetLabel() { + labels[label.GetName()] = label.GetValue() + } + switch family.GetName() { + case "spindle_mill_placement_results_total": + if labels["result"] == "rejected" { + rejectedResult += metric.GetCounter().GetValue() + } + case "spindle_mill_placement_admission_total": + admissionTotal += metric.GetCounter().GetValue() + } + } + } + if rejectedResult != 1 || admissionTotal != 1 { + t.Fatalf("quota denial metrics = rejected result %v admission decisions %v, want one of each", rejectedResult, admissionTotal) + } } func TestPlaceRejectsRebidWithDifferentResources(t *testing.T) { diff --git a/spindle/mill/restore.go b/spindle/mill/restore.go index de752a3a..ee86e776 100644 --- a/spindle/mill/restore.go +++ b/spindle/mill/restore.go @@ -7,6 +7,7 @@ import ( "time" "tangled.org/core/spindle/db" + "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" millproto "tangled.org/core/spindle/mill/proto" @@ -26,17 +27,18 @@ func (m *Mill) persistLease(lease *RemoteLease, state string) error { quotaID := lease.quotaID lease.mu.Unlock() return m.db.SaveMillLease(db.MillLease{ - LeaseID: lease.id, - NodeID: lease.nodeID, - Epoch: lease.epoch, - Engine: lease.engine, - Knot: lease.wid.Knot, - Rkey: lease.wid.Rkey, - Workflow: lease.wid.Name, - State: state, - QuotaReservationID: quotaID, - OwnerDID: lease.ownerDID, - RepoDID: lease.repoDID, + LeaseID: lease.id, + NodeID: lease.nodeID, + Epoch: lease.epoch, + Engine: lease.engine, + Knot: lease.wid.Knot, + Rkey: lease.wid.Rkey, + Workflow: lease.wid.Name, + State: state, + QuotaReservationID: quotaID, + OwnerDID: lease.ownerDID, + RepoDID: lease.repoDID, + MillRecordsTerminalMetrics: lease.millRecordsTerminalMetrics, }) } @@ -87,6 +89,7 @@ func (m *Mill) RestoreState() error { lease.quotaID = r.QuotaReservationID lease.ownerDID = r.OwnerDID lease.repoDID = r.RepoDID + lease.millRecordsTerminalMetrics = r.MillRecordsTerminalMetrics // restored leases start as orphans, an executor must reclaim it via // its first snapshot, or the sweep will fail it lease.orphaned = true @@ -223,7 +226,11 @@ func (m *Mill) finishLiveLease(lease *RemoteLease, status, reason string) error Status: mapTerminalStatusString(status), Error: reason, }) - return m.cleanupLeaseLocked(lease) + m.recordSyntheticTerminal(lease, status) + if err := m.cleanupLeaseLocked(lease); err != nil { + return err + } + return nil } func (m *Mill) finishOrphan(lease *RemoteLease, status string, errMsg *string, exitCode *int64) error { @@ -236,7 +243,35 @@ func (m *Mill) finishOrphan(lease *RemoteLease, status string, errMsg *string, e return err } lease.markDone() - return m.cleanupLeaseLocked(lease) + m.recordSyntheticTerminal(lease, status) + if err := m.cleanupLeaseLocked(lease); err != nil { + return err + } + return nil +} + +func (m *Mill) recordSyntheticTerminal(lease *RemoteLease, status string) { + if !lease.millRecordsTerminalMetrics { + return + } + result := "failure" + class := engine.FailureClassInfrastructure + reason := engine.FailureReasonExecutorLost + switch status { + case string(models.StatusKindSuccess): + result = "success" + class = engine.FailureClassNone + reason = engine.FailureReasonSuccess + case string(models.StatusKindTimeout): + result = "timeout" + class = engine.FailureClassUser + reason = engine.FailureReasonTimeout + case string(models.StatusKindCancelled): + result = "cancelled" + class = engine.FailureClassUser + reason = engine.FailureReasonCancelled + } + m.metrics.RecordWorkflowTerminal(lease.engine, result, string(class), string(reason)) } func mapTerminalStatusString(s string) millv1.TerminalStatus { diff --git a/spindle/mill/restore_test.go b/spindle/mill/restore_test.go index 62e24ae2..8b7ecb6c 100644 --- a/spindle/mill/restore_test.go +++ b/spindle/mill/restore_test.go @@ -11,6 +11,7 @@ import ( millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models" + "tangled.org/core/spindle/observability" ) func restoreTestMill(t *testing.T, cfg Config) (*Mill, *db.DB) { @@ -384,3 +385,36 @@ func TestOrphanTerminalFailureKeepsLeaseAndSeqnoRetryable(t *testing.T) { t.Fatalf("durable leases after successful retry = %+v, err = %v; want none", rows, err) } } + +func TestSyntheticTerminalRespectsPersistedMetricAuthority(t *testing.T) { + m, _ := restoreTestMill(t, Config{ReconnectGrace: time.Minute}) + metrics := observability.NewMetrics() + m.RegisterMetrics(metrics) + + lease := newLease("lease-local-metrics", "node-old", "epoch-old", "dummy") + lease.wid = models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "pipeline"}, + Name: "build", + } + lease.millRecordsTerminalMetrics = false + if err := m.persistLease(lease, leaseRowRunning); err != nil { + t.Fatal(err) + } + m.mu.Lock() + m.leases[lease.id] = lease + m.mu.Unlock() + + reason := "executor lost" + if err := m.finishOrphan(lease, string(models.StatusKindFailed), &reason, nil); err != nil { + t.Fatal(err) + } + families, err := metrics.Registry().Gather() + if err != nil { + t.Fatal(err) + } + for _, family := range families { + if family.GetName() == "spindle_workflows_total" { + t.Fatal("mill synthesized a terminal while the executor retained metric authority") + } + } +} diff --git a/spindle/observability/metrics.go b/spindle/observability/metrics.go index f3a764da..728be908 100644 --- a/spindle/observability/metrics.go +++ b/spindle/observability/metrics.go @@ -17,8 +17,41 @@ import ( "time" ) +var workflowDurationBuckets = []float64{ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200, +} + +var queueDurationBuckets = []float64{ + 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300, 600, 900, 1800, 3600, 7200, 21600, 43200, 86400, +} + +var quotaWaitDurationBuckets = append( + prometheus.ExponentialBuckets(0.1, 3, 8), + 300, 600, 900, 1800, 3600, 7200, +) + type contextKey struct{} +type workflowTerminalObserverKey struct{} + +type WorkflowTerminalObserver func(engine, result, failureClass, reason string) + +func WithWorkflowTerminalObserver(ctx context.Context, observer WorkflowTerminalObserver) context.Context { + return context.WithValue(ctx, workflowTerminalObserverKey{}, observer) +} + +func NotifyWorkflowTerminal(ctx context.Context, engine, result, failureClass, reason string) bool { + if ctx == nil { + return false + } + observer, ok := ctx.Value(workflowTerminalObserverKey{}).(WorkflowTerminalObserver) + if !ok || observer == nil { + return false + } + observer(engine, result, failureClass, reason) + return true +} + func WithMetrics(ctx context.Context, m *Metrics) context.Context { return context.WithValue(ctx, contextKey{}, m) } @@ -114,11 +147,14 @@ type Metrics struct { eventIngestion *prometheus.CounterVec - jobQueueActivity *prometheus.CounterVec + jobQueueActivity *prometheus.CounterVec + jobDequeueLatency prometheus.Histogram - workflowsActive *prometheus.GaugeVec - workflowsTotal *prometheus.CounterVec - workflowDuration *prometheus.HistogramVec + workflowsActive *prometheus.GaugeVec + workflowsTotal *prometheus.CounterVec + workflowTerminations *prometheus.CounterVec + workflowDuration *prometheus.HistogramVec + workflowStartupDelay *prometheus.HistogramVec stepsActive *prometheus.GaugeVec stepsTotal *prometheus.CounterVec @@ -133,6 +169,7 @@ type Metrics struct { millPlacementAdmission *prometheus.CounterVec millPlacementResults *prometheus.CounterVec millReconnects prometheus.Counter + millPlacementWait *prometheus.HistogramVec jumpActive prometheus.Gauge jumpMax prometheus.Gauge @@ -192,6 +229,12 @@ func newMetricsWithClock(clock clock) *Metrics { Help: "Total count of job queue operations.", }, []string{"action"}), + jobDequeueLatency: prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "spindle_job_dequeue_latency_seconds", + Help: "Time from durable pipeline admission until a worker dequeues the job.", + Buckets: queueDurationBuckets, + }), + workflowsActive: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "spindle_workflows_active", Help: "Current number of active workflows by engine.", @@ -202,12 +245,23 @@ func newMetricsWithClock(clock clock) *Metrics { Help: "Total number of completed workflows by engine and result.", }, []string{"engine", "result"}), + workflowTerminations: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "spindle_workflow_terminations_total", + Help: "Total number of completed workflows by engine, result, failure class and bounded reason.", + }, []string{"engine", "result", "failure_class", "reason"}), + workflowDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "spindle_workflow_duration_seconds", Help: "Duration of completed workflows by engine and result.", - Buckets: prometheus.DefBuckets, + Buckets: workflowDurationBuckets, }, []string{"engine", "result"}), + workflowStartupDelay: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "spindle_workflow_startup_delay_seconds", + Help: "Time from pending admission until a workflow starts running.", + Buckets: queueDurationBuckets, + }, []string{"engine", "placement"}), + stepsActive: prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "spindle_steps_active", Help: "Current number of active steps by engine.", @@ -221,7 +275,7 @@ func newMetricsWithClock(clock clock) *Metrics { stepDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "spindle_step_duration_seconds", Help: "Duration of completed steps by engine and status.", - Buckets: prometheus.DefBuckets, + Buckets: workflowDurationBuckets, }, []string{"engine", "status"}), poolMemory: prometheus.NewGaugeVec(prometheus.GaugeOpts{ @@ -264,6 +318,12 @@ func newMetricsWithClock(clock clock) *Metrics { Help: "Total number of executor reconnects to the mill.", }), + millPlacementWait: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "spindle_mill_placement_wait_seconds", + Help: "Time spent waiting for mill placement by engine and result.", + Buckets: queueDurationBuckets, + }, []string{"engine", "result"}), + jumpActive: prometheus.NewGauge(prometheus.GaugeOpts{ Name: "spindle_jump_active_connections", Help: "Current number of active debug jump connections.", @@ -307,7 +367,7 @@ func newMetricsWithClock(clock clock) *Metrics { quotaWaitTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "spindle_quota_wait_duration_seconds", Help: "Time spent waiting for quota before a grant, by kind and the resource that was short.", - Buckets: prometheus.ExponentialBuckets(0.1, 3, 8), + Buckets: quotaWaitDurationBuckets, }, []string{"kind", "resource"}), collectionFailures: prometheus.NewCounter(prometheus.CounterOpts{ Name: "spindle_collection_failures_total", @@ -321,9 +381,12 @@ func newMetricsWithClock(clock clock) *Metrics { m.httpInFlight, m.eventIngestion, m.jobQueueActivity, + m.jobDequeueLatency, m.workflowsActive, m.workflowsTotal, + m.workflowTerminations, m.workflowDuration, + m.workflowStartupDelay, m.stepsActive, m.stepsTotal, m.stepDuration, @@ -335,6 +398,7 @@ func newMetricsWithClock(clock clock) *Metrics { m.millPlacementAdmission, m.millPlacementResults, m.millReconnects, + m.millPlacementWait, m.jumpActive, m.jumpMax, m.jumpRejections, @@ -452,6 +516,71 @@ func quotaDecision(allowed, temporary bool) string { } } +func boundWorkflowResult(result string) string { + switch result { + case "success", "failure", "timeout", "cancelled": + return result + default: + return "unknown" + } +} + +func boundFailureClass(class string) string { + switch class { + case "none", "user", "infrastructure", "policy": + return class + default: + return "unknown" + } +} + +func boundFailureReason(reason string) string { + switch reason { + case "success", "command_failed", "resource_exhausted", "configuration_failed", + "workflow_invalid", "setup_failed", "runtime_failed", "capacity_unavailable", + "quota_denied", "executor_lost", "timeout", "cancelled": + return reason + default: + return "unknown" + } +} + +func boundEngine(engine string) string { + switch engine { + case "dummy", "microvm", "nixery": + return engine + default: + return "unknown" + } +} + +func boundPlacement(placement string) string { + switch placement { + case "local", "mill": + return placement + default: + return "unknown" + } +} + +func boundMillPlacementResult(result string) string { + switch result { + case "success", "timeout", "cancelled", "error", "rejected": + return result + default: + return "unknown" + } +} + +func boundMillPlacementReason(reason string) string { + switch reason { + case "allowed", "max_pending_reached": + return reason + default: + return "unknown" + } +} + func (m *Metrics) RecordCacheUpload(backend, result string) { if m == nil { return @@ -637,6 +766,13 @@ func (m *Metrics) RecordJobQueueActivity(action string) { m.jobQueueActivity.WithLabelValues(action).Inc() } +func (m *Metrics) RecordJobDequeueLatency(ctx context.Context, d time.Duration) { + if m == nil || d < 0 { + return + } + observeWithExemplar(ctx, m.jobDequeueLatency, d.Seconds()) +} + func (m *Metrics) SetQuotaDefaultLimit(scope, resource string, limit int64) { if m == nil { return @@ -653,7 +789,7 @@ func (m *Metrics) RecordWorkflowStart(engine string) { if m == nil { return } - m.workflowsActive.WithLabelValues(engine).Inc() + m.workflowsActive.WithLabelValues(boundEngine(engine)).Inc() } func observeWithExemplar(ctx context.Context, observer prometheus.Observer, val float64) { @@ -670,16 +806,57 @@ func observeWithExemplar(ctx context.Context, observer prometheus.Observer, val observer.Observe(val) } -func (m *Metrics) RecordWorkflowEnd(ctx context.Context, engine, result string, duration time.Duration) { +func (m *Metrics) RecordWorkflowEnd( + ctx context.Context, + engine, result, failureClass, reason string, + duration time.Duration, +) { + if m == nil { + return + } + m.RecordWorkflowExecutionEnd(ctx, engine, result, duration) + m.RecordWorkflowTerminal(engine, result, failureClass, reason) +} + +func (m *Metrics) RecordWorkflowExecutionEnd( + ctx context.Context, + engine, result string, + duration time.Duration, +) { if m == nil { return } + engine = boundEngine(engine) + result = boundWorkflowResult(result) m.workflowsActive.WithLabelValues(engine).Dec() - m.workflowsTotal.WithLabelValues(engine, result).Inc() observer := m.workflowDuration.WithLabelValues(engine, result) observeWithExemplar(ctx, observer, duration.Seconds()) } +func (m *Metrics) RecordWorkflowTerminal(engine, result, failureClass, reason string) { + if m == nil { + return + } + engine = boundEngine(engine) + result = boundWorkflowResult(result) + failureClass = boundFailureClass(failureClass) + reason = boundFailureReason(reason) + m.workflowsTotal.WithLabelValues(engine, result).Inc() + m.workflowTerminations.WithLabelValues(engine, result, failureClass, reason).Inc() +} + +func (m *Metrics) RecordWorkflowStartupDelay( + ctx context.Context, + engine, placement string, + d time.Duration, +) { + if m == nil || d < 0 { + return + } + observer := m.workflowStartupDelay.WithLabelValues(boundEngine(engine), boundPlacement(placement)) + observeWithExemplar(ctx, observer, d.Seconds()) +} + func (m *Metrics) RecordStepStart(engine string) { if m == nil { return @@ -773,14 +950,17 @@ func (m *Metrics) RecordMillPlacementAdmission(allowed bool, reason string) { if allowed { status = "allowed" } - m.millPlacementAdmission.WithLabelValues(status, reason).Inc() + m.millPlacementAdmission.WithLabelValues(status, boundMillPlacementReason(reason)).Inc() } -func (m *Metrics) RecordMillPlacementResult(result string) { - if m == nil { +func (m *Metrics) RecordMillPlacementResult(ctx context.Context, engine, result string, d time.Duration) { + if m == nil || d < 0 { return } + engine = boundEngine(engine) + result = boundMillPlacementResult(result) m.millPlacementResults.WithLabelValues(result).Inc() + observeWithExemplar(ctx, m.millPlacementWait.WithLabelValues(engine, result), d.Seconds()) } func (m *Metrics) RecordMillReconnect() { @@ -790,10 +970,14 @@ func (m *Metrics) RecordMillReconnect() { m.millReconnects.Inc() } -func (m *Metrics) RegisterExecutorGauges(reservations, jobs, outboxBytes func() float64) { +func (m *Metrics) RegisterExecutorGauges(seats, reservations, jobs, outboxBytes func() float64) { if m == nil { return } + m.reg.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "spindle_executor_seats", + Help: "Configured executor seat capacity.", + }, seats)) m.reg.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{ Name: "spindle_executor_reservations_active", Help: "Current number of seat reservations on the executor.", diff --git a/spindle/observability/metrics_test.go b/spindle/observability/metrics_test.go index 11d72707..c404a65c 100644 --- a/spindle/observability/metrics_test.go +++ b/spindle/observability/metrics_test.go @@ -15,6 +15,7 @@ import ( "time" "github.com/go-chi/chi/v5" + "github.com/prometheus/client_golang/prometheus" dto "github.com/prometheus/client_model/go" "go.opentelemetry.io/otel" sdktrace "go.opentelemetry.io/otel/sdk/trace" @@ -667,7 +668,7 @@ func BenchmarkRecordWorkflow(b *testing.B) { b.ResetTimer() for i := 0; i < b.N; i++ { m.RecordWorkflowStart("nixery") - m.RecordWorkflowEnd(context.Background(), "nixery", "success", 123*time.Millisecond) + m.RecordWorkflowEnd(context.Background(), "nixery", "success", "none", "success", 123*time.Millisecond) } } @@ -729,7 +730,7 @@ func TestRecordWorkflowEnd_Exemplar(t *testing.T) { ctx, span := Tracer().Start(context.Background(), "test-workflow") defer span.End() - m.RecordWorkflowEnd(ctx, "nixery", "success", 123*time.Millisecond) + m.RecordWorkflowEnd(ctx, "nixery", "success", "none", "success", 123*time.Millisecond) m.RecordStepEnd(ctx, "nixery", "success", 45*time.Millisecond) families, err := m.Registry().Gather() @@ -786,3 +787,123 @@ func TestRecordWorkflowEnd_Exemplar(t *testing.T) { } } } + +func TestWorkflowQueueAndExecutorMetrics(t *testing.T) { + m := NewMetrics() + m.RecordWorkflowStart("microvm") + m.RecordWorkflowEnd( + context.Background(), + "microvm", + "failure", + "user", + "command_failed", + 15*time.Minute, + ) + m.RecordWorkflowTerminal("arbitrary", "invalid_status", "did:plc:123", "dial tcp 10.0.0.1:5432: connection refused") + m.RecordJobDequeueLatency(context.Background(), 20*time.Minute) + m.RecordWorkflowStartupDelay(context.Background(), "microvm", "mill", 12*time.Minute) + m.RecordQuotaWait("workflow", "workflows", time.Second) + m.RecordMillPlacementResult(context.Background(), "microvm", "success", 11*time.Minute) + m.RegisterExecutorGauges( + func() float64 { return 8 }, + func() float64 { return 2 }, + func() float64 { return 5 }, + func() float64 { return 1024 }, + ) + + families := gather(t, m) + workflows := families["spindle_workflows_total"] + if workflows == nil { + t.Fatal("spindle_workflows_total missing") + } + for _, metric := range workflows.GetMetric() { + if got := len(metric.GetLabel()); got != 2 { + t.Fatalf("spindle_workflows_total label count = %d, want existing engine/result schema", got) + } + } + + terminations := families["spindle_workflow_terminations_total"] + if terminations == nil { + t.Fatal("spindle_workflow_terminations_total missing") + } + if len(terminations.GetMetric()) != 2 { + t.Fatalf("spindle_workflow_terminations_total series = %d, want 2", len(terminations.GetMetric())) + } + var sawClassified, sawBounded bool + for _, metric := range terminations.GetMetric() { + switch labelValue(metric, "engine") { + case "microvm": + sawClassified = labelValue(metric, "result") == "failure" && + labelValue(metric, "failure_class") == "user" && + labelValue(metric, "reason") == "command_failed" + case "unknown": + sawBounded = labelValue(metric, "result") == "unknown" && + labelValue(metric, "failure_class") == "unknown" && + labelValue(metric, "reason") == "unknown" + } + } + if !sawClassified || !sawBounded { + t.Fatalf("workflow classification: classified=%t bounded=%t", sawClassified, sawBounded) + } + + for _, name := range []string{ + "spindle_workflow_duration_seconds", + "spindle_job_dequeue_latency_seconds", + "spindle_workflow_startup_delay_seconds", + "spindle_mill_placement_wait_seconds", + } { + family := families[name] + if family == nil || len(family.GetMetric()) == 0 { + t.Fatalf("%s missing", name) + } + histogram := family.GetMetric()[0].GetHistogram() + if histogram == nil { + t.Fatalf("%s is not a histogram", name) + } + maxBucket := float64(0) + for _, bucket := range histogram.GetBucket() { + if bucket.GetUpperBound() > maxBucket { + maxBucket = bucket.GetUpperBound() + } + } + + wantMin := float64(7200) + if name != "spindle_workflow_duration_seconds" { + wantMin = 86400 + } + if maxBucket < wantMin { + t.Fatalf("%s max finite bucket = %v, want at least %v", name, maxBucket, wantMin) + } + } + workflowHistogram := families["spindle_workflow_duration_seconds"].GetMetric()[0].GetHistogram() + for _, required := range prometheus.DefBuckets { + found := false + for _, bucket := range workflowHistogram.GetBucket() { + if bucket.GetUpperBound() == required { + found = true + break + } + } + if !found { + t.Fatalf("workflow histogram dropped deployed bucket %v", required) + } + } + quotaHistogram := families["spindle_quota_wait_duration_seconds"].GetMetric()[0].GetHistogram() + for _, required := range prometheus.ExponentialBuckets(0.1, 3, 8) { + found := false + for _, bucket := range quotaHistogram.GetBucket() { + if bucket.GetUpperBound() == required { + found = true + break + } + } + if !found { + t.Fatalf("quota wait histogram dropped deployed bucket %v", required) + } + } + + seats := families["spindle_executor_seats"] + if seats == nil || len(seats.GetMetric()) != 1 || seats.GetMetric()[0].GetGauge().GetValue() != 8 { + t.Fatalf("spindle_executor_seats = %v, want 8", seats) + } +} diff --git a/spindle/quota/manager.go b/spindle/quota/manager.go index 8affafec..5f5d22a6 100644 --- a/spindle/quota/manager.go +++ b/spindle/quota/manager.go @@ -11,6 +11,8 @@ import ( "time" ) +var ErrDenied = errors.New("quota denied permanently") + type waiter struct { ctx context.Context req ReserveRequest @@ -456,7 +458,7 @@ func (m *Manager) Acquire(ctx context.Context, req ReserveRequest) (Lease, error m.mu.Unlock() m.lifecycleMu.Unlock() m.recordDecision(string(req.Kind), res.Resource, false, false, res.Reason) - return nil, fmt.Errorf("quota denied permanently: %s", res.Reason) + return nil, fmt.Errorf("%w: %s", ErrDenied, res.Reason) } w := &waiter{ @@ -704,7 +706,7 @@ func (m *Manager) processQueue() { } else { if !res.Temporary { select { - case w.errChan <- fmt.Errorf("quota denied permanently: %s", res.Reason): + case w.errChan <- fmt.Errorf("%w: %s", ErrDenied, res.Reason): default: } m.recordDecision(string(w.req.Kind), res.Resource, false, false, res.Reason) diff --git a/spindle/server.go b/spindle/server.go index dde9549c..6ce8c946 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -1291,6 +1291,13 @@ func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { } jobCtx := observability.ExtractFromTraceparentAndTracestate(ctx, job.Traceparent, job.Tracestate) + if job.CreatedAtNs > 0 { + queueDelay := time.Since(time.Unix(0, job.CreatedAtNs)) + if queueDelay < 0 { + queueDelay = 0 + } + s.metrics.RecordJobDequeueLatency(jobCtx, queueDelay) + } jobCtx, span := observability.Tracer().Start(jobCtx, "job.run") if span.IsRecording() { @@ -1349,6 +1356,12 @@ func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) + s.metrics.RecordWorkflowTerminal( + fmt.Sprint(w.Engine), + "failure", + string(engine.FailureClassUser), + string(engine.FailureReasonWorkflowInvalid), + ) continue } @@ -1359,6 +1372,13 @@ func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("init workflow: %s", err), -1, s.n) + failureClass, failureReason := engine.FailureAttribution("failure", err) + s.metrics.RecordWorkflowTerminal( + fmt.Sprint(w.Engine), + "failure", + failureClass, + failureReason, + ) continue } ewf.RunID = fmt.Sprintf("%d", job.Id) @@ -1379,7 +1399,7 @@ func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) - if err := s.db.EnqueueJob( + if err := s.db.EnqueueJobWithPending( s.rootCtx, repoDid.String(), pipelineId, @@ -1387,28 +1407,16 @@ func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl t tpl, traceparent, tracestate, + s.n, ); err != nil { return fmt.Errorf("failed to enqueue durable job: %w", err) } s.l.Info("pipeline enqueued successfully to db", "id", pipelineId) - // wake up an idle worker to pick up more jobs if any select { case s.jobWake <- struct{}{}: default: } - // pipelines visible from now on, they are sitting in queue - for _, w := range tpl.Workflows { - if w == nil { - continue - } - if err := s.db.StatusPending(models.WorkflowId{ - PipelineId: pipelineId, - Name: w.Name, - }, s.n); err != nil { - return fmt.Errorf("db.StatusPending: %w", err) - } - } return nil }