Repository navigation
feat(agents): issue-585-feat-incidents-store-and-return-trace-id #589
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||||
|---|---|---|---|---|---|---|---|---|
|
|
@@ -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{} | ||||||||
| 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, | ||||||||
| 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, | ||||||||
| 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 | ||||||||
|
There was a problem hiding this comment. P3 — The first
That is the shape a producer will have as soon as it starts sending
workflow_name = CASE
WHEN EXCLUDED.workflow_name <> '' THEN EXCLUDED.workflow_name
WHEN EXCLUDED.trace_id <> '' THEN ''
ELSE alerts.workflow_name END...or, if blanking is intended, add the row to the test table so a future reader knows it was a decision.
Member
There was a problem hiding this comment. Intended, and now documented and tested in f2472b6. trace_id and workflow_name describe the same run, so a new trace with no name must not stay paired with the previous run's name. There is a comment on the clause, and a new |
||||||||
| 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, 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, | ||||||||
| 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 := 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 := 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 ` + 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 := scanAlert(rows, &a); err != nil { | ||||||||
| slog.Error("alerts store: scan failed", "error", err) | ||||||||
| continue | ||||||||
| } | ||||||||
|
|
||||||||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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") | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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 | ||
|
There was a problem hiding this comment. P3 — "They describe the newest occurrence" is not quite what the dedup rules implement, and the gap is the same staleness The dedup
That is the first-run answer presented as fresh — exactly what the Keeping the trace rather than dropping it is the right default, so no behaviour change is needed. Worth narrowing the sentence to what the code does, e.g. "they describe the newest occurrence that reported a trace, or the newest occurrence while none has", so a responder reading |
||
| // 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. | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
P3 — this branch blanks a known
workflow_name, which is the correct reading of the new "they move together" policy but is the one dedup combination the test does not cover.The pairing fix is good:
TestTraceID_Dedupnow covers new-trace-replaces-both, no-trace-keeps-both, and newer-name-while-untraced. What is left unexercised isEXCLUDED.trace_id <> '' AND EXCLUDED.workflow_name = ''— an occurrence that reports a trace but no workflow name. The firstWHENthen assigns the emptyEXCLUDED.workflow_name, so a row that held(wf-a, "")becomes("", T_b): the incident keeps a trace but loses the workflow link it previously had in the response.That is self-consistent (the alternative, keeping
wf-anext toT_b, is exactly the desync this commit removes), so no SQL change is needed. But a future reader will ask, and the comment above theCASEdoes not say it. One line inTestTraceID_Dedup:plus a half-sentence in the doc comment ("a trace reported without a name clears the stale name rather than pairing it with the new trace") would make the choice deliberate rather than incidental.