From b95be5f74d6748415537dc684b143ad3f458c819 Mon Sep 17 00:00:00 2001 From: seemeroland Date: Fri, 17 Jul 2026 21:12:56 +0000 Subject: [PATCH] Add sandbox startup failure observability and improve error messages - Add oz_worker_task_sandbox_startup_failures_total metric that tracks task failures occurring within the first 5 minutes of task assignment. This counter is labeled by failure reason and is primed at startup so dashboards can query it immediately. A spike indicates systemic sandbox provisioning or image-pull issues rather than long-running task errors. - Add RecordSandboxStartupFailure(reason) helper in the metrics package so worker code can increment the new counter when a startup-window failure is detected. - In executeTask, check whether a task failure occurred within the sandboxStartupWindow (5 min). When it did, emit a Warn-level log entry with the task ID, elapsed duration, failure phase, and reason, plus a task.sandbox_startup_failure trace event with the same attributes. - Improve userFacingTaskError to produce actionable, reason-specific messages for backendFailureError values instead of always falling through to the raw 'Failed to execute task: ...' default. The new messages cover: active_deadline, container_create/start, sidecar_prep, image_pull, unschedulable, and container_oom. - Update README metric catalog and sample alerts with the new counter and recommended PromQL alert queries for sandbox startup failures. Co-Authored-By: Oz --- README.md | 13 ++++++ internal/metrics/metrics.go | 64 +++++++++++++++++++--------- internal/metrics/metrics_test.go | 28 +++++++++++++ internal/worker/errors.go | 22 +++++++++- internal/worker/worker.go | 17 ++++++++ internal/worker/worker_test.go | 72 ++++++++++++++++++++++++++++++++ 6 files changed, 194 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 0c5d261..9b5732d 100644 --- a/README.md +++ b/README.md @@ -365,6 +365,13 @@ shows up as a distinct series. classification for task failures, such as `phase="backend"` with `reason="image_pull"`, `reason="unschedulable"`, or `reason="container_oom"`. +- `oz_worker_task_sandbox_startup_failures_total{reason}` (counter): task + failures that occurred within the first 5 minutes of task assignment, + labeled by failure reason. A sustained increase indicates systemic sandbox + startup issues such as slow image pulls, cluster resource pressure, or + admission failures. This counter is a subset of `oz_worker_task_failures_total` + and is intended to help triage startup-specific regressions separately from + long-running task failures. - `oz_worker_websocket_reconnects_total{reason}` (counter): reconnect attempts; spikes indicate flapping workers. - `oz_worker_info{version,backend,worker_id}` (gauge, value `1`): build and @@ -386,6 +393,12 @@ Direct mappings for the questions enterprise operators most commonly ask: `sum by (phase, reason) (rate(oz_worker_task_failures_total[5m]))` - **Reconnect storms:** `sum(rate(oz_worker_websocket_reconnects_total[5m])) > 0.1` +- **Sandbox startup failures:** + `sum(rate(oz_worker_task_sandbox_startup_failures_total[5m])) > 0` + Broken down by reason: + `sum by (reason) (rate(oz_worker_task_sandbox_startup_failures_total[5m]))` + Alert when `reason="active_deadline"` or `reason="image_pull"` are non-zero + to catch systemic provisioning regressions early. ## License diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 8a347d4..bd3906a 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -67,16 +67,17 @@ type Config struct { // a single MeterProvider. Helpers read this struct atomically so that Init can // hot-swap from the no-op set to the SDK-backed set without locking. type instruments struct { - connected metric.Int64Gauge - tasksActive metric.Int64UpDownCounter - tasksMaxConcurrent metric.Int64Gauge - tasksClaimed metric.Int64Counter - tasksRejected metric.Int64Counter - tasksCompleted metric.Int64Counter - taskDuration metric.Float64Histogram - taskFailures metric.Int64Counter - wsReconnects metric.Int64Counter - workerInfo metric.Int64Gauge + connected metric.Int64Gauge + tasksActive metric.Int64UpDownCounter + tasksMaxConcurrent metric.Int64Gauge + tasksClaimed metric.Int64Counter + tasksRejected metric.Int64Counter + tasksCompleted metric.Int64Counter + taskDuration metric.Float64Histogram + taskFailures metric.Int64Counter + wsReconnects metric.Int64Counter + workerInfo metric.Int64Gauge + sandboxStartupFailures metric.Int64Counter } // activeInstruments is the current instrument set. It always points to a @@ -295,6 +296,11 @@ func primeInstruments(ctx context.Context, set *instruments) { ) } } + for _, reason := range taskFailureReasons { + set.sandboxStartupFailures.Add(ctx, 0, + metric.WithAttributes(attribute.String("reason", reason)), + ) + } for _, r := range []string{WSReconnectReasonDialFailed, WSReconnectReasonRemoteClose} { set.wsReconnects.Add(ctx, 0, metric.WithAttributes(attribute.String("reason", r)), @@ -394,17 +400,25 @@ func buildInstruments(m metric.Meter) (*instruments, error) { if err != nil { return nil, err } + sandboxStartupFailures, err := m.Int64Counter( + "oz_worker_task_sandbox_startup_failures_total", + metric.WithDescription("Task failures that occurred within the sandbox startup window, labeled by failure reason. A spike indicates systemic sandbox provisioning or image-pull issues."), + ) + if err != nil { + return nil, err + } return &instruments{ - connected: connected, - tasksActive: tasksActive, - tasksMaxConcurrent: tasksMaxConcurrent, - tasksClaimed: tasksClaimed, - tasksRejected: tasksRejected, - tasksCompleted: tasksCompleted, - taskDuration: taskDuration, - taskFailures: taskFailures, - wsReconnects: wsReconnects, - workerInfo: workerInfo, + connected: connected, + tasksActive: tasksActive, + tasksMaxConcurrent: tasksMaxConcurrent, + tasksClaimed: tasksClaimed, + tasksRejected: tasksRejected, + tasksCompleted: tasksCompleted, + taskDuration: taskDuration, + taskFailures: taskFailures, + wsReconnects: wsReconnects, + workerInfo: workerInfo, + sandboxStartupFailures: sandboxStartupFailures, }, nil } @@ -489,6 +503,16 @@ func RecordWebsocketReconnect(reason string) { ) } +// RecordSandboxStartupFailure records a task failure that occurred within the +// sandbox startup window. reason should be one of the TaskFailureReason* constants. +// A sustained increase in this counter indicates systemic sandbox startup issues +// such as slow image pulls, cluster resource pressure, or admission failures. +func RecordSandboxStartupFailure(reason string) { + current().sandboxStartupFailures.Add(context.Background(), 1, + metric.WithAttributes(attribute.String("reason", reason)), + ) +} + // SetWorkerInfo emits a constant gauge with build metadata. It is intended // to be called once, immediately after Init. func SetWorkerInfo(version, backend, workerID string) { diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 9b9f082..40e689a 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -85,6 +85,7 @@ func TestHelpersSafeBeforeInit(t *testing.T) { RecordTaskCompleted(TaskResultFailed, 1*time.Second) RecordTaskCompleted(TaskResultCancelled, 500*time.Millisecond) RecordTaskFailure(TaskFailurePhaseBackend, TaskFailureReasonImagePull) + RecordSandboxStartupFailure(TaskFailureReasonActiveDeadline) RecordWebsocketReconnect("dial_failed") SetWorkerInfo("v0.0.0", "docker", "test") ctx, span := StartTaskSpan(context.Background(), "task-1", "test task") @@ -252,6 +253,7 @@ func TestPrimeInstrumentsExposesAllSeriesAtStartup(t *testing.T) { "oz_worker_tasks_completed_total", "oz_worker_task_failures_total", "oz_worker_websocket_reconnects_total", + "oz_worker_task_sandbox_startup_failures_total", } for _, name := range want { findMetric(t, rm, name) // fails the test if missing @@ -336,6 +338,32 @@ func TestPrimeInstrumentsExposesAllSeriesAtStartup(t *testing.T) { } } +func TestRecordSandboxStartupFailureTagsReason(t *testing.T) { + reader := withTestReader(t, Config{WorkerID: "w1", Backend: "kubernetes"}) + + RecordSandboxStartupFailure(TaskFailureReasonActiveDeadline) + RecordSandboxStartupFailure(TaskFailureReasonActiveDeadline) + RecordSandboxStartupFailure(TaskFailureReasonImagePull) + + rm := collect(t, reader) + startupFailures := findMetric(t, rm, "oz_worker_task_sandbox_startup_failures_total") + sum, ok := startupFailures.Data.(metricdata.Sum[int64]) + if !ok { + t.Fatalf("expected Sum[int64], got %T", startupFailures.Data) + } + byReason := map[string]int64{} + for _, dp := range sum.DataPoints { + v, _ := dp.Attributes.Value("reason") + byReason[v.AsString()] = dp.Value + } + if got := byReason[TaskFailureReasonActiveDeadline]; got != 2 { + t.Errorf("active_deadline count = %d, want 2", got) + } + if got := byReason[TaskFailureReasonImagePull]; got != 1 { + t.Errorf("image_pull count = %d, want 1", got) + } +} + func TestShouldInitTracesRequiresExporter(t *testing.T) { t.Setenv("OTEL_TRACES_EXPORTER", "") if shouldInitTraces() { diff --git a/internal/worker/errors.go b/internal/worker/errors.go index 934b0e0..7f24df7 100644 --- a/internal/worker/errors.go +++ b/internal/worker/errors.go @@ -44,8 +44,9 @@ func taskFailureLabels(err error) (phase, reason string) { } // userFacingTaskError returns a user-friendly error message for a task execution -// failure. Well-known infrastructure errors (context cancellation, deadline exceeded) -// are translated into clear, actionable messages instead of exposing raw Go error strings. +// failure. Well-known infrastructure errors (context cancellation, deadline exceeded, +// and specific backend failure reasons) are translated into clear, actionable messages +// instead of exposing raw Go error strings. func userFacingTaskError(err error) string { switch { case errors.Is(err, context.Canceled): @@ -53,6 +54,23 @@ func userFacingTaskError(err error) string { case errors.Is(err, context.DeadlineExceeded): return "The task exceeded its maximum allowed execution time and was terminated. Consider breaking the task into smaller steps or increasing the timeout." default: + var failure *backendFailureError + if errors.As(err, &failure) { + switch failure.reason { + case metrics.TaskFailureReasonActiveDeadline: + return "The agent sandbox did not complete before the job deadline was exceeded. This may indicate slow container image pulls, cluster resource pressure, or sandbox startup delays — please try again." + case metrics.TaskFailureReasonContainerCreate, metrics.TaskFailureReasonContainerStart: + return "The agent sandbox failed to start. This may indicate a container image issue, insufficient cluster resources, or a configuration problem — please try again." + case metrics.TaskFailureReasonSidecarPrep: + return "The agent sandbox failed to prepare its required dependencies. This may indicate a container image issue or a network connectivity problem — please try again." + case metrics.TaskFailureReasonImagePull: + return "The agent sandbox could not pull its container image. Verify the image is accessible from the worker and try again." + case metrics.TaskFailureReasonUnschedulable: + return "The agent sandbox could not be scheduled due to insufficient cluster resources. Check available capacity and try again." + case metrics.TaskFailureReasonContainerOOM: + return "The agent sandbox ran out of memory and was terminated. Consider breaking the task into smaller steps or requesting a larger runner." + } + } return fmt.Sprintf("Failed to execute task: %v", err) } } diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 106cca0..c63054c 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -31,6 +31,12 @@ const ( BackendShutdownTimeout = 10 * time.Second warpServerRootURLEnv = "WARP_SERVER_ROOT_URL" + + // sandboxStartupWindow is the duration after task assignment within which a + // task failure is classified as a sandbox startup failure. Failures that + // occur within this window are likely caused by slow image pulls, container + // scheduling delays, or admission failures rather than task-execution errors. + sandboxStartupWindow = 5 * time.Minute ) type Config struct { @@ -613,6 +619,17 @@ func (w *Worker) executeTask(ctx context.Context, taskCancel context.CancelFunc, span.RecordError(err) span.SetStatus(codes.Error, reason) log.Errorf(ctx, "Task execution failed: taskID=%s, error=%v", taskID, err) + sandboxStartupDuration := time.Since(receivedAt) + if sandboxStartupDuration < sandboxStartupWindow { + log.Warnf(ctx, "Task failed within sandbox startup window (taskID=%s, duration=%v, failure.phase=%s, failure.reason=%s): possible sandbox startup issue", + taskID, sandboxStartupDuration.Round(time.Millisecond), phase, reason) + metrics.RecordSandboxStartupFailure(reason) + metrics.AddTaskEvent(ctx, "task.sandbox_startup_failure", + attribute.String("failure.phase", phase), + attribute.String("failure.reason", reason), + attribute.Float64("startup.duration_seconds", sandboxStartupDuration.Seconds()), + ) + } if statusErr := w.sendTaskFailed(taskID, userFacingTaskError(err)); statusErr != nil { log.Errorf(ctx, "Failed to send task failed message: %v", statusErr) } diff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go index 6f7afc0..cc22072 100644 --- a/internal/worker/worker_test.go +++ b/internal/worker/worker_test.go @@ -324,6 +324,78 @@ func TestUserFacingTaskError(t *testing.T) { err: errors.New("boom"), want: "Failed to execute task: boom", }, + { + name: "active deadline backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonActiveDeadline, + errors.New("job deadline exceeded"), + ), + want: "The agent sandbox did not complete before the job deadline was exceeded. This may indicate slow container image pulls, cluster resource pressure, or sandbox startup delays — please try again.", + }, + { + name: "container start backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonContainerStart, + errors.New("failed to start container"), + ), + want: "The agent sandbox failed to start. This may indicate a container image issue, insufficient cluster resources, or a configuration problem — please try again.", + }, + { + name: "container create backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonContainerCreate, + errors.New("failed to create container"), + ), + want: "The agent sandbox failed to start. This may indicate a container image issue, insufficient cluster resources, or a configuration problem — please try again.", + }, + { + name: "sidecar prep backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonSidecarPrep, + errors.New("sidecar prep failed"), + ), + want: "The agent sandbox failed to prepare its required dependencies. This may indicate a container image issue or a network connectivity problem — please try again.", + }, + { + name: "image pull backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonImagePull, + errors.New("pull failed"), + ), + want: "The agent sandbox could not pull its container image. Verify the image is accessible from the worker and try again.", + }, + { + name: "unschedulable backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonUnschedulable, + errors.New("pod unschedulable"), + ), + want: "The agent sandbox could not be scheduled due to insufficient cluster resources. Check available capacity and try again.", + }, + { + name: "container OOM backend failure", + err: newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonContainerOOM, + errors.New("oom killed"), + ), + want: "The agent sandbox ran out of memory and was terminated. Consider breaking the task into smaller steps or requesting a larger runner.", + }, + { + name: "wrapped backend failure preserves reason-specific message", + err: fmt.Errorf("outer: %w", newBackendFailure( + metrics.TaskFailurePhaseBackend, + metrics.TaskFailureReasonImagePull, + errors.New("pull failed"), + )), + want: "The agent sandbox could not pull its container image. Verify the image is accessible from the worker and try again.", + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) {