SRE → triage alerts
Mark an alert as a false positive or ignored, list investigation groups, and resolve finished threads from code.
| Step | What the SDK does |
|---|---|
| 1. List | list_alerts / list_investigation_groups |
| 2. Triage | ignore_alert or set_alert_attention |
| 3. Close | resolve_investigations / cancel_investigations |
Use --dry-run on the sample to inspect without writing.
Minimal
from stackgen import StackgenClient, StackgenConfig
client = StackgenClient(
StackgenConfig(
base_url="https://app.stackgen.com",
api_token="stackgen_…",
workspace_id="<workspace-uuid>",
)
)
client.sre.set_alert_attention(alert_id, "false_positive", reason="known flake")
# or: client.sre.ignore_alert(alert_id, reason="noise")
groups = client.sre.list_investigation_groups(page_size=20)
print(groups)
client.sre.resolve_investigations([investigation_id])import { StackgenClient, StackgenConfig } from "stackgen-sdk";
const client = new StackgenClient(
new StackgenConfig({
baseUrl: "https://app.stackgen.com",
apiToken: "stackgen_…",
workspaceId: "<workspace-uuid>",
}),
);
await client.sre.setAlertAttention(alertId, "false_positive", "known flake");
console.log(await client.sre.listInvestigationGroups({ page_size: 20 }));
await client.sre.resolveInvestigations([investigationId]);_, err := client.SRE.SetAlertAttention(ctx, alertID, "false_positive", "known flake")
if err != nil { return err }
groups, err := client.SRE.ListInvestigationGroups(ctx, map[string]string{"page_size": "20"})
if err != nil { return err }
fmt.Println(groups)
_, err = client.SRE.ResolveInvestigations(ctx, []string{investigationID})
return errHow to run
export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>
python examples/sre/triage-alerts.py --dry-run
python examples/sre/triage-alerts.py --alert-id <uuid> --false-positive --resolvenpx tsx examples/sre/triage-alerts.ts --dry-rungo run ./examples/go/sre/triage-alerts.go --dry-runComplete script
- Python:
examples/sre/triage-alerts.py - TypeScript:
examples/sre/triage-alerts.ts - Go:
examples/go/sre/triage-alerts.go
#!/usr/bin/env python3
"""SRE: triage an alert and close finished investigation threads.
Documented at:
https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-triage-alerts/
Credentials (flags override env):
STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID (or STACKGEN_ORG_ID / STACKGEN_PROJECT)
Usage:
python examples/sre/triage-alerts.py --dry-run
python examples/sre/triage-alerts.py --alert-id <uuid> --false-positive
python examples/sre/triage-alerts.py --alert-id <uuid> --ignore-reason noise --resolve
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from stackgen import StackgenClient, StackgenConfig
from stackgen.errors import StackgenError
def _org_id(cli_value: str | None) -> str:
return (
(cli_value or "").strip()
or os.environ.get("STACKGEN_WORKSPACE_ID", "").strip()
or os.environ.get("STACKGEN_ORG_ID", "").strip()
or os.environ.get("STACKGEN_PROJECT", "").strip()
or os.environ.get("STACKGEN_PROJECT_ID", "").strip()
)
def _config(args: argparse.Namespace) -> StackgenConfig:
base = (args.base_url or os.environ.get("STACKGEN_URL", "")).strip()
token = (args.api_token or os.environ.get("STACKGEN_TOKEN", "")).strip()
org = _org_id(args.workspace_id)
if not base or not token:
raise SystemExit(
"set --base-url/--api-token or STACKGEN_URL / STACKGEN_TOKEN"
)
return StackgenConfig(base_url=base, api_token=token, workspace_id=org)
def _alert_rows(payload: object) -> list[dict]:
if not isinstance(payload, dict):
return []
rows = payload.get("alerts") or payload.get("items") or []
return [r for r in rows if isinstance(r, dict)]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base-url", default="")
parser.add_argument("--api-token", default="")
parser.add_argument("--workspace-id",
"--org-id", default="")
parser.add_argument("--alert-id", default="")
parser.add_argument("--q", default="", help="Alert search when --alert-id is omitted")
parser.add_argument("--ignore-reason", default="")
parser.add_argument(
"--false-positive",
action="store_true",
help="Set attention to false_positive",
)
parser.add_argument(
"--resolve",
action="store_true",
help="Resolve completed/failed/draft investigations for the alert",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="List candidates and exit without writing",
)
args = parser.parse_args()
client = StackgenClient(_config(args))
alert_id = args.alert_id.strip()
if not alert_id:
listed = client.sre.list_alerts(status="active", q=args.q or None, page_size=5)
print("alerts:", json.dumps(listed, indent=2, default=str))
rows = _alert_rows(listed)
if rows:
alert_id = str(rows[0].get("id") or "")
groups = client.sre.list_investigation_groups(page_size=5)
print("groups:", json.dumps(groups, indent=2, default=str))
if args.dry_run:
if alert_id:
print(f"alert_id={alert_id}", file=sys.stderr)
return 0
if not alert_id:
print("error: pass --alert-id or ensure list_alerts returns a row", file=sys.stderr)
return 1
if args.false_positive:
updated = client.sre.set_alert_attention(
alert_id,
"false_positive",
reason=args.ignore_reason or None,
)
print("attention:", json.dumps(updated, indent=2, default=str))
elif args.ignore_reason:
updated = client.sre.ignore_alert(alert_id, reason=args.ignore_reason)
print("ignored:", json.dumps(updated, indent=2, default=str))
elif not args.resolve:
print(
"pass --false-positive, --ignore-reason TEXT, --resolve, or --dry-run",
file=sys.stderr,
)
return 2
if args.resolve:
inv = client.sre.list_investigations(alert_id=alert_id, page_size=50)
print("investigations:", json.dumps(inv, indent=2, default=str))
rows: list[dict] = []
if isinstance(inv, dict):
rows = [r for r in (inv.get("investigations") or []) if isinstance(r, dict)]
done = [
str(r["id"])
for r in rows
if r.get("id")
and str(r.get("status") or "") in {"completed", "failed", "draft"}
]
if done:
resolved = client.sre.resolve_investigations(done)
print("resolved:", json.dumps(resolved, indent=2, default=str))
else:
print("resolve: no completed/failed/draft threads", file=sys.stderr)
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except StackgenError as exc:
print(f"error: {exc}", file=sys.stderr)
if "401" in str(exc) or "missing identity" in str(exc).lower():
print(
"hint: SRE /app/sre routes need the edge to inject workspace "
"identity (X-Appcd-Org).",
file=sys.stderr,
)
raise SystemExit(1)/**
* SRE: triage an alert and close finished investigation threads.
*
* Documented at:
* https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-triage-alerts/
*
* Usage:
* npx tsx examples/sre/triage-alerts.ts --dry-run
* npx tsx examples/sre/triage-alerts.ts --alert-id <uuid> --false-positive
*/
import { parseArgs } from "node:util";
import { StackgenClient, StackgenConfig, StackgenError } from "stackgen-sdk";
function asRecord(value: unknown): Record<string, unknown> | undefined {
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: undefined;
}
function show(label: string, value: unknown): void {
console.log(`${label}:`, JSON.stringify(value, null, 2));
}
function alertRows(payload: unknown): Record<string, unknown>[] {
const rec = asRecord(payload);
const rows = (rec?.alerts ?? rec?.items ?? []) as unknown[];
return rows.filter(
(r): r is Record<string, unknown> => !!r && typeof r === "object" && !Array.isArray(r),
);
}
async function main(): Promise<number> {
const { values } = parseArgs({
options: {
"base-url": { type: "string", default: "" },
"api-token": { type: "string", default: "" },
"org-id": { type: "string" },
"workspace-id": { type: "string", default: "" },
"alert-id": { type: "string", default: "" },
q: { type: "string", default: "" },
"ignore-reason": { type: "string", default: "" },
"false-positive": { type: "boolean", default: false },
resolve: { type: "boolean", default: false },
"dry-run": { type: "boolean", default: false },
help: { type: "boolean", short: "h", default: false },
},
});
if (values.help) {
console.log(
"usage: triage-alerts.ts [--dry-run] [--alert-id ID] [--false-positive] " +
"[--ignore-reason TEXT] [--resolve]",
);
return 0;
}
const baseUrl = (values["base-url"] || process.env.STACKGEN_URL || "").trim();
const apiToken = (values["api-token"] || process.env.STACKGEN_TOKEN || "").trim();
const orgId = (
(values["org-id"] || values["workspace-id"]) ||
process.env.STACKGEN_WORKSPACE_ID ||
process.env.STACKGEN_ORG_ID ||
process.env.STACKGEN_PROJECT ||
process.env.STACKGEN_PROJECT_ID ||
""
).trim();
if (!baseUrl || !apiToken) {
console.error("set --base-url/--api-token or STACKGEN_URL / STACKGEN_TOKEN");
return 2;
}
const client = new StackgenClient(new StackgenConfig({ baseUrl, apiToken, workspaceId: orgId }));
let alertId = (values["alert-id"] ?? "").trim();
if (!alertId) {
const listed = await client.sre.listAlerts({
status: "active",
q: values.q || undefined,
page_size: 5,
});
show("alerts", listed);
const rows = alertRows(listed);
if (rows[0]?.id) {
alertId = String(rows[0].id);
}
}
show("groups", await client.sre.listInvestigationGroups({ page_size: 5 }));
if (values["dry-run"]) {
if (alertId) {
console.error(`alert_id=${alertId}`);
}
return 0;
}
if (!alertId) {
console.error("error: pass --alert-id or ensure listAlerts returns a row");
return 1;
}
if (values["false-positive"]) {
show(
"attention",
await client.sre.setAlertAttention(
alertId,
"false_positive",
values["ignore-reason"] || undefined,
),
);
} else if (values["ignore-reason"]) {
show(
"ignored",
await client.sre.ignoreAlert(alertId, values["ignore-reason"]),
);
} else if (!values.resolve) {
console.error(
"pass --false-positive, --ignore-reason TEXT, --resolve, or --dry-run",
);
return 2;
}
if (values.resolve) {
const inv = await client.sre.listInvestigations({
alert_id: alertId,
page_size: 50,
});
show("investigations", inv);
const rec = asRecord(inv);
const rows = Array.isArray(rec?.investigations)
? (rec.investigations as unknown[]).filter(
(r): r is Record<string, unknown> =>
!!r && typeof r === "object" && !Array.isArray(r),
)
: [];
const done = rows
.filter((r) => ["completed", "failed", "draft"].includes(String(r.status ?? "")))
.map((r) => String(r.id))
.filter(Boolean);
if (done.length) {
show("resolved", await client.sre.resolveInvestigations(done));
} else {
console.error("resolve: no completed/failed/draft threads");
}
}
return 0;
}
main().then(
(code) => process.exit(code),
(err) => {
console.error(`error: ${err}`);
if (err instanceof StackgenError && String(err).includes("401")) {
console.error(
"hint: SRE /app/sre routes need the edge to inject workspace identity.",
);
}
process.exit(1);
},
);// Triage an SRE alert and close finished investigation threads.
//
// Documented at https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-triage-alerts/
//
// go run ./examples/go/sre/triage-alerts.go --dry-run
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"strings"
stackgen "github.com/appcd-dev/stackgen-sdk/clients/go"
)
func envOr(keys ...string) string {
for _, key := range keys {
if v := strings.TrimSpace(os.Getenv(key)); v != "" {
return v
}
}
return ""
}
func main() {
baseURL := flag.String("base-url", "", "Mothership URL")
apiToken := flag.String("api-token", "", "API token")
orgID := flag.String("workspace-id", "", "Workspace UUID (sent as orgId)")
flag.StringVar(orgID, "org-id", *orgID, "Deprecated: use -workspace-id")
alertID := flag.String("alert-id", "", "Alert UUID")
query := flag.String("q", "", "Alert search when --alert-id is omitted")
ignoreReason := flag.String("ignore-reason", "", "Ignore reason")
falsePositive := flag.Bool("false-positive", false, "Set attention to false_positive")
resolve := flag.Bool("resolve", false, "Resolve finished investigations")
dryRun := flag.Bool("dry-run", false, "List candidates and exit")
flag.Parse()
base := strings.TrimSpace(*baseURL)
if base == "" {
base = envOr("STACKGEN_URL")
}
token := strings.TrimSpace(*apiToken)
if token == "" {
token = envOr("STACKGEN_TOKEN")
}
org := strings.TrimSpace(*orgID)
if org == "" {
org = envOr("STACKGEN_WORKSPACE_ID", "STACKGEN_ORG_ID", "STACKGEN_PROJECT", "STACKGEN_PROJECT_ID")
}
if base == "" || token == "" {
fmt.Fprintln(os.Stderr, "set STACKGEN_URL and STACKGEN_TOKEN")
os.Exit(2)
}
client, err := stackgen.NewClient(stackgen.Config{BaseURL: base, APIToken: token, WorkspaceID: org})
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
ctx := context.Background()
id := strings.TrimSpace(*alertID)
if id == "" {
q := map[string]string{"status": "active", "page_size": "5"}
if strings.TrimSpace(*query) != "" {
q["q"] = *query
}
listed, err := client.SRE.ListAlerts(ctx, q)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ := json.MarshalIndent(listed, "", " ")
fmt.Println("alerts:", string(raw))
if body, ok := listed.(map[string]any); ok {
items, _ := body["alerts"].([]any)
if len(items) == 0 {
items, _ = body["items"].([]any)
}
if len(items) > 0 {
if first, ok := items[0].(map[string]any); ok {
id, _ = first["id"].(string)
}
}
}
}
groups, err := client.SRE.ListInvestigationGroups(ctx, map[string]string{"page_size": "5"})
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ := json.MarshalIndent(groups, "", " ")
fmt.Println("groups:", string(raw))
if *dryRun {
if id != "" {
fmt.Fprintln(os.Stderr, "alert_id=", id)
}
return
}
if id == "" {
fmt.Fprintln(os.Stderr, "error: pass --alert-id or ensure list alerts returns a row")
os.Exit(1)
}
if *falsePositive {
updated, err := client.SRE.SetAlertAttention(ctx, id, "false_positive", *ignoreReason)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ = json.MarshalIndent(updated, "", " ")
fmt.Println("attention:", string(raw))
} else if strings.TrimSpace(*ignoreReason) != "" {
updated, err := client.SRE.IgnoreAlert(ctx, id, *ignoreReason)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ = json.MarshalIndent(updated, "", " ")
fmt.Println("ignored:", string(raw))
} else if !*resolve {
fmt.Fprintln(os.Stderr, "pass --false-positive, --ignore-reason TEXT, --resolve, or --dry-run")
os.Exit(2)
}
if *resolve {
inv, err := client.SRE.ListInvestigations(ctx, map[string]string{"alert_id": id, "page_size": "50"})
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ = json.MarshalIndent(inv, "", " ")
fmt.Println("investigations:", string(raw))
var done []string
if body, ok := inv.(map[string]any); ok {
for _, item := range body["investigations"].([]any) {
row, _ := item.(map[string]any)
status, _ := row["status"].(string)
rid, _ := row["id"].(string)
if rid != "" && (status == "completed" || status == "failed" || status == "draft") {
done = append(done, rid)
}
}
}
if len(done) == 0 {
fmt.Fprintln(os.Stderr, "resolve: no completed/failed/draft threads")
return
}
resolved, err := client.SRE.ResolveInvestigations(ctx, done)
if err != nil {
fmt.Fprintln(os.Stderr, "error:", err)
os.Exit(1)
}
raw, _ = json.MarshalIndent(resolved, "", " ")
fmt.Println("resolved:", string(raw))
}
}Notes
- Operators set status to
ignoredoractive. Resolution from the source sync setsresolved. Do not sendstatusandattentionin the same patch. - Attention values:
needs_attention,needs_review,false_positive,others. fork_investigationstarts a new thread that keeps the parent RCA context.list_remediation_checksreturns post-remediation verification rows.
Related: Investigate alert · Pull alerts