diff --git a/internal/alerts/store.go b/internal/alerts/store.go index bbaa61ea..14d066e0 100644 --- a/internal/alerts/store.go +++ b/internal/alerts/store.go @@ -22,6 +22,7 @@ import ( "strings" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" ) @@ -59,6 +60,8 @@ CREATE INDEX IF NOT EXISTS alerts_created_at ON alerts (created_at DESC); ALTER TABLE alerts ADD COLUMN IF NOT EXISTS fingerprint TEXT NOT NULL DEFAULT ''; ALTER TABLE alerts ADD COLUMN IF NOT EXISTS occurrence_count INTEGER NOT NULL DEFAULT 1; ALTER TABLE alerts ADD COLUMN IF NOT EXISTS last_seen_at TIMESTAMPTZ; +ALTER TABLE alerts ADD COLUMN IF NOT EXISTS trace_id TEXT NOT NULL DEFAULT ''; +ALTER TABLE alerts ADD COLUMN IF NOT EXISTS workflow_name TEXT NOT NULL DEFAULT ''; DROP INDEX IF EXISTS alerts_fingerprint_open; CREATE UNIQUE INDEX IF NOT EXISTS alerts_tenant_fingerprint_open @@ -75,6 +78,25 @@ CREATE TABLE IF NOT EXISTS alert_evidence ( CREATE INDEX IF NOT EXISTS alert_evidence_alert_id ON alert_evidence (alert_id); ` +// alertColumns is the single SELECT/RETURNING column list; scanAlert is its +// matching Scan target list. Keep them in lockstep. +const alertColumns = `id, source, type, tenant, service, summary, severity, status, + analysis, proposed_fix, pr_url, pr_number, confidence, + created_at, updated_at, resolved_at, acknowledged_by, acknowledged_at, + fingerprint, occurrence_count, last_seen_at, trace_id, workflow_name` + +func scanAlert(row pgx.Row, a *Alert) error { + if err := row.Scan( + &a.ID, &a.Source, &a.Type, &a.Tenant, &a.Service, &a.Summary, &a.Severity, &a.Status, + &a.Analysis, &a.ProposedFix, &a.PRURL, &a.PRNumber, &a.Confidence, + &a.CreatedAt, &a.UpdatedAt, &a.ResolvedAt, &a.AcknowledgedBy, &a.AcknowledgedAt, + &a.Fingerprint, &a.OccurrenceCount, &a.LastSeenAt, &a.TraceID, &a.WorkflowName, + ); err != nil { + return err + } + return nil +} + // Store is a PostgreSQL-backed alert store. type Store struct { pool *pgxpool.Pool @@ -110,30 +132,31 @@ func (s *Store) Create(ctx context.Context, a *Alert) (*Alert, error) { } stored := &Alert{} - err := s.pool.QueryRow(ctx, + row := s.pool.QueryRow(ctx, `INSERT INTO alerts (id, source, type, tenant, service, summary, severity, status, analysis, proposed_fix, pr_url, pr_number, confidence, created_at, updated_at, resolved_at, - acknowledged_by, acknowledged_at, fingerprint, occurrence_count, last_seen_at) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21) + acknowledged_by, acknowledged_at, fingerprint, occurrence_count, last_seen_at, + trace_id, workflow_name) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23) ON CONFLICT (tenant, fingerprint) WHERE fingerprint <> '' AND status NOT IN ('resolved','suppressed') DO UPDATE SET occurrence_count = alerts.occurrence_count + 1, last_seen_at = EXCLUDED.updated_at, - updated_at = EXCLUDED.updated_at - RETURNING id, source, type, tenant, service, summary, severity, status, - analysis, proposed_fix, pr_url, pr_number, confidence, - created_at, updated_at, resolved_at, acknowledged_by, acknowledged_at, - fingerprint, occurrence_count, last_seen_at`, + updated_at = EXCLUDED.updated_at, + trace_id = CASE WHEN EXCLUDED.trace_id <> '' THEN EXCLUDED.trace_id ELSE alerts.trace_id END, + -- trace_id and workflow_name describe one run and move + -- together: a trace reported without a name clears the + -- stale name rather than pairing it with the new trace. + workflow_name = CASE + WHEN EXCLUDED.trace_id <> '' THEN EXCLUDED.workflow_name + WHEN alerts.trace_id = '' AND EXCLUDED.workflow_name <> '' THEN EXCLUDED.workflow_name + ELSE alerts.workflow_name END + RETURNING `+alertColumns, a.ID, a.Source, a.Type, a.Tenant, a.Service, a.Summary, a.Severity, a.Status, a.Analysis, a.ProposedFix, a.PRURL, a.PRNumber, a.Confidence, a.CreatedAt, a.UpdatedAt, a.ResolvedAt, a.AcknowledgedBy, a.AcknowledgedAt, - a.Fingerprint, a.OccurrenceCount, a.LastSeenAt, - ).Scan( - &stored.ID, &stored.Source, &stored.Type, &stored.Tenant, &stored.Service, &stored.Summary, - &stored.Severity, &stored.Status, &stored.Analysis, &stored.ProposedFix, &stored.PRURL, - &stored.PRNumber, &stored.Confidence, &stored.CreatedAt, &stored.UpdatedAt, &stored.ResolvedAt, - &stored.AcknowledgedBy, &stored.AcknowledgedAt, &stored.Fingerprint, &stored.OccurrenceCount, - &stored.LastSeenAt, + a.Fingerprint, a.OccurrenceCount, a.LastSeenAt, a.TraceID, a.WorkflowName, ) + err := scanAlert(row, stored) if err != nil { var pgErr *pgconn.PgError if errors.As(err, &pgErr) && pgErr.Code == "23505" && pgErr.ConstraintName == "alerts_pkey" { @@ -185,10 +208,12 @@ func (s *Store) Update(ctx context.Context, a *Alert) error { _, err := s.pool.Exec(ctx, `UPDATE alerts SET status=$2, analysis=$3, proposed_fix=$4, pr_url=$5, pr_number=$6, - confidence=$7, updated_at=$8, resolved_at=$9, acknowledged_by=$10, acknowledged_at=$11 + confidence=$7, updated_at=$8, resolved_at=$9, acknowledged_by=$10, acknowledged_at=$11, + trace_id=CASE WHEN $12::text <> '' THEN $12::text ELSE trace_id END WHERE id=$1`, a.ID, a.Status, a.Analysis, a.ProposedFix, a.PRURL, a.PRNumber, a.Confidence, a.UpdatedAt, a.ResolvedAt, a.AcknowledgedBy, a.AcknowledgedAt, + a.TraceID, ) if err != nil { return fmt.Errorf("alerts store: update: %w", err) @@ -211,17 +236,8 @@ func (s *Store) Update(ctx context.Context, a *Alert) error { // Get returns a single alert with evidence. func (s *Store) Get(ctx context.Context, id string) (*Alert, error) { a := &Alert{} - err := s.pool.QueryRow(ctx, - `SELECT id, source, type, tenant, service, summary, severity, status, - analysis, proposed_fix, pr_url, pr_number, confidence, - created_at, updated_at, resolved_at, acknowledged_by, acknowledged_at, - fingerprint, occurrence_count, last_seen_at - FROM alerts WHERE id=$1`, id).Scan( - &a.ID, &a.Source, &a.Type, &a.Tenant, &a.Service, &a.Summary, &a.Severity, &a.Status, - &a.Analysis, &a.ProposedFix, &a.PRURL, &a.PRNumber, &a.Confidence, - &a.CreatedAt, &a.UpdatedAt, &a.ResolvedAt, &a.AcknowledgedBy, &a.AcknowledgedAt, - &a.Fingerprint, &a.OccurrenceCount, &a.LastSeenAt, - ) + err := scanAlert(s.pool.QueryRow(ctx, + `SELECT `+alertColumns+` FROM alerts WHERE id=$1`, id), a) if err != nil { return nil, fmt.Errorf("alerts store: get: %w", err) } @@ -237,17 +253,8 @@ func (s *Store) GetByPrefix(ctx context.Context, prefix string) (*Alert, error) } a := &Alert{} - err := s.pool.QueryRow(ctx, - `SELECT id, source, type, tenant, service, summary, severity, status, - analysis, proposed_fix, pr_url, pr_number, confidence, - created_at, updated_at, resolved_at, acknowledged_by, acknowledged_at, - fingerprint, occurrence_count, last_seen_at - FROM alerts WHERE id LIKE $1 ORDER BY created_at DESC LIMIT 1`, prefix+"%").Scan( - &a.ID, &a.Source, &a.Type, &a.Tenant, &a.Service, &a.Summary, &a.Severity, &a.Status, - &a.Analysis, &a.ProposedFix, &a.PRURL, &a.PRNumber, &a.Confidence, - &a.CreatedAt, &a.UpdatedAt, &a.ResolvedAt, &a.AcknowledgedBy, &a.AcknowledgedAt, - &a.Fingerprint, &a.OccurrenceCount, &a.LastSeenAt, - ) + err := scanAlert(s.pool.QueryRow(ctx, + `SELECT `+alertColumns+` FROM alerts WHERE id LIKE $1 ORDER BY created_at DESC LIMIT 1`, prefix+"%"), a) if err != nil { return nil, fmt.Errorf("alerts store: get by prefix: %w", err) } @@ -334,11 +341,7 @@ func (s *Store) List(ctx context.Context, f ListFilter) ([]Alert, error) { argIdx++ } - query := `SELECT id, source, type, tenant, service, summary, severity, status, - analysis, proposed_fix, pr_url, pr_number, confidence, - created_at, updated_at, resolved_at, acknowledged_by, acknowledged_at, - fingerprint, occurrence_count, last_seen_at - FROM alerts` + query := `SELECT ` + alertColumns + ` FROM alerts` if len(conditions) > 0 { query += " WHERE " + strings.Join(conditions, " AND ") } @@ -360,12 +363,7 @@ func (s *Store) List(ctx context.Context, f ListFilter) ([]Alert, error) { var alerts []Alert for rows.Next() { var a Alert - if err := rows.Scan( - &a.ID, &a.Source, &a.Type, &a.Tenant, &a.Service, &a.Summary, &a.Severity, &a.Status, - &a.Analysis, &a.ProposedFix, &a.PRURL, &a.PRNumber, &a.Confidence, - &a.CreatedAt, &a.UpdatedAt, &a.ResolvedAt, &a.AcknowledgedBy, &a.AcknowledgedAt, - &a.Fingerprint, &a.OccurrenceCount, &a.LastSeenAt, - ); err != nil { + if err := scanAlert(rows, &a); err != nil { slog.Error("alerts store: scan failed", "error", err) continue } diff --git a/internal/alerts/store_trace_test.go b/internal/alerts/store_trace_test.go new file mode 100644 index 00000000..d5ee1d79 --- /dev/null +++ b/internal/alerts/store_trace_test.go @@ -0,0 +1,161 @@ +// Copyright 2025 MCTL Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package alerts + +import ( + "context" + "testing" +) + +const ( + traceA = "431486061c652efb376a860d9d656f25" + traceB = "008ad8979d34c0509e74d8062b7f92ca" +) + +// assertTrace fails unless a carries the stored trace id and workflow name. +// trace_source is response-only and set by the handlers, so the store leaves +// it empty. +func assertTrace(t *testing.T, path string, a *Alert, wantTrace, wantWorkflow string) { + t.Helper() + if a.TraceID != wantTrace || a.WorkflowName != wantWorkflow || a.TraceSource != "" { + t.Fatalf("%s: trace_id=%q workflow_name=%q trace_source=%q, want %q / %q / \"\"", + path, a.TraceID, a.WorkflowName, a.TraceSource, wantTrace, wantWorkflow) + } +} + +// mctl-api#585 T1 + T3: every read path returns the stored trace identity. +// A read path that stops selecting trace_id (or a schema that drops it) +// fails here rather than silently serving incidents without their trace. +func TestTraceID_RoundTripsThroughEveryReadPath(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + in := newAlert("argo-wf-trace-1", "workflow_failed:trace:roundtrip") + in.TraceID = traceA + in.WorkflowName = "mctl-agents-implement-dae24112" + + created, err := s.Create(ctx, in) + if err != nil { + t.Fatalf("create: %v", err) + } + assertTrace(t, "Create RETURNING", created, traceA, in.WorkflowName) + + got, err := s.Get(ctx, in.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + assertTrace(t, "Get", got, traceA, in.WorkflowName) + + byPrefix, err := s.GetByPrefix(ctx, "argo-wf-trace") + if err != nil { + t.Fatalf("get by prefix: %v", err) + } + assertTrace(t, "GetByPrefix", byPrefix, traceA, in.WorkflowName) + + list, err := s.List(ctx, ListFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(list) != 1 { + t.Fatalf("list: %d rows, want 1", len(list)) + } + assertTrace(t, "List", &list[0], traceA, in.WorkflowName) + + // Ack/resolve round-trip the row they read through Update: the trace + // must survive that unchanged. + got.Status = StatusAcknowledged + if err := s.Update(ctx, got); err != nil { + t.Fatalf("ack-style update: %v", err) + } + after, err := s.Get(ctx, in.ID) + if err != nil { + t.Fatalf("get after ack: %v", err) + } + assertTrace(t, "Get after ack-style Update", after, traceA, in.WorkflowName) + + // A PATCH carrying a new trace replaces it. Update does not persist + // workflow_name, so the changed value below is dropped, not rejected. + after.TraceID = traceB + after.WorkflowName = "something-else" + if err := s.Update(ctx, after); err != nil { + t.Fatalf("trace update: %v", err) + } + replaced, err := s.Get(ctx, in.ID) + if err != nil { + t.Fatalf("get after trace update: %v", err) + } + assertTrace(t, "Get after trace Update", replaced, traceB, in.WorkflowName) + + // A partial update that does not carry the trace must not blank it. + replaced.TraceID = "" + if err := s.Update(ctx, replaced); err != nil { + t.Fatalf("partial update: %v", err) + } + kept, err := s.Get(ctx, in.ID) + if err != nil { + t.Fatalf("get after partial update: %v", err) + } + assertTrace(t, "Get after Update without trace", kept, traceB, in.WorkflowName) +} + +// An incident created without a trace reads back with no trace and no +// trace_source, so "reported" is never claimed for an empty value. +func TestTraceID_AbsentStaysAbsent(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + if _, err := s.Create(ctx, newAlert("argo-wf-notrace", "workflow_failed:trace:none")); err != nil { + t.Fatalf("create: %v", err) + } + got, err := s.Get(ctx, "argo-wf-notrace") + if err != nil { + t.Fatalf("get: %v", err) + } + assertTrace(t, "Get", got, "", "") +} + +// mctl-api#585 T2: trace_id and workflow_name describe the same run, so a +// dedup hit moves them together. A new valid trace replaces both; a hit +// without a trace keeps both; and while no trace is stored a newer +// workflow_name replaces the older one, so derivation uses the latest run. +func TestTraceID_Dedup(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + occur := func(id, fp, trace, workflow string, wantCount int) *Alert { + t.Helper() + a := newAlert(id, fp) + a.TraceID, a.WorkflowName = trace, workflow + got, err := s.Create(ctx, a) + if err != nil { + t.Fatalf("create %s: %v", id, err) + } + if got.OccurrenceCount != wantCount { + t.Fatalf("create %s: occurrence_count=%d, want %d", id, got.OccurrenceCount, wantCount) + } + return got + } + + const fp = "workflow_failed:trace:dedup" + occur("argo-dedup-1", fp, traceA, "wf-first", 1) + assertTrace(t, "new trace replaces the pair", occur("argo-dedup-2", fp, traceB, "wf-second", 2), traceB, "wf-second") + assertTrace(t, "no trace keeps the pair", occur("argo-dedup-3", fp, "", "wf-third", 3), traceB, "wf-second") + assertTrace(t, "a trace without a name clears the stale name", occur("argo-dedup-4", fp, traceA, "", 4), traceA, "") + + const fpNoTrace = "workflow_failed:trace:dedup-untraced" + occur("argo-untraced-1", fpNoTrace, "", "wf-a", 1) + assertTrace(t, "untraced row takes the newer workflow", occur("argo-untraced-2", fpNoTrace, "", "wf-b", 2), "", "wf-b") + assertTrace(t, "empty workflow keeps the stored one", occur("argo-untraced-3", fpNoTrace, "", "", 3), "", "wf-b") +} diff --git a/internal/alerts/types.go b/internal/alerts/types.go index b1bde837..290c77b4 100644 --- a/internal/alerts/types.go +++ b/internal/alerts/types.go @@ -75,6 +75,18 @@ type Alert struct { OccurrenceCount int `json:"occurrence_count,omitempty"` LastSeenAt *time.Time `json:"last_seen_at,omitempty"` Evidence []Evidence `json:"evidence,omitempty"` + // TraceID is the 32-hex W3C trace id reported for the run that handled + // this incident. WorkflowName is the Argo workflow it came from. Both are + // stored; empty means not reported. They describe the newest occurrence + // and move together on dedup. There is no way to clear a stored trace id + // through the API (absent and invalid values both leave it unchanged), and + // WorkflowName is set only on create or by a dedup hit, never by PATCH. + TraceID string `json:"trace_id,omitempty"` + WorkflowName string `json:"workflow_name,omitempty"` + // TraceSource ("reported" | "workflow_audit") and TraceUnavailable are + // response-only and never persisted. + TraceSource string `json:"trace_source,omitempty"` + TraceUnavailable string `json:"trace_unavailable,omitempty"` } // Evidence attached to an alert. diff --git a/internal/api/handlers_alerts.go b/internal/api/handlers_alerts.go index af1fae5c..d1ca304e 100644 --- a/internal/api/handlers_alerts.go +++ b/internal/api/handlers_alerts.go @@ -23,6 +23,7 @@ import ( "github.com/go-chi/chi/v5" "github.com/mctlhq/mctl-api/internal/alerts" + "github.com/mctlhq/mctl-api/internal/audit" "github.com/mctlhq/mctl-api/internal/auth" "github.com/mctlhq/mctl-api/internal/operations" ) @@ -46,11 +47,28 @@ func (h *Handlers) CreateIncident(w http.ResponseWriter, r *http.Request) { return } - var a alerts.Alert - if err := json.NewDecoder(r.Body).Decode(&a); err != nil { + // The trace fields shadow Alert's so a wrong JSON type for them is + // ignored instead of failing the whole request. + var body struct { + alerts.Alert + TraceID json.RawMessage `json:"trace_id"` + WorkflowName json.RawMessage `json:"workflow_name"` + TraceSource json.RawMessage `json:"trace_source"` + TraceUnavailable json.RawMessage `json:"trace_unavailable"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { writeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) return } + a := body.Alert + a.TraceID = validIncidentTraceID(body.TraceID) + a.WorkflowName = validIncidentWorkflowName(body.WorkflowName) + // Argo producers send no workflow_name, only a summary link. Store the + // name from it so a dedup hit, which keeps the first occurrence's + // summary, still records the newest run and derivation follows it. + if a.WorkflowName == "" && a.Source == incidentSourceArgo { + a.WorkflowName = workflowNameFromSummary(a.Summary) + } if a.ID == "" || a.Source == "" || a.Type == "" || a.Tenant == "" || a.Summary == "" || a.Severity == "" { writeError(w, http.StatusBadRequest, "missing required fields: id, source, type, tenant, summary, severity") @@ -84,6 +102,7 @@ func (h *Handlers) CreateIncident(w http.ResponseWriter, r *http.Request) { return } + labelStoredTrace(stored, traceNotDerived) writeJSON(w, http.StatusCreated, stored) } @@ -186,6 +205,11 @@ func (h *Handlers) ListIncidents(w http.ResponseWriter, r *http.Request) { if items == nil { items = []alerts.Alert{} } + // The list does not run the audit derivation; mark the absence so it is + // never read as "this incident has no trace". + for i := range items { + labelStoredTrace(&items[i], traceNotDerivedOnList) + } writeJSON(w, http.StatusOK, map[string]interface{}{ "items": items, @@ -218,6 +242,14 @@ func (h *Handlers) GetIncident(w http.ResponseWriter, r *http.Request) { return } + var lookup workflowAuditLookup + if h.opts.AuditLog != nil { + lookup = h.opts.AuditLog + } + deriveIncidentTrace(r.Context(), a, lookup, func(e *audit.Entry) bool { + return canReadAuditEntry(user, e) + }) + writeJSON(w, http.StatusOK, a) } @@ -250,12 +282,13 @@ func (h *Handlers) UpdateIncident(w http.ResponseWriter, r *http.Request) { } var patch struct { - Status *string `json:"status"` - Analysis *string `json:"analysis"` - ProposedFix *string `json:"proposed_fix"` - PRURL *string `json:"pr_url"` - PRNumber *int `json:"pr_number"` - Confidence *string `json:"confidence"` + Status *string `json:"status"` + Analysis *string `json:"analysis"` + ProposedFix *string `json:"proposed_fix"` + PRURL *string `json:"pr_url"` + PRNumber *int `json:"pr_number"` + Confidence *string `json:"confidence"` + TraceID json.RawMessage `json:"trace_id"` } if err := json.NewDecoder(r.Body).Decode(&patch); err != nil { writeError(w, http.StatusBadRequest, "invalid JSON: "+err.Error()) @@ -281,11 +314,16 @@ func (h *Handlers) UpdateIncident(w http.ResponseWriter, r *http.Request) { existing.Confidence = *patch.Confidence } + if id := validIncidentTraceID(patch.TraceID); id != "" { + existing.TraceID = id + } + if err := h.opts.AlertStore.Update(r.Context(), existing); err != nil { writeError(w, http.StatusInternalServerError, "failed to update incident: "+err.Error()) return } + labelStoredTrace(existing, traceNotDerived) writeJSON(w, http.StatusOK, existing) } @@ -329,6 +367,7 @@ func (h *Handlers) AcknowledgeIncident(w http.ResponseWriter, r *http.Request) { return } + labelStoredTrace(existing, traceNotDerived) writeJSON(w, http.StatusOK, existing) } @@ -377,6 +416,7 @@ func (h *Handlers) ResolveIncident(w http.ResponseWriter, r *http.Request) { return } + labelStoredTrace(existing, traceNotDerived) writeJSON(w, http.StatusOK, existing) } diff --git a/internal/api/incident_trace.go b/internal/api/incident_trace.go new file mode 100644 index 00000000..3b154dbc --- /dev/null +++ b/internal/api/incident_trace.go @@ -0,0 +1,190 @@ +// Copyright 2025 MCTL Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package api + +import ( + "context" + "encoding/json" + "log/slog" + "regexp" + "time" + + "github.com/mctlhq/mctl-api/internal/alerts" + "github.com/mctlhq/mctl-api/internal/audit" + "github.com/mctlhq/mctl-api/internal/auth" + "github.com/mctlhq/mctl-api/internal/telemetry" + "k8s.io/apimachinery/pkg/util/validation" +) + +// trace_unavailable reasons. Exactly one is set when no trace id is returned. +const ( + traceNotReported = "not_reported" + traceNotDerivedOnList = "not_derived_on_list" + // traceNotDerived marks write responses (create, patch, ack, resolve), + // which return the stored row without running the audit derivation. + traceNotDerived = "not_derived" + traceWorkflowUnknown = "workflow_name_unknown" + traceAuditUnavailable = "audit_unavailable" + traceAuditNotFound = "audit_record_not_found" + traceNoTraceparent = "no_traceparent_in_audit_record" + traceAuditForbidden = "audit_record_forbidden" +) + +// trace_source values. +const ( + traceSourceReported = "reported" + traceSourceAuditDerive = "workflow_audit" +) + +const ( + incidentSourceArgo = "argo-workflows" + incidentAuditTimeout = 2 * time.Second +) + +// workflowURLRe finds the Argo UI link the onExit incident legs put in the +// summary. It is the fallback for rows posted without workflow_name (every +// argo incident today); producers that send workflow_name never reach it. +// The name must end alphanumeric so sentence punctuation after the URL is +// not captured, and the URL may follow whitespace or an opening parenthesis. +var workflowURLRe = regexp.MustCompile(`(?:^|[\s(])https://workflows\.mctl\.ai/workflows/[a-z0-9-]+/([a-z0-9](?:[a-z0-9.-]*[a-z0-9])?)`) + +// workflowNameFromSummary returns the workflow named by the Argo UI link in +// summary, or "" if there is none. +func workflowNameFromSummary(summary string) string { + if m := workflowURLRe.FindStringSubmatch(summary); len(m) == 2 && len(validation.IsDNS1123Subdomain(m[1])) == 0 { + return m[1] + } + return "" +} + +// labelStoredTrace sets the response-only trace fields from the stored row +// alone, for responses that do not run deriveIncidentTrace. underived is the +// reason given to an argo incident whose trace could still be derived by +// GET /incidents/{id}; any other source without a trace is not_reported. +func labelStoredTrace(a *alerts.Alert, underived string) { + a.TraceSource, a.TraceUnavailable = "", "" + switch { + case a.TraceID != "": + a.TraceSource = traceSourceReported + case a.Source == incidentSourceArgo: + a.TraceUnavailable = underived + default: + a.TraceUnavailable = traceNotReported + } +} + +// canReadAuditEntry is the gate for deriving a trace from a workflow's audit +// record. The workflow name is caller-controlled (summary, workflow_name), so +// a non-admin may only read entries of a team they belong to. An entry with +// no team is an AdminOnly platform operation and is admin-only here, as in +// GetWorkflowLogs. This is deliberately stricter than GetWorkflow, which +// tolerates an empty team: the incident path has no reason to widen access, +// and argo incidents are posted to the admins tenant anyway. +func canReadAuditEntry(user *auth.User, e *audit.Entry) bool { + if user.IsAdmin() { + return true + } + team := auditEntryTenant(e) + return team != "" && user.HasTenantAccess(team) +} + +// workflowAuditLookup finds the audit record of a workflow. (nil, nil) means +// there is no such record. +type workflowAuditLookup interface { + LookupByWorkflow(ctx context.Context, name string) (*audit.Entry, error) +} + +// validIncidentTraceID returns the trace id in raw if it is a JSON string +// holding a valid trace id, else "". Never fails the request. +func validIncidentTraceID(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err != nil || !telemetry.ValidTraceID(s) { + if string(raw) != "null" { + slog.Debug("incident: ignoring invalid trace_id") + } + return "" + } + return s +} + +// validIncidentWorkflowName returns the workflow name in raw if it is a JSON +// string holding a valid Kubernetes object name, else "". +func validIncidentWorkflowName(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err != nil || s == "" || len(validation.IsDNS1123Subdomain(s)) > 0 { + if string(raw) != "null" { + slog.Debug("incident: ignoring invalid workflow_name") + } + return "" + } + return s +} + +// deriveIncidentTrace fills the response-only trace fields of a. It never +// writes anything back and never fails. +func deriveIncidentTrace(ctx context.Context, a *alerts.Alert, lookup workflowAuditLookup, canRead func(*audit.Entry) bool) { + a.TraceSource, a.TraceUnavailable = "", "" + if a.TraceID != "" { + a.TraceSource = traceSourceReported + return + } + if a.Source != incidentSourceArgo { + a.TraceUnavailable = traceNotReported + return + } + + name := a.WorkflowName + if name == "" { + name = workflowNameFromSummary(a.Summary) + } + if name == "" { + a.TraceUnavailable = traceWorkflowUnknown + return + } + if lookup == nil { + a.TraceUnavailable = traceAuditUnavailable + return + } + + lctx, cancel := context.WithTimeout(ctx, incidentAuditTimeout) + defer cancel() + entry, err := lookup.LookupByWorkflow(lctx, name) + if err != nil { + slog.Warn("incident: audit lookup for trace failed", "workflow", name, "error", err) + a.TraceUnavailable = traceAuditUnavailable + return + } + if entry == nil { + a.TraceUnavailable = traceAuditNotFound + return + } + if canRead != nil && !canRead(entry) { + a.TraceUnavailable = traceAuditForbidden + return + } + id, ok := telemetry.TraceIDFromTraceparent(entry.Parameters["traceparent"]) + if !ok { + a.TraceUnavailable = traceNoTraceparent + return + } + a.TraceID = id + a.TraceSource = traceSourceAuditDerive +} diff --git a/internal/api/incident_trace_handlers_test.go b/internal/api/incident_trace_handlers_test.go new file mode 100644 index 00000000..2927309a --- /dev/null +++ b/internal/api/incident_trace_handlers_test.go @@ -0,0 +1,398 @@ +// Copyright 2025 MCTL Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package api + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/google/uuid" + "github.com/mctlhq/mctl-api/internal/alerts" + "github.com/mctlhq/mctl-api/internal/audit" + "github.com/mctlhq/mctl-api/internal/auth" +) + +const ( + daeTraceID = "431486061c652efb376a860d9d656f25" + otherTraceID = "008ad8979d34c0509e74d8062b7f92ca" + daeIncidentID = "argo-mctl-agents-implement-dae24112-1791756910" + daeWorkflow = "mctl-agents-implement-dae24112" + daeTraceparent = "00-431486061c652efb376a860d9d656f25-7a16c732dc164e9e-01" + // Copied verbatim from the production incident (mctl_get_incident, 2026-10-12). + daeSummary = "implement implement issue-757-tool-availability-sli-counts-caller-erro Failed after 437.333132s — https://workflows.mctl.ai/workflows/argo-workflows/mctl-agents-implement-dae24112" +) + +// countingAuditLog records how often the incident handlers consult the audit +// log, so a test can prove the list endpoint never does. +type countingAuditLog struct { + *audit.Logger + lookups int +} + +func (c *countingAuditLog) LookupByWorkflow(ctx context.Context, name string) (*audit.Entry, error) { + c.lookups++ + return c.Logger.LookupByWorkflow(ctx, name) +} + +func decodeIncident(t *testing.T, rec *httptest.ResponseRecorder) map[string]any { + t.Helper() + var m map[string]any + if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil { + t.Fatalf("decode %q: %v", rec.Body.String(), err) + } + return m +} + +// mctl-api#585 T4: an invalid trace_id is ignored, never rejected. Each body +// still creates the incident (201) and stores no trace; a later PATCH with an +// invalid value keeps the stored trace and a valid one replaces it. +func TestIncidentTrace_InvalidInputIgnoredNotRejected(t *testing.T) { + store := newTestAlertStore(t) + h := &Handlers{opts: Options{AlertStore: store, AuditLog: audit.NewLogger()}} + admin := adminUser() + + for name, traceJSON := range map[string]string{ + "uppercase": `"431486061C652EFB376A860D9D656F25"`, + "31 chars": `"` + daeTraceID[:31] + `"`, + "all zeros": `"00000000000000000000000000000000"`, + "number": `123`, + "object": `{"id":"x"}`, + "bad workflow": `"` + daeTraceID + `", "workflow_name": 5`, + } { + t.Run("create "+name, func(t *testing.T) { + id := "trace-in-" + uuid.NewString()[:8] + body := `{"id":"` + id + `","source":"manual","type":"workflow_failed","tenant":"admins",` + + `"summary":"s","severity":"warning","fingerprint":"fp-` + id + `","trace_id":` + traceJSON + `}` + rec := httptest.NewRecorder() + h.CreateIncident(rec, withUser(httptest.NewRequest(http.MethodPost, "/api/v1/incidents", strings.NewReader(body)), admin)) + if rec.Code != http.StatusCreated { + t.Fatalf("status %d, want 201: %s", rec.Code, rec.Body.String()) + } + if resp := decodeIncident(t, rec); name != "bad workflow" && resp["trace_unavailable"] != "not_reported" { + t.Fatalf("create response for a manual incident without a trace: %v", resp["trace_unavailable"]) + } + stored, err := store.Get(context.Background(), id) + if err != nil { + t.Fatalf("get: %v", err) + } + if name == "bad workflow" { + // The trace is valid; only the workflow name is dropped. + if stored.TraceID != daeTraceID || stored.WorkflowName != "" { + t.Fatalf("stored trace=%q workflow=%q", stored.TraceID, stored.WorkflowName) + } + return + } + if stored.TraceID != "" { + t.Fatalf("invalid trace_id %s was stored as %q", traceJSON, stored.TraceID) + } + }) + } + + // Seed one incident with a valid trace, then PATCH it. + id := "trace-patch-" + uuid.NewString()[:8] + if _, err := store.Create(context.Background(), &alerts.Alert{ + ID: id, Source: "manual", Type: "workflow_failed", Tenant: "admins", Summary: "s", + Severity: "warning", Status: alerts.StatusOpen, Fingerprint: "fp-" + id, TraceID: daeTraceID, + }); err != nil { + t.Fatalf("seed: %v", err) + } + patch := func(body string) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + h.UpdateIncident(rec, withURLParam(withUser(httptest.NewRequest(http.MethodPatch, "/api/v1/incidents/"+id, strings.NewReader(body)), admin), "id", id)) + return rec + } + for _, bad := range []string{`{"trace_id":"NOT-HEX"}`, `{"trace_id":42}`, `{"trace_id":null}`, `{"analysis":"a","trace_id":""}`} { + if rec := patch(bad); rec.Code != http.StatusOK { + t.Fatalf("PATCH %s: status %d, want 200: %s", bad, rec.Code, rec.Body.String()) + } + if got, _ := store.Get(context.Background(), id); got.TraceID != daeTraceID { + t.Fatalf("PATCH %s replaced the stored trace with %q", bad, got.TraceID) + } + } + if rec := patch(`{"trace_id":"` + otherTraceID + `"}`); rec.Code != http.StatusOK { + t.Fatalf("valid PATCH: status %d: %s", rec.Code, rec.Body.String()) + } + if got, _ := store.Get(context.Background(), id); got.TraceID != otherTraceID { + t.Fatalf("valid PATCH did not replace the trace: %q", got.TraceID) + } +} + +// mctl-api#585 T8: the list endpoint never runs the audit derivation, and an +// item without a stored trace says so instead of looking traceless. +func TestIncidentTrace_ListMarksAbsenceWithoutAuditLookup(t *testing.T) { + store := newTestAlertStore(t) + log := &countingAuditLog{Logger: audit.NewLogger()} + log.Log(audit.Entry{ID: uuid.NewString(), UserID: "mctl-agent", Operation: "mctl-agents-implement", + WorkflowName: daeWorkflow, Status: "failed", Parameters: map[string]string{"traceparent": daeTraceparent}}) + h := &Handlers{opts: Options{AlertStore: store, AuditLog: log}} + + ctx := context.Background() + if _, err := store.Create(ctx, &alerts.Alert{ID: "with-trace", Source: "manual", Type: "workflow_failed", + Tenant: "admins", Summary: "s", Severity: "warning", Status: alerts.StatusOpen, + Fingerprint: "fp-with-trace", TraceID: otherTraceID}); err != nil { + t.Fatalf("seed reported: %v", err) + } + if _, err := store.Create(ctx, &alerts.Alert{ID: daeIncidentID, Source: "argo-workflows", Type: "workflow_failed", + Tenant: "admins", Service: "mctl-agents", Summary: daeSummary, Severity: "warning", + Status: alerts.StatusOpen, Fingerprint: "fp-dae"}); err != nil { + t.Fatalf("seed argo: %v", err) + } + + if _, err := store.Create(ctx, &alerts.Alert{ID: "manual-untraced", Source: "manual", Type: "workflow_failed", + Tenant: "admins", Summary: "s", Severity: "warning", Status: alerts.StatusOpen, + Fingerprint: "fp-manual-untraced"}); err != nil { + t.Fatalf("seed manual: %v", err) + } + + rec := httptest.NewRecorder() + h.ListIncidents(rec, withUser(httptest.NewRequest(http.MethodGet, "/api/v1/incidents", nil), adminUser())) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var resp struct { + Items []map[string]any `json:"items"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + byID := map[string]map[string]any{} + for _, it := range resp.Items { + byID[it["id"].(string)] = it + } + + reported := byID["with-trace"] + if reported["trace_id"] != otherTraceID || reported["trace_source"] != "reported" || reported["trace_unavailable"] != nil { + t.Fatalf("reported item: %v", reported) + } + argo := byID[daeIncidentID] + if _, ok := argo["trace_id"]; ok { + t.Fatalf("list derived a trace it must not derive: %v", argo) + } + if argo["trace_unavailable"] != "not_derived_on_list" { + t.Fatalf("argo item without a stored trace is unmarked: %v", argo) + } + // Derivation could never produce a trace for a non-argo incident, so the + // list gives the terminal reason rather than sending callers to GET it. + if manual := byID["manual-untraced"]; manual["trace_unavailable"] != "not_reported" { + t.Fatalf("non-argo item without a trace: %v", manual) + } + if log.lookups != 0 { + t.Fatalf("list consulted the audit log %d times, want 0", log.lookups) + } +} + +// mctl-api#585 acceptance, end to end through the handler: the production +// dae24112 incident resolves to its trace from the workflow's audit record. +func TestIncidentTrace_GetDerivesDae24112FromAudit(t *testing.T) { + store := newTestAlertStore(t) + log := &countingAuditLog{Logger: audit.NewLogger()} + log.Log(audit.Entry{ID: uuid.NewString(), UserID: "mctl-agent", Operation: "mctl-agents-implement", + WorkflowName: daeWorkflow, Status: "failed", Parameters: map[string]string{"traceparent": daeTraceparent}}) + h := &Handlers{opts: Options{AlertStore: store, AuditLog: log}} + + if _, err := store.Create(context.Background(), &alerts.Alert{ID: daeIncidentID, Source: "argo-workflows", + Type: "workflow_failed", Tenant: "admins", Service: "mctl-agents", Summary: daeSummary, + Severity: "warning", Status: alerts.StatusResolved, + Fingerprint: "workflow_failed:implement:mctl-telegram:issue-757-tool-availability-sli-counts-caller-erro"}); err != nil { + t.Fatalf("seed: %v", err) + } + + rec := httptest.NewRecorder() + h.GetIncident(rec, withURLParam(withUser(httptest.NewRequest(http.MethodGet, "/api/v1/incidents/"+daeIncidentID, nil), adminUser()), "id", daeIncidentID)) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got := decodeIncident(t, rec) + if got["trace_id"] != daeTraceID || got["trace_source"] != "workflow_audit" { + t.Fatalf("got trace_id=%v trace_source=%v unavailable=%v", got["trace_id"], got["trace_source"], got["trace_unavailable"]) + } + if log.lookups != 1 { + t.Fatalf("audit lookups = %d, want 1", log.lookups) + } + // The derived trace is response-only: the row stays without a trace. + if stored, _ := store.Get(context.Background(), daeIncidentID); stored.TraceID != "" { + t.Fatalf("GET wrote the derived trace back: %q", stored.TraceID) + } +} + +// The trace derivation must not hand a non-admin the audit record of a +// workflow outside their teams. The workflow name is caller-controlled, and +// an entry without a team is an AdminOnly platform operation. +func TestCanReadAuditEntry(t *testing.T) { + platform := &audit.Entry{Parameters: map[string]string{"traceparent": daeTraceparent}} + acme := &audit.Entry{Parameters: map[string]string{"tenant_name": "acme", "traceparent": daeTraceparent}} + member := &auth.User{ID: "dev", Groups: []string{"acme"}} + outsider := &auth.User{ID: "dev2", Groups: []string{"other"}} + + for name, tc := range map[string]struct { + user *auth.User + entry *audit.Entry + want bool + }{ + "admin, platform entry": {adminUser(), platform, true}, + "admin, tenant entry": {adminUser(), acme, true}, + "member, own tenant": {member, acme, true}, + "outsider, other tenant": {outsider, acme, false}, + "non-admin, platform entry": {member, platform, false}, + "non-admin, entry no params": {member, &audit.Entry{}, false}, + } { + if got := canReadAuditEntry(tc.user, tc.entry); got != tc.want { + t.Errorf("%s: canReadAuditEntry = %v, want %v", name, got, tc.want) + } + } +} + +// The summary fallback must find the workflow name wherever the URL sits in +// ordinary prose: at the end of a sentence, inside parentheses. +func TestDeriveIncidentTrace_SummaryURLInProse(t *testing.T) { + entry := &audit.Entry{Parameters: map[string]string{"traceparent": daeTraceparent}} + for _, summary := range []string{ + daeSummary, + "Workflow failed: https://workflows.mctl.ai/workflows/argo-workflows/" + daeWorkflow + ".", + "Workflow failed (https://workflows.mctl.ai/workflows/argo-workflows/" + daeWorkflow + ")", + "Workflow failed (https://workflows.mctl.ai/workflows/argo-workflows/" + daeWorkflow + ").", + "see https://workflows.mctl.ai/workflows/argo-workflows/" + daeWorkflow + "-", + } { + var asked string + lookup := lookupFunc(func(_ context.Context, name string) (*audit.Entry, error) { + asked = name + return entry, nil + }) + a := &alerts.Alert{Source: "argo-workflows", Summary: summary} + deriveIncidentTrace(context.Background(), a, lookup, func(*audit.Entry) bool { return true }) + if asked != daeWorkflow || a.TraceID != daeTraceID { + t.Errorf("summary %q: looked up %q, trace %q (unavailable=%q)", summary, asked, a.TraceID, a.TraceUnavailable) + } + } +} + +type lookupFunc func(context.Context, string) (*audit.Entry, error) + +func (f lookupFunc) LookupByWorkflow(ctx context.Context, name string) (*audit.Entry, error) { + return f(ctx, name) +} + +// The gate as GetIncident applies it, with real non-admin users: a member of +// the incident's tenant gets the trace only from an audit entry of a team +// they belong to. +func TestIncidentTrace_GetAsNonAdmin(t *testing.T) { + store := newTestAlertStore(t) + log := audit.NewLogger() + h := &Handlers{opts: Options{AlertStore: store, AuditLog: log}} + member := &auth.User{ID: "dev", Groups: []string{"acme"}} + + for name, tc := range map[string]struct { + params map[string]string + wantTrace string + wantReason string + }{ + "own team": {map[string]string{"tenant_name": "acme", "traceparent": daeTraceparent}, daeTraceID, ""}, + "other team": {map[string]string{"tenant_name": "other", "traceparent": daeTraceparent}, "", "audit_record_forbidden"}, + "platform op": {map[string]string{"traceparent": daeTraceparent}, "", "audit_record_forbidden"}, + } { + t.Run(name, func(t *testing.T) { + wf := "wf-" + uuid.NewString()[:8] + log.Log(audit.Entry{ID: uuid.NewString(), UserID: "x", Operation: "op", WorkflowName: wf, Status: "failed", Parameters: tc.params}) + id := "argo-" + wf + if _, err := store.Create(context.Background(), &alerts.Alert{ID: id, Source: "argo-workflows", + Type: "workflow_failed", Tenant: "acme", Summary: "failed https://workflows.mctl.ai/workflows/acme/" + wf, + Severity: "warning", Status: alerts.StatusOpen, Fingerprint: "fp-" + wf}); err != nil { + t.Fatalf("seed: %v", err) + } + rec := httptest.NewRecorder() + h.GetIncident(rec, withURLParam(withUser(httptest.NewRequest(http.MethodGet, "/api/v1/incidents/"+id, nil), member), "id", id)) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + got := decodeIncident(t, rec) + trace, _ := got["trace_id"].(string) + reason, _ := got["trace_unavailable"].(string) + if trace != tc.wantTrace || reason != tc.wantReason { + t.Fatalf("trace_id=%q trace_unavailable=%q, want %q / %q", trace, reason, tc.wantTrace, tc.wantReason) + } + }) + } +} + +// Write responses return the stored row without deriving, and say so: an +// argo incident without a trace is not_derived, never silently traceless. +func TestIncidentTrace_WriteResponsesAreLabelled(t *testing.T) { + store := newTestAlertStore(t) + h := &Handlers{opts: Options{AlertStore: store, AuditLog: audit.NewLogger()}} + admin := adminUser() + + rec := httptest.NewRecorder() + h.CreateIncident(rec, withUser(httptest.NewRequest(http.MethodPost, "/api/v1/incidents", strings.NewReader( + `{"id":"argo-label-1","source":"argo-workflows","type":"workflow_failed","tenant":"admins","summary":"s","severity":"warning","fingerprint":"fp-label-1"}`)), admin)) + if rec.Code != http.StatusCreated { + t.Fatalf("create: %d %s", rec.Code, rec.Body.String()) + } + if got := decodeIncident(t, rec); got["trace_unavailable"] != "not_derived" || got["trace_id"] != nil { + t.Fatalf("create response: trace_id=%v trace_unavailable=%v", got["trace_id"], got["trace_unavailable"]) + } + + rec = httptest.NewRecorder() + h.UpdateIncident(rec, withURLParam(withUser(httptest.NewRequest(http.MethodPatch, "/api/v1/incidents/argo-label-1", + strings.NewReader(`{"trace_id":"`+daeTraceID+`"}`)), admin), "id", "argo-label-1")) + if rec.Code != http.StatusOK { + t.Fatalf("patch: %d %s", rec.Code, rec.Body.String()) + } + if got := decodeIncident(t, rec); got["trace_id"] != daeTraceID || got["trace_source"] != "reported" || got["trace_unavailable"] != nil { + t.Fatalf("patch response: %v / %v / %v", got["trace_id"], got["trace_source"], got["trace_unavailable"]) + } +} + +// A recurring argo failure dedups into one open incident whose summary keeps +// the first run's link. GET must derive the newest run's trace, not the +// first one's: the producer sends no workflow_name, so create records it +// from each occurrence's summary and the dedup hit moves it forward. +func TestIncidentTrace_DedupedArgoIncidentFollowsNewestRun(t *testing.T) { + store := newTestAlertStore(t) + log := audit.NewLogger() + h := &Handlers{opts: Options{AlertStore: store, AuditLog: log}} + admin := adminUser() + + runs := []struct{ wf, traceparent string }{ + {"mctl-agents-implement-run1", "00-" + otherTraceID + "-7a16c732dc164e9e-01"}, + {"mctl-agents-implement-run2", daeTraceparent}, + } + for i, run := range runs { + log.Log(audit.Entry{ID: uuid.NewString(), UserID: "mctl-agent", Operation: "mctl-agents-implement", + WorkflowName: run.wf, Status: "failed", Parameters: map[string]string{"traceparent": run.traceparent}}) + body := `{"id":"argo-` + run.wf + `","source":"argo-workflows","type":"workflow_failed","tenant":"admins",` + + `"service":"mctl-agents","severity":"warning","fingerprint":"workflow_failed:implement:recurring",` + + `"summary":"implement Failed — https://workflows.mctl.ai/workflows/argo-workflows/` + run.wf + `"}` + rec := httptest.NewRecorder() + h.CreateIncident(rec, withUser(httptest.NewRequest(http.MethodPost, "/api/v1/incidents", strings.NewReader(body)), admin)) + if rec.Code != http.StatusCreated { + t.Fatalf("occurrence %d: status %d: %s", i+1, rec.Code, rec.Body.String()) + } + } + + const id = "argo-mctl-agents-implement-run1" + rec := httptest.NewRecorder() + h.GetIncident(rec, withURLParam(withUser(httptest.NewRequest(http.MethodGet, "/api/v1/incidents/"+id, nil), admin), "id", id)) + got := decodeIncident(t, rec) + if got["occurrence_count"] != float64(2) { + t.Fatalf("occurrences did not dedup: %v", got["occurrence_count"]) + } + if got["trace_id"] != daeTraceID || got["workflow_name"] != "mctl-agents-implement-run2" { + t.Fatalf("derived trace_id=%v workflow_name=%v, want the newest run (%s, run2)", got["trace_id"], got["workflow_name"], daeTraceID) + } +} diff --git a/internal/api/incident_trace_test.go b/internal/api/incident_trace_test.go new file mode 100644 index 00000000..907d5f84 --- /dev/null +++ b/internal/api/incident_trace_test.go @@ -0,0 +1,99 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/mctlhq/mctl-api/internal/alerts" + "github.com/mctlhq/mctl-api/internal/audit" +) + +type fakeLookup struct { + entry *audit.Entry + err error + calls int +} + +func (f *fakeLookup) LookupByWorkflow(context.Context, string) (*audit.Entry, error) { + f.calls++ + return f.entry, f.err +} + +const dae = "https://workflows.mctl.ai/workflows/argo-workflows/mctl-agents-implement-dae24112" + +func argoAlert(summary string) *alerts.Alert { + return &alerts.Alert{ID: "argo-x-1", Source: "argo-workflows", Summary: summary} +} + +func TestDeriveIncidentTrace(t *testing.T) { + tp := map[string]string{"traceparent": "00-431486061c652efb376a860d9d656f25-7a16c732dc164e9e-01"} + allow := func(*audit.Entry) bool { return true } + deny := func(*audit.Entry) bool { return false } + + tests := []struct { + name string + a *alerts.Alert + lookup *fakeLookup + nilLookup bool + canRead func(*audit.Entry) bool + wantID string + wantSource string + wantReason string + wantCalls int + }{ + {"summary url", argoAlert("failed: " + dae), &fakeLookup{entry: &audit.Entry{Parameters: tp}}, false, allow, "431486061c652efb376a860d9d656f25", "workflow_audit", "", 1}, + {"stored workflow name", &alerts.Alert{Source: "argo-workflows", WorkflowName: "wf-1"}, &fakeLookup{entry: &audit.Entry{Parameters: tp}}, false, allow, "431486061c652efb376a860d9d656f25", "workflow_audit", "", 1}, + {"reported wins", &alerts.Alert{Source: "argo-workflows", TraceID: "431486061c652efb376a860d9d656f25"}, &fakeLookup{}, false, allow, "431486061c652efb376a860d9d656f25", "reported", "", 0}, + {"not found", argoAlert(dae), &fakeLookup{}, false, allow, "", "", "audit_record_not_found", 1}, + {"lookup error", argoAlert(dae), &fakeLookup{err: errors.New("db")}, false, allow, "", "", "audit_unavailable", 1}, + {"nil lookup", argoAlert(dae), nil, true, allow, "", "", "audit_unavailable", 0}, + {"no name", argoAlert("no url"), &fakeLookup{}, false, allow, "", "", "workflow_name_unknown", 0}, + {"no traceparent", argoAlert(dae), &fakeLookup{entry: &audit.Entry{}}, false, allow, "", "", "no_traceparent_in_audit_record", 1}, + {"forbidden", argoAlert(dae), &fakeLookup{entry: &audit.Entry{Parameters: tp}}, false, deny, "", "", "audit_record_forbidden", 1}, + {"non-argo", &alerts.Alert{Source: "alertmanager"}, &fakeLookup{}, false, allow, "", "", "not_reported", 0}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var l workflowAuditLookup + if !tc.nilLookup { + l = tc.lookup + } + deriveIncidentTrace(context.Background(), tc.a, l, tc.canRead) + if tc.a.TraceID != tc.wantID || tc.a.TraceSource != tc.wantSource || tc.a.TraceUnavailable != tc.wantReason { + t.Fatalf("got id=%q source=%q reason=%q", tc.a.TraceID, tc.a.TraceSource, tc.a.TraceUnavailable) + } + if tc.lookup != nil && tc.lookup.calls != tc.wantCalls { + t.Fatalf("lookup calls = %d, want %d", tc.lookup.calls, tc.wantCalls) + } + }) + } +} + +func TestValidIncidentTraceInputs(t *testing.T) { + const id = "431486061c652efb376a860d9d656f25" + if got := validIncidentTraceID(json.RawMessage(`"` + id + `"`)); got != id { + t.Errorf("valid: %q", got) + } + for _, bad := range []string{`123`, `"` + id[:31] + `"`, `"431486061C652EFB376A860D9D656F25"`, `"00000000000000000000000000000000"`, `null`} { + if got := validIncidentTraceID(json.RawMessage(bad)); got != "" { + t.Errorf("%s accepted: %q", bad, got) + } + } + if validIncidentWorkflowName(json.RawMessage(`"Bad_Name"`)) != "" || validIncidentWorkflowName(json.RawMessage(`5`)) != "" { + t.Error("invalid workflow name accepted") + } + if validIncidentWorkflowName(json.RawMessage(`"wf-1"`)) != "wf-1" { + t.Error("valid workflow name rejected") + } +} + +func TestAlertJSONTraceKey(t *testing.T) { + b, _ := json.Marshal(alerts.Alert{TraceID: "abc"}) + var m map[string]any + _ = json.Unmarshal(b, &m) + if m["trace_id"] != "abc" { + t.Fatalf("trace_id missing: %s", b) + } +} diff --git a/internal/audit/logger.go b/internal/audit/logger.go index 1ef75550..c40e2047 100644 --- a/internal/audit/logger.go +++ b/internal/audit/logger.go @@ -15,6 +15,7 @@ package audit import ( + "context" "log/slog" "sync" "time" @@ -57,6 +58,9 @@ type Log interface { Log(entry Entry) List(limit int) []Entry GetByWorkflow(name string) *Entry + // LookupByWorkflow is GetByWorkflow that can tell "no such record" + // ((nil, nil)) from "could not look" ((nil, err)). + LookupByWorkflow(ctx context.Context, name string) (*Entry, error) // UpdateStatus moves an entry out of a non-terminal status. It is a no-op // (returning false) when no row matches or the row is already terminal, // which is what makes a redelivered webhook safe. @@ -157,6 +161,12 @@ func (l *Logger) GetByWorkflow(workflowName string) *Entry { return nil } +// LookupByWorkflow returns the entry for workflowName, or (nil, nil) if none. +// The in-memory store cannot fail. +func (l *Logger) LookupByWorkflow(_ context.Context, workflowName string) (*Entry, error) { + return l.GetByWorkflow(workflowName), nil +} + // UpdateStatus closes out the newest non-terminal entry for workflowName. func (l *Logger) UpdateStatus(workflowName, status, message string) bool { if workflowName == "" { diff --git a/internal/audit/lookup_postgres_test.go b/internal/audit/lookup_postgres_test.go new file mode 100644 index 00000000..8eed2a1f --- /dev/null +++ b/internal/audit/lookup_postgres_test.go @@ -0,0 +1,62 @@ +// Copyright 2025 MCTL Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package audit + +import ( + "context" + "os" + "testing" + + "github.com/google/uuid" +) + +// mctl-api#585 task 2: LookupByWorkflow tells "no such record" ((nil, nil)) +// from "could not look" ((nil, err)), which GetByWorkflow cannot. +func TestPostgresLogger_LookupByWorkflow(t *testing.T) { + connStr := os.Getenv("TEST_DATABASE_URL") + if connStr == "" { + t.Skip("TEST_DATABASE_URL not set; skipping Postgres-backed audit test") + } + ctx := context.Background() + p, err := NewPostgresLogger(ctx, connStr) + if err != nil { + t.Fatal(err) + } + id := uuid.NewString() + wf := "lookup-" + id[:8] + t.Cleanup(func() { + _, _ = p.pool.Exec(ctx, `DELETE FROM audit_events WHERE id=$1`, id) + p.pool.Close() + }) + p.Log(Entry{ID: id, UserID: "mctl-agent", Operation: "mctl-agents-implement", WorkflowName: wf, + Status: "submitted", Parameters: map[string]string{"traceparent": "00-431486061c652efb376a860d9d656f25-7a16c732dc164e9e-01"}}) + + e, err := p.LookupByWorkflow(ctx, wf) + if err != nil || e == nil || e.Parameters["traceparent"] == "" { + t.Fatalf("found: entry=%+v err=%v", e, err) + } + + e, err = p.LookupByWorkflow(ctx, "no-such-"+id[:8]) + if err != nil || e != nil { + t.Fatalf("missing record: entry=%+v err=%v, want (nil, nil)", e, err) + } + + cancelled, cancel := context.WithCancel(ctx) + cancel() + e, err = p.LookupByWorkflow(cancelled, wf) + if err == nil || e != nil { + t.Fatalf("failed read: entry=%+v err=%v, want (nil, error)", e, err) + } +} diff --git a/internal/audit/postgres.go b/internal/audit/postgres.go index 240c3c85..353011a9 100644 --- a/internal/audit/postgres.go +++ b/internal/audit/postgres.go @@ -17,11 +17,13 @@ package audit import ( "context" "encoding/json" + "errors" "fmt" "log/slog" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -181,6 +183,23 @@ func (p *PostgresLogger) GetByWorkflow(workflowName string) *Entry { return &e } +// LookupByWorkflow is GetByWorkflow with the caller's context, mapping +// pgx.ErrNoRows to (nil, nil) and any other failure to an error. +func (p *PostgresLogger) LookupByWorkflow(ctx context.Context, workflowName string) (*Entry, error) { + row := p.pool.QueryRow(ctx, + `SELECT id, timestamp, user_id, operation, params, workflow, status, risk_level, message, client_ip, user_agent, request_id + FROM audit_events WHERE workflow = $1 ORDER BY timestamp DESC LIMIT 1`, workflowName) + + e, err := scanAuditEntry(row.Scan) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + return nil, fmt.Errorf("audit postgres: lookup by workflow: %w", err) + } + return &e, nil +} + func scanAuditEntry(scan func(dest ...any) error) (Entry, error) { var e Entry var params []byte diff --git a/internal/mcp/server.go b/internal/mcp/server.go index f5a7ce5a..49dd6628 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -2116,7 +2116,19 @@ func (s *Server) toolGetIncident() (mcplib.Tool, server.ToolHandlerFunc) { mcplib.WithDescription(`Get full details of an incident including evidence, analysis, and PR information. Accepts a full incident ID or an 8-character prefix (as shown in Telegram notifications). -Returns the incident with all collected evidence (logs, alerts, workflow data).`), +Returns the incident with all collected evidence (logs, alerts, workflow data). + +Also returns trace_id (the 32-hex trace to open in Tempo, then pivot to Loki) with +trace_source "reported" (sent by the producer) or "workflow_audit" (derived from the +failed workflow's submit audit record). When no trace id could be obtained, +trace_unavailable gives the reason. A missing trace_id means "not known", never +"no trace exists". trace_unavailable is "not_reported" when no trace can exist for +the incident's source; "not_derived_on_list" for an argo incident on a list +response (call this tool to resolve it); "not_derived" on a create, patch, +acknowledge or resolve response; or, when this tool ran the derivation and could +not produce a trace, one of "workflow_name_unknown", "audit_unavailable", +"audit_record_not_found", "audit_record_forbidden" or +"no_traceparent_in_audit_record".`), mcplib.WithString("id", mcplib.Required(), mcplib.Description("Incident ID (full UUID or 8-char prefix)"), diff --git a/internal/operations/executor.go b/internal/operations/executor.go index 0d7d79dc..65f79fa6 100644 --- a/internal/operations/executor.go +++ b/internal/operations/executor.go @@ -22,7 +22,6 @@ import ( "github.com/mctlhq/mctl-api/internal/telemetry" "log/slog" "os" - "regexp" "strings" "time" @@ -424,11 +423,6 @@ func trimWorkflowStatus(obj map[string]interface{}) map[string]interface{} { return result } -var ( - w3cTraceparent = regexp.MustCompile(`^00-([0-9a-f]{32})-[0-9a-f]{16}-[0-9a-f]{2}$`) - w3cTraceID = regexp.MustCompile(`^[0-9a-f]{32}$`) -) - // workflowTraceID names the trace a workflow belongs to, so a caller holding // only a workflow name (an ArgoWorkflowFailed alert, a ticket) can open the // trace in Tempo and from there the logs (mctlhq/mctl-agent#97). @@ -450,8 +444,8 @@ func workflowTraceID(obj map[string]interface{}) (string, string) { continue } value, _ := param["value"].(string) - if m := w3cTraceparent.FindStringSubmatch(value); len(m) == 2 && !allZero(m[1]) { - return m[1], "traceparent-argument" + if id, ok := telemetry.TraceIDFromTraceparent(value); ok { + return id, "traceparent-argument" } } } @@ -459,7 +453,7 @@ func workflowTraceID(obj map[string]interface{}) (string, string) { if meta, ok := obj["metadata"].(map[string]interface{}); ok { if ann, ok := meta["annotations"].(map[string]interface{}); ok { value, _ := ann["mctl.ai/trace-id"].(string) - if w3cTraceID.MatchString(value) && !allZero(value) { + if telemetry.ValidTraceID(value) { return value, "submit-annotation" } } @@ -467,10 +461,6 @@ func workflowTraceID(obj map[string]interface{}) (string, string) { return "", "" } -func allZero(s string) bool { - return strings.Trim(s, "0") == "" -} - // labelValue makes s usable as a Kubernetes label value. A value that is // already valid is returned unchanged, so existing selectors keep matching. // Anything else -- a CI principal "ci:owner/repo" (mctl-api#530), an OIDC diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index d4b6618f..0cd60b6d 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -10,6 +10,7 @@ import ( "context" "log/slog" "os" + "regexp" "strings" "sync/atomic" "time" @@ -136,3 +137,25 @@ func TraceparentFrom(ctx context.Context) string { propagation.TraceContext{}.Inject(ctx, carrier) return carrier.Get("traceparent") } + +var ( + traceIDRe = regexp.MustCompile(`^[0-9a-f]{32}$`) + traceparentRe = regexp.MustCompile(`^00-([0-9a-f]{32})-[0-9a-f]{16}-[0-9a-f]{2}$`) +) + +// ValidTraceID reports whether s is exactly 32 lowercase hex characters and +// not all zeros. +func ValidTraceID(s string) bool { + return traceIDRe.MatchString(s) && strings.Trim(s, "0") != "" +} + +// TraceIDFromTraceparent extracts the trace id from a W3C version-00 +// traceparent. It returns false when the value is malformed or the trace id +// is all zeros. +func TraceIDFromTraceparent(tp string) (string, bool) { + m := traceparentRe.FindStringSubmatch(tp) + if len(m) != 2 || !ValidTraceID(m[1]) { + return "", false + } + return m[1], true +} diff --git a/internal/telemetry/traceid_test.go b/internal/telemetry/traceid_test.go new file mode 100644 index 00000000..2412f4f6 --- /dev/null +++ b/internal/telemetry/traceid_test.go @@ -0,0 +1,23 @@ +package telemetry + +import "testing" + +func TestValidTraceIDAndTraceparent(t *testing.T) { + const id = "431486061c652efb376a860d9d656f25" + if !ValidTraceID(id) { + t.Fatal("valid id rejected") + } + for _, bad := range []string{"", id[:31], id + "0", "431486061C652EFB376A860D9D656F25", "zz1486061c652efb376a860d9d656f25", "00000000000000000000000000000000"} { + if ValidTraceID(bad) { + t.Errorf("ValidTraceID(%q) = true", bad) + } + } + if got, ok := TraceIDFromTraceparent("00-" + id + "-7a16c732dc164e9e-01"); !ok || got != id { + t.Errorf("valid traceparent: %q %v", got, ok) + } + for _, bad := range []string{"", "01-" + id + "-7a16c732dc164e9e-01", "00-00000000000000000000000000000000-7a16c732dc164e9e-01", "00-" + id + "-short-01"} { + if _, ok := TraceIDFromTraceparent(bad); ok { + t.Errorf("TraceIDFromTraceparent(%q) ok", bad) + } + } +}