diff --git a/cmd/fanout-docgen/main.go b/cmd/fanout-docgen/main.go index 53874463..dd6e8027 100644 --- a/cmd/fanout-docgen/main.go +++ b/cmd/fanout-docgen/main.go @@ -45,7 +45,7 @@ import ( func main() { var ( source = flag.String("source", "internal/config/config.go", "path to the file declaring config.Config") - alerts = flag.String("alert-source", "internal/alert/types.go", "path to the file declaring alert.AlertEnv") + alerts = flag.String("alert-source", "internal/alert/types.go", "path to the file declaring alert.Env") // Every directory that registers routes. cmd/fanout carries the // operational, protocol and SPA routes; internal/agent the investigator. // A directory listed here that registers nothing is an error, so this @@ -178,16 +178,16 @@ func run(source, alertSource, routeDirs, outDir, envExample string, check bool) // without threading the slice back through run's return. var alertEnvCount []field -// collectAlertEnv reads alert.AlertEnv the way collect reads config.Config: +// collectAlertEnv reads alert.Env the way collect reads config.Config: // reflection for the contract, the source for the prose, and a cross-check that // refuses to proceed when the two disagree. func collectAlertEnv(source string) ([]field, error) { - docs, err := docComments(source, "AlertEnv") + docs, err := docComments(source, "Env") if err != nil { return nil, err } - typ := reflect.TypeOf(alert.AlertEnv{}) + typ := reflect.TypeOf(alert.Env{}) out := make([]field, 0, typ.NumField()) seen := make(map[string]bool, typ.NumField()) @@ -199,7 +199,7 @@ func collectAlertEnv(source string) ([]field, error) { name := sf.Tag.Get("expr") if name == "" { return nil, fmt.Errorf( - "alert.AlertEnv field %s has no expr tag, so a rule author has no name to write for it", + "alert.Env field %s has no expr tag, so a rule author has no name to write for it", sf.Name, ) } @@ -215,7 +215,7 @@ func collectAlertEnv(source string) ([]field, error) { for name := range docs { if !seen[name] { return nil, fmt.Errorf( - "%s declares field %s, which is not in alert.AlertEnv — is --alert-source correct?", + "%s declares field %s, which is not in alert.Env — is --alert-source correct?", source, name, ) } diff --git a/internal/agent/tools_integration_test.go b/internal/agent/tools_integration_test.go index c0c504dd..677e0dcc 100644 --- a/internal/agent/tools_integration_test.go +++ b/internal/agent/tools_integration_test.go @@ -19,8 +19,8 @@ type registryQueries struct{} type registryIntelligence struct{} -func (registryIntelligence) LatestSnapshot() *intelligence.IntelligenceSnapshot { - return &intelligence.IntelligenceSnapshot{Summary: "healthy", HealthScore: 100} +func (registryIntelligence) LatestSnapshot() *intelligence.Snapshot { + return &intelligence.Snapshot{Summary: "healthy", HealthScore: 100} } func (registryQueries) Dependencies(context.Context, observability.Scope, observability.DependencyOptions) (observability.Result[observability.Dependencies], error) { diff --git a/internal/alert/engine.go b/internal/alert/engine.go index 68a9e0b8..46717003 100644 --- a/internal/alert/engine.go +++ b/internal/alert/engine.go @@ -28,7 +28,7 @@ type Engine struct { interval time.Duration histDays int lastPrune time.Time - envOverride map[string]AlertEnv // for testing — bypasses DuckDB query + envOverride map[string]Env // for testing — bypasses DuckDB query } // NewEngine creates an Engine ready to Run. @@ -79,11 +79,11 @@ func (e *Engine) Run(ctx context.Context) { } } -// BuildAllEnvs returns all current AlertEnvs keyed by service name. -func (e *Engine) BuildAllEnvs(ctx context.Context) map[string]AlertEnv { +// BuildAllEnvs returns all current Envs keyed by service name. +func (e *Engine) BuildAllEnvs(ctx context.Context) map[string]Env { envs := e.buildEnvs(ctx) if envs == nil { - return map[string]AlertEnv{} + return map[string]Env{} } return envs } @@ -157,7 +157,7 @@ func (e *Engine) compileRules(rules []Rule) { // resolveServices returns the sorted list of services to evaluate for a rule. // '*' matches all services present in envs; a specific name targets just that one. -func (e *Engine) resolveServices(rule Rule, envs map[string]AlertEnv) []string { +func (e *Engine) resolveServices(rule Rule, envs map[string]Env) []string { if rule.Service == "" || rule.Service == "*" { svcs := make([]string, 0, len(envs)) for svc := range envs { @@ -175,7 +175,7 @@ func (e *Engine) resolveServices(rule Rule, envs map[string]AlertEnv) []string { } // transition implements the alert state machine for a (rule, service) pair. -func (e *Engine) transition(rule Rule, svc string, triggered bool, env AlertEnv) { +func (e *Engine) transition(rule Rule, svc string, triggered bool, env Env) { now := time.Now() nowStr := now.Format(time.RFC3339) @@ -339,7 +339,7 @@ func (e *Engine) transition(rule Rule, svc string, triggered bool, env AlertEnv) // primaryValue picks the most relevant metric value based on the rule expression. // If the expression mentions p95, use P95. If it mentions throughput, use that. // Default to error_rate as the most common alert metric. -func primaryValue(rule Rule, env AlertEnv) float64 { +func primaryValue(rule Rule, env Env) float64 { expr := rule.Expression if strings.Contains(expr, "p95") { return env.P95 @@ -357,7 +357,7 @@ func primaryValue(rule Rule, env AlertEnv) float64 { } // fireWebhookAsync fires the webhook for an alert in a goroutine and updates delivery status. -func (e *Engine) fireWebhookAsync(rule Rule, alert Alert, env AlertEnv, event string) { +func (e *Engine) fireWebhookAsync(rule Rule, alert Alert, env Env, event string) { go func() { defer func() { if r := recover(); r != nil { @@ -382,14 +382,14 @@ func (e *Engine) fireWebhookAsync(rule Rule, alert Alert, env AlertEnv, event st }() } -// buildEnvs returns AlertEnv keyed by service name. If envOverride is set it +// buildEnvs returns Env keyed by service name. If envOverride is set it // is returned directly (for tests). Otherwise DuckDB is queried. -func (e *Engine) buildEnvs(ctx context.Context) map[string]AlertEnv { +func (e *Engine) buildEnvs(ctx context.Context) map[string]Env { if e.envOverride != nil { return e.envOverride } if e.duck == nil { - return map[string]AlertEnv{} + return map[string]Env{} } const envSQL = ` @@ -440,13 +440,13 @@ FROM current c LEFT JOIN previous p ON c.service = p.service` } } - envs := make(map[string]AlertEnv, len(resp.Results)) + envs := make(map[string]Env, len(resp.Results)) for _, row := range resp.Results { svc, _ := row["service"].(string) if svc == "" { continue } - env := AlertEnv{ + env := Env{ Service: svc, ErrorRate: toFloat(row["error_rate"]), P50: toFloat(row["p50"]), diff --git a/internal/alert/engine_test.go b/internal/alert/engine_test.go index 1e3b5ce7..1cf3daa8 100644 --- a/internal/alert/engine_test.go +++ b/internal/alert/engine_test.go @@ -49,8 +49,8 @@ func TestEngine_Transition_NoneToFiring(t *testing.T) { rule := makeTestRule(t, store, "error_rate > 0.1", 0) - env := AlertEnv{Service: "svc-a", ErrorRate: 0.5} - eng.envOverride = map[string]AlertEnv{"svc-a": env} + env := Env{Service: "svc-a", ErrorRate: 0.5} + eng.envOverride = map[string]Env{"svc-a": env} if err := eng.RecompileRule(rule.ID, rule.Expression); err != nil { t.Fatalf("RecompileRule: %v", err) @@ -76,8 +76,8 @@ func TestEngine_Transition_NoneToPending(t *testing.T) { rule := makeTestRule(t, store, "error_rate > 0.1", 60) - env := AlertEnv{Service: "svc-a", ErrorRate: 0.5} - eng.envOverride = map[string]AlertEnv{"svc-a": env} + env := Env{Service: "svc-a", ErrorRate: 0.5} + eng.envOverride = map[string]Env{"svc-a": env} if err := eng.RecompileRule(rule.ID, rule.Expression); err != nil { t.Fatalf("RecompileRule: %v", err) @@ -104,8 +104,8 @@ func TestEngine_Transition_PendingClears(t *testing.T) { rule := makeTestRule(t, store, "error_rate > 0.1", 60) - env := AlertEnv{Service: "svc-a", ErrorRate: 0.5} - eng.envOverride = map[string]AlertEnv{"svc-a": env} + env := Env{Service: "svc-a", ErrorRate: 0.5} + eng.envOverride = map[string]Env{"svc-a": env} if err := eng.RecompileRule(rule.ID, rule.Expression); err != nil { t.Fatalf("RecompileRule: %v", err) @@ -138,8 +138,8 @@ func TestEngine_Transition_FiringToResolved(t *testing.T) { rule := makeTestRule(t, store, "error_rate > 0.1", 0) rule.NotifyOnResolve = true - env := AlertEnv{Service: "svc-a", ErrorRate: 0.5} - eng.envOverride = map[string]AlertEnv{"svc-a": env} + env := Env{Service: "svc-a", ErrorRate: 0.5} + eng.envOverride = map[string]Env{"svc-a": env} if err := eng.RecompileRule(rule.ID, rule.Expression); err != nil { t.Fatalf("RecompileRule: %v", err) @@ -157,7 +157,7 @@ func TestEngine_Transition_FiringToResolved(t *testing.T) { } // Resolve it. - envClear := AlertEnv{Service: "svc-a", ErrorRate: 0.01} + envClear := Env{Service: "svc-a", ErrorRate: 0.01} eng.transition(rule, "svc-a", false, envClear) a, err = store.GetAlert(rule.ID, "svc-a") @@ -197,7 +197,7 @@ func TestEngine_ResolveServices_Wildcard(t *testing.T) { eng, _ := newTestEngine(t) rule := Rule{Service: "*"} - envs := map[string]AlertEnv{ + envs := map[string]Env{ "svc-a": {}, "svc-b": {}, "svc-c": {}, @@ -220,7 +220,7 @@ func TestEngine_ResolveServices_Specific(t *testing.T) { eng, _ := newTestEngine(t) rule := Rule{Service: "svc-b"} - envs := map[string]AlertEnv{ + envs := map[string]Env{ "svc-a": {}, "svc-b": {}, "svc-c": {}, @@ -250,7 +250,7 @@ func TestEngine_RunOnce(t *testing.T) { t.Fatalf("CreateRule: %v", err) } - eng.envOverride = map[string]AlertEnv{ + eng.envOverride = map[string]Env{ "api": {Service: "api", ErrorRate: 0.20}, } diff --git a/internal/alert/expr.go b/internal/alert/expr.go index 5804fbf2..15fbe08f 100644 --- a/internal/alert/expr.go +++ b/internal/alert/expr.go @@ -14,7 +14,7 @@ import ( // the package (and callers) refer only to alert.Program. type Program = cel.Program -// alertCELEnv is the CEL environment shared by every compile. Each AlertEnv field +// alertCELEnv is the CEL environment shared by every compile. Each Env field // is declared as a typed variable so the type checker rejects unknown names and // non-boolean expressions at compile time. CrossTypeNumericComparisons lets a // double variable be compared against an integer literal (e.g. `p95 > 500`) @@ -62,12 +62,12 @@ func CompileExpression(expression string) (Program, error) { return prog, nil } -// celActivation binds the CEL variables to an AlertEnv's values. The keys must +// celActivation binds the CEL variables to an Env's values. The keys must // stay in sync with the cel.Variable declarations in mustAlertCELEnv above; -// TestAlertEnvVariablesDeclaredAndBound enforces that against the AlertEnv `expr` +// TestEnvVariablesDeclaredAndBound enforces that against the Env `expr` // field tags, so a drift fails the build instead of silently breaking a rule at // eval time. -func celActivation(env AlertEnv) map[string]any { +func celActivation(env Env) map[string]any { return map[string]any{ "error_rate": env.ErrorRate, "p50": env.P50, @@ -84,7 +84,7 @@ func celActivation(env AlertEnv) map[string]any { } // EvalExpression runs a compiled program with the given environment. -func EvalExpression(prog Program, env AlertEnv) (bool, error) { +func EvalExpression(prog Program, env Env) (bool, error) { out, _, err := prog.Eval(celActivation(env)) if err != nil { return false, fmt.Errorf("eval expression: %w", err) @@ -98,7 +98,7 @@ func EvalExpression(prog Program, env AlertEnv) (bool, error) { // SafeEval wraps EvalExpression with panic recovery, logging unexpected panics // via slog. Returns false on panic. -func SafeEval(prog Program, env AlertEnv) (result bool, err error) { +func SafeEval(prog Program, env Env) (result bool, err error) { defer func() { if r := recover(); r != nil { slog.Error("alert: expression eval panic", "panic", r, "stack", string(debug.Stack())) diff --git a/internal/alert/expr_test.go b/internal/alert/expr_test.go index 6861d857..0f1fdb22 100644 --- a/internal/alert/expr_test.go +++ b/internal/alert/expr_test.go @@ -71,13 +71,13 @@ func TestEvalExpression(t *testing.T) { cases := []struct { name string - env AlertEnv + env Env expected bool }{ - {"both above threshold", AlertEnv{ErrorRate: 0.1, P95: 600}, true}, - {"error_rate below", AlertEnv{ErrorRate: 0.01, P95: 600}, false}, - {"p95 below", AlertEnv{ErrorRate: 0.1, P95: 100}, false}, - {"both below", AlertEnv{ErrorRate: 0.01, P95: 100}, false}, + {"both above threshold", Env{ErrorRate: 0.1, P95: 600}, true}, + {"error_rate below", Env{ErrorRate: 0.01, P95: 600}, false}, + {"p95 below", Env{ErrorRate: 0.1, P95: 100}, false}, + {"both below", Env{ErrorRate: 0.01, P95: 100}, false}, } for _, tc := range cases { @@ -99,7 +99,7 @@ func TestSafeEval_NoPanic(t *testing.T) { t.Fatalf("CompileExpression: %v", err) } - env := AlertEnv{ErrorRate: 0.5} + env := Env{ErrorRate: 0.5} result, err := SafeEval(prog, env) if err != nil { t.Fatalf("SafeEval: unexpected error: %v", err) @@ -109,29 +109,29 @@ func TestSafeEval_NoPanic(t *testing.T) { } } -// TestAlertEnvVariablesDeclaredAndBound makes the AlertEnv struct the single +// TestEnvVariablesDeclaredAndBound makes the Env struct the single // source of truth for alert variables: every field's `expr` tag must be both a // declared CEL variable (so a rule can reference it) and a key in celActivation // (so it's bound at eval). Without this, adding a field to the struct + the // cel.Variable list but forgetting celActivation would compile fine yet fail // every 30s eval cycle with "no such attribute" — a silent, partial failure. -func TestAlertEnvVariablesDeclaredAndBound(t *testing.T) { - act := celActivation(AlertEnv{}) - tp := reflect.TypeOf(AlertEnv{}) +func TestEnvVariablesDeclaredAndBound(t *testing.T) { + act := celActivation(Env{}) + tp := reflect.TypeOf(Env{}) tags := map[string]bool{} for i := 0; i < tp.NumField(); i++ { f := tp.Field(i) name := f.Tag.Get("expr") if name == "" { - t.Errorf("AlertEnv.%s has no `expr` tag", f.Name) + t.Errorf("Env.%s has no `expr` tag", f.Name) continue } tags[name] = true // (1) bound at eval time. if _, ok := act[name]; !ok { - t.Errorf("variable %q (AlertEnv.%s) is not bound in celActivation", name, f.Name) + t.Errorf("variable %q (Env.%s) is not bound in celActivation", name, f.Name) } // (2) declared in the CEL env — a type-appropriate boolean expression @@ -143,18 +143,18 @@ func TestAlertEnvVariablesDeclaredAndBound(t *testing.T) { case reflect.String: expr = name + ` == ""` default: - t.Errorf("AlertEnv.%s has unsupported kind %s for a CEL variable", f.Name, f.Type.Kind()) + t.Errorf("Env.%s has unsupported kind %s for a CEL variable", f.Name, f.Type.Kind()) continue } if _, err := CompileExpression(expr); err != nil { - t.Errorf("variable %q (AlertEnv.%s) not declared/usable in the CEL env: %v", name, f.Name, err) + t.Errorf("variable %q (Env.%s) not declared/usable in the CEL env: %v", name, f.Name, err) } } // Reverse direction: no activation key without a backing struct field. for k := range act { if !tags[k] { - t.Errorf("celActivation binds %q with no matching AlertEnv `expr` tag", k) + t.Errorf("celActivation binds %q with no matching Env `expr` tag", k) } } } @@ -169,12 +169,12 @@ func TestEvalExpression_StringVar(t *testing.T) { } cases := []struct { name string - env AlertEnv + env Env want bool }{ - {"match + over threshold", AlertEnv{Service: "checkout", ErrorRate: 0.1}, true}, - {"service mismatch", AlertEnv{Service: "cart", ErrorRate: 0.1}, false}, - {"match but under threshold", AlertEnv{Service: "checkout", ErrorRate: 0.01}, false}, + {"match + over threshold", Env{Service: "checkout", ErrorRate: 0.1}, true}, + {"service mismatch", Env{Service: "cart", ErrorRate: 0.1}, false}, + {"match but under threshold", Env{Service: "checkout", ErrorRate: 0.01}, false}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/internal/alert/store.go b/internal/alert/store.go index 51bf58cb..1768400d 100644 --- a/internal/alert/store.go +++ b/internal/alert/store.go @@ -45,16 +45,16 @@ func ruleToCreateParams(r Rule) generated.CreateRuleParams { ForSeconds: int64(r.ForSeconds), CooldownS: int64(r.CooldownS), RepeatIntervalS: int64(r.RepeatIntervalS), - WebhookUrl: nullString(r.WebhookURL), + WebhookURL: nullString(r.WebhookURL), WebhookHeaders: nullString(r.WebhookHeaders), WebhookTemplate: nullString(r.WebhookTemplate), NotifyOnResolve: boolToInt64(r.NotifyOnResolve), } } -// ruleToUpdateParams converts a domain Rule to generated UpdateRuleParams. -func ruleToUpdateParams(r Rule) generated.UpdateRuleParams { - return generated.UpdateRuleParams{ +// ruleToReplaceParams converts a domain Rule to generated ReplaceRuleParams. +func ruleToReplaceParams(r Rule) generated.ReplaceRuleParams { + return generated.ReplaceRuleParams{ ID: r.ID, Name: r.Name, Description: nullString(r.Description), @@ -65,7 +65,7 @@ func ruleToUpdateParams(r Rule) generated.UpdateRuleParams { ForSeconds: int64(r.ForSeconds), CooldownS: int64(r.CooldownS), RepeatIntervalS: int64(r.RepeatIntervalS), - WebhookUrl: nullString(r.WebhookURL), + WebhookURL: nullString(r.WebhookURL), WebhookHeaders: nullString(r.WebhookHeaders), WebhookTemplate: nullString(r.WebhookTemplate), NotifyOnResolve: boolToInt64(r.NotifyOnResolve), @@ -85,7 +85,7 @@ func alertRuleToRule(ar generated.AlertRule) Rule { ForSeconds: int(ar.ForSeconds), CooldownS: int(ar.CooldownS), RepeatIntervalS: int(ar.RepeatIntervalS), - WebhookURL: ar.WebhookUrl.String, + WebhookURL: ar.WebhookURL.String, WebhookHeaders: ar.WebhookHeaders.String, WebhookTemplate: ar.WebhookTemplate.String, NotifyOnResolve: ar.NotifyOnResolve != 0, @@ -184,9 +184,9 @@ func (s *Store) ListEnabledRules() ([]Rule, error) { return rules, nil } -// UpdateRule updates a rule by ID and returns the updated record. -func (s *Store) UpdateRule(r Rule) (Rule, error) { - ar, err := s.q.UpdateRule(context.Background(), ruleToUpdateParams(r)) +// ReplaceRule replaces a rule by ID and returns the persisted record. +func (s *Store) ReplaceRule(r Rule) (Rule, error) { + ar, err := s.q.ReplaceRule(context.Background(), ruleToReplaceParams(r)) if errors.Is(err, sql.ErrNoRows) { return Rule{}, fmt.Errorf("store: rule %q not found", r.ID) } @@ -333,12 +333,12 @@ func (s *Store) UpdateDeliveryStatus(ruleID, service, status, deliveredAt string } // AlertSummary returns counts of alerts grouped by state. -func (s *Store) AlertSummary() (AlertSummary, error) { +func (s *Store) AlertSummary() (Summary, error) { rows, err := s.q.AlertSummary(context.Background()) if err != nil { - return AlertSummary{}, fmt.Errorf("store: alert summary: %w", err) + return Summary{}, fmt.Errorf("store: alert summary: %w", err) } - var sum AlertSummary + var sum Summary for _, row := range rows { switch row.State { case "firing": diff --git a/internal/alert/store_test.go b/internal/alert/store_test.go index b162a0c0..04cf336a 100644 --- a/internal/alert/store_test.go +++ b/internal/alert/store_test.go @@ -79,7 +79,7 @@ func TestStore_ListRules(t *testing.T) { } } -func TestStore_UpdateRule(t *testing.T) { +func TestStore_ReplaceRule(t *testing.T) { s := newTestStore(t) r, err := s.CreateRule(makeRule("update-me")) @@ -90,9 +90,9 @@ func TestStore_UpdateRule(t *testing.T) { r.Name = "updated-name" r.Enabled = false - updated, err := s.UpdateRule(r) + updated, err := s.ReplaceRule(r) if err != nil { - t.Fatalf("UpdateRule: %v", err) + t.Fatalf("ReplaceRule: %v", err) } if updated.Name != "updated-name" { t.Errorf("Name = %q, want %q", updated.Name, "updated-name") @@ -127,8 +127,8 @@ func TestStore_ListEnabledRules(t *testing.T) { r2, _ := s.CreateRule(makeRule("disabled")) r2.Enabled = false - if _, err := s.UpdateRule(r2); err != nil { - t.Fatalf("UpdateRule: %v", err) + if _, err := s.ReplaceRule(r2); err != nil { + t.Fatalf("ReplaceRule: %v", err) } enabled, err := s.ListEnabledRules() diff --git a/internal/alert/types.go b/internal/alert/types.go index f791c904..a7fb7f40 100644 --- a/internal/alert/types.go +++ b/internal/alert/types.go @@ -38,13 +38,13 @@ type Alert struct { CreatedAt string `json:"created_at"` } -// AlertEnv holds the metric values available to alert expressions. +// Env holds the metric values available to alert expressions. // // The `expr` tag is the name a rule author writes, and each field's doc comment // is published as that variable's description by cmd/fanout-docgen. Adding a // field here adds a row to the reference; describe it in a sentence a rule // author would find useful, not in terms of how it is computed. -type AlertEnv struct { +type Env struct { // Mean of minute-rollup span error proportions over the evaluation window, // from 0 to 1; not a request-weighted whole-window error rate. ErrorRate float64 `expr:"error_rate"` @@ -81,8 +81,8 @@ type AlertEnv struct { Service string `expr:"service"` } -// AlertSummary aggregates alert state counts. -type AlertSummary struct { +// Summary aggregates alert state counts. +type Summary struct { Firing int `json:"firing"` Pending int `json:"pending"` Resolved int `json:"resolved"` @@ -92,7 +92,7 @@ type AlertSummary struct { type ActionContext struct { Rule Rule `json:"rule"` Alert Alert `json:"alert"` - Env AlertEnv `json:"env"` + Env Env `json:"env"` Event string `json:"event"` Time time.Time `json:"time"` } diff --git a/internal/api/alerts.go b/internal/api/alerts.go index df942deb..518409cd 100644 --- a/internal/api/alerts.go +++ b/internal/api/alerts.go @@ -20,7 +20,7 @@ func RegisterAlertRoutes(e *echo.Echo, store *alert.Store, engine *alert.Engine) e.GET("/api/alerting/summary", h.AlertSummary, read) e.GET("/api/alerting/rules", h.ListRules, read) e.POST("/api/alerting/rules", h.CreateRule, manage) - e.PUT("/api/alerting/rules/:id", h.UpdateRule, manage) + e.PUT("/api/alerting/rules/:id", h.ReplaceRule, manage) e.DELETE("/api/alerting/rules/:id", h.DeleteRule, manage) e.POST("/api/alerting/rules/:id/test", h.TestRule, manage) } @@ -53,7 +53,7 @@ func (h *alertHandler) ListAlerts(c *echo.Context) error { // AlertSummary returns firing/pending/resolved counts. func (h *alertHandler) AlertSummary(c *echo.Context) error { if h.store == nil { - return c.JSON(200, alert.AlertSummary{}) + return c.JSON(200, alert.Summary{}) } summary, err := h.store.AlertSummary() if err != nil { @@ -114,8 +114,8 @@ func (h *alertHandler) CreateRule(c *echo.Context) error { return c.JSON(201, created) } -// UpdateRule updates an existing alert rule. -func (h *alertHandler) UpdateRule(c *echo.Context) error { +// ReplaceRule replaces an existing alert rule. +func (h *alertHandler) ReplaceRule(c *echo.Context) error { if h.store == nil { return echo.NewHTTPError(http.StatusServiceUnavailable, "alerts not configured") } @@ -133,7 +133,7 @@ func (h *alertHandler) UpdateRule(c *echo.Context) error { } } - updated, err := h.store.UpdateRule(req) + updated, err := h.store.ReplaceRule(req) if err != nil { if strings.Contains(err.Error(), "not found") { return echo.NewHTTPError(http.StatusNotFound, "rule not found") @@ -209,10 +209,10 @@ func (h *alertHandler) TestRule(c *echo.Context) error { envs := h.engine.BuildAllEnvs(c.Request().Context()) type testResult struct { - Service string `json:"service"` - Triggered bool `json:"triggered"` - Error string `json:"error,omitempty"` - Env alert.AlertEnv `json:"env"` + Service string `json:"service"` + Triggered bool `json:"triggered"` + Error string `json:"error,omitempty"` + Env alert.Env `json:"env"` } var results []testResult @@ -237,7 +237,7 @@ func (h *alertHandler) TestRule(c *echo.Context) error { }) } -func resolveTestServices(rule alert.Rule, envs map[string]alert.AlertEnv) []string { +func resolveTestServices(rule alert.Rule, envs map[string]alert.Env) []string { if rule.Service == "" || rule.Service == "*" { svcs := make([]string, 0, len(envs)) for svc := range envs { diff --git a/internal/api/intelligence.go b/internal/api/intelligence.go index 5bc32f5d..45573944 100644 --- a/internal/api/intelligence.go +++ b/internal/api/intelligence.go @@ -11,7 +11,7 @@ import ( // other consumers. Keeping the interface narrow makes the endpoint independent // of how snapshots are generated or retained. type IntelligenceSnapshots interface { - LatestSnapshot() *intelligence.IntelligenceSnapshot + LatestSnapshot() *intelligence.Snapshot } // RegisterIntelligenceRoutes exposes the detector's latest completed snapshot. diff --git a/internal/api/intelligence_test.go b/internal/api/intelligence_test.go index 05318ff5..347788d2 100644 --- a/internal/api/intelligence_test.go +++ b/internal/api/intelligence_test.go @@ -12,15 +12,15 @@ import ( ) type fakeIntelligenceSnapshots struct { - snapshot *intelligence.IntelligenceSnapshot + snapshot *intelligence.Snapshot } -func (f fakeIntelligenceSnapshots) LatestSnapshot() *intelligence.IntelligenceSnapshot { +func (f fakeIntelligenceSnapshots) LatestSnapshot() *intelligence.Snapshot { return f.snapshot } func TestIntelligenceLatestReturnsSnapshot(t *testing.T) { - want := &intelligence.IntelligenceSnapshot{ + want := &intelligence.Snapshot{ GeneratedAt: time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC), Timeframe: "last_15m", Summary: "System healthy (98/100).", @@ -38,7 +38,7 @@ func TestIntelligenceLatestReturnsSnapshot(t *testing.T) { if rec.Code != http.StatusOK { t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String()) } - var got intelligence.IntelligenceSnapshot + var got intelligence.Snapshot if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { t.Fatalf("decode response: %v", err) } diff --git a/internal/dashboard/service.go b/internal/dashboard/service.go index 6875fcf3..3de6e71b 100644 --- a/internal/dashboard/service.go +++ b/internal/dashboard/service.go @@ -107,7 +107,7 @@ func (s *Service) VersionRecord(ctx context.Context, owner, id string, version i return VersionRecord{}, err } var spec panel.Dashboard - if err := json.Unmarshal([]byte(r.SpecJson), &spec); err != nil { + if err := json.Unmarshal([]byte(r.SpecJSON), &spec); err != nil { return VersionRecord{}, err } record := Record{ID: r.ID, Name: spec.Name, Description: spec.Description, IsDefault: r.IsDefault == 1, Version: int(r.Version), Spec: spec, CreatedAt: r.DashboardCreatedAt, UpdatedAt: r.VersionCreatedAt} @@ -121,7 +121,7 @@ func (s *Service) VersionRecord(ctx context.Context, owner, id string, version i if err != nil { return VersionRecord{}, err } - if err := json.Unmarshal([]byte(prior.SpecJson), &before); err != nil { + if err := json.Unmarshal([]byte(prior.SpecJSON), &before); err != nil { return VersionRecord{}, err } } @@ -185,7 +185,7 @@ func (s *Service) get(ctx context.Context, ownerID, id string) (Record, error) { func decodeRecord(row generated.Dashboard) (Record, error) { record := Record{ID: row.ID, Name: row.Name, Description: row.Description, IsDefault: row.IsDefault == 1, Version: int(row.Version), CreatedAt: row.CreatedAt, UpdatedAt: row.UpdatedAt} - if err := json.Unmarshal([]byte(row.SpecJson), &record.Spec); err != nil { + if err := json.Unmarshal([]byte(row.SpecJSON), &record.Spec); err != nil { return Record{}, fmt.Errorf("decode dashboard %s: %w", row.ID, err) } return record, nil @@ -252,7 +252,7 @@ func (s *Service) CreateWithChanges(ctx context.Context, ownerID string, spec pa defer func() { _ = tx.Rollback() }() q := generated.New(tx) now := s.now().UTC().Format(time.RFC3339Nano) - affected, err := q.InsertDashboardBelowOwnerLimit(ctx, generated.InsertDashboardBelowOwnerLimitParams{ID: id, OwnerID: ownerID, Name: spec.Name, Description: spec.Description, Version: 1, SpecJson: string(raw), PanelCount: int64(len(spec.Panels)), CreatedAt: now, UpdatedAt: now}) + affected, err := q.InsertDashboardBelowOwnerLimit(ctx, generated.InsertDashboardBelowOwnerLimitParams{ID: id, OwnerID: ownerID, Name: spec.Name, Description: spec.Description, Version: 1, SpecJSON: string(raw), PanelCount: int64(len(spec.Panels)), CreatedAt: now, UpdatedAt: now}) if err != nil { if isUnique(err) { return Mutation{}, ErrConflict @@ -274,7 +274,7 @@ func (s *Service) CreateWithChanges(ctx context.Context, ownerID string, spec pa return Mutation{}, errors.New("dashboard build origin thread is not owned by the dashboard owner") } } - if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: id, Version: 1, SpecJson: string(raw), AuthorKind: author.Kind, AuthorID: author.ID, Message: "Created", CreatedAt: now}); err != nil { + if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: id, Version: 1, SpecJSON: string(raw), AuthorKind: author.Kind, AuthorID: author.ID, Message: "Created", CreatedAt: now}); err != nil { return Mutation{}, err } row, err := q.GetDashboard(ctx, generated.GetDashboardParams{ID: id, OwnerID: ownerID}) @@ -432,7 +432,7 @@ func (s *Service) write(ctx context.Context, ownerID, id string, base int, spec q := generated.New(tx) now := s.now().UTC().Format(time.RFC3339Nano) next := int64(base + 1) - affected, err := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: spec.Name, Description: spec.Description, NextVersion: next, SpecJson: raw, PanelCount: int64(len(spec.Panels)), UpdatedAt: now, ID: id, OwnerID: ownerID, BaseVersion: int64(base)}) + affected, err := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: spec.Name, Description: spec.Description, NextVersion: next, SpecJSON: raw, PanelCount: int64(len(spec.Panels)), UpdatedAt: now, ID: id, OwnerID: ownerID, BaseVersion: int64(base)}) if err != nil { if isUnique(err) { return Record{}, false, ErrConflict @@ -442,7 +442,7 @@ func (s *Service) write(ctx context.Context, ownerID, id string, base int, spec if affected == 0 { return Record{}, false, nil } - if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: id, Version: next, SpecJson: raw, AuthorKind: author.Kind, AuthorID: author.ID, Message: strings.TrimSpace(message), CreatedAt: now}); err != nil { + if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: id, Version: next, SpecJSON: raw, AuthorKind: author.Kind, AuthorID: author.ID, Message: strings.TrimSpace(message), CreatedAt: now}); err != nil { return Record{}, false, err } if err := q.PruneDashboardVersions(ctx, generated.PruneDashboardVersionsParams{DashboardID: id, KeepFrom: next - keepVersions + 1}); err != nil { diff --git a/internal/db/dashboards_schema_test.go b/internal/db/dashboards_schema_test.go index cfe16f5e..c7a8b61e 100644 --- a/internal/db/dashboards_schema_test.go +++ b/internal/db/dashboards_schema_test.go @@ -18,17 +18,17 @@ func TestDashboardVersionsCascadeAndOptimisticUpdate(t *testing.T) { } q := generated.New(sqlite.DB) ctx := t.Context() - if n, err := q.InsertDashboardBelowOwnerLimit(ctx, generated.InsertDashboardBelowOwnerLimitParams{ID: "d1", OwnerID: "u1", Name: "Ops", Version: 1, SpecJson: "{}", PanelCount: 0, CreatedAt: "t", UpdatedAt: "t"}); err != nil || n != 1 { + if n, err := q.InsertDashboardBelowOwnerLimit(ctx, generated.InsertDashboardBelowOwnerLimitParams{ID: "d1", OwnerID: "u1", Name: "Ops", Version: 1, SpecJSON: "{}", PanelCount: 0, CreatedAt: "t", UpdatedAt: "t"}); err != nil || n != 1 { t.Fatal(err) } - if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: "d1", Version: 1, SpecJson: "{}", AuthorKind: "system", CreatedAt: "t"}); err != nil { + if err := q.InsertDashboardVersion(ctx, generated.InsertDashboardVersionParams{DashboardID: "d1", Version: 1, SpecJSON: "{}", AuthorKind: "system", CreatedAt: "t"}); err != nil { t.Fatal(err) } - n, err := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: "Ops", NextVersion: 2, SpecJson: "{}", UpdatedAt: "t2", ID: "d1", OwnerID: "u1", BaseVersion: 1}) + n, err := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: "Ops", NextVersion: 2, SpecJSON: "{}", UpdatedAt: "t2", ID: "d1", OwnerID: "u1", BaseVersion: 1}) if err != nil || n != 1 { t.Fatalf("update = %d %v", n, err) } - if n, _ := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: "Ops", NextVersion: 2, SpecJson: "{}", UpdatedAt: "t3", ID: "d1", OwnerID: "u1", BaseVersion: 1}); n != 0 { + if n, _ := q.UpdateDashboard(ctx, generated.UpdateDashboardParams{Name: "Ops", NextVersion: 2, SpecJSON: "{}", UpdatedAt: "t3", ID: "d1", OwnerID: "u1", BaseVersion: 1}); n != 0 { t.Fatal("a stale base version updated the row") } if _, err := sqlite.DB.Exec(`DELETE FROM users WHERE id = 'u1'`); err != nil { diff --git a/internal/db/generated/dashboards.sql.go b/internal/db/generated/dashboards.sql.go index 840c3a6b..924ff8c2 100644 --- a/internal/db/generated/dashboards.sql.go +++ b/internal/db/generated/dashboards.sql.go @@ -57,7 +57,7 @@ func (q *Queries) GetDashboard(ctx context.Context, arg GetDashboardParams) (Das &i.Description, &i.IsDefault, &i.Version, - &i.SpecJson, + &i.SpecJSON, &i.PanelCount, &i.CreatedAt, &i.UpdatedAt, @@ -102,7 +102,7 @@ type GetDashboardVersionRecordRow struct { IsDefault int64 `json:"is_default"` DashboardCreatedAt string `json:"dashboard_created_at"` Version int64 `json:"version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` AuthorKind string `json:"author_kind"` AuthorID string `json:"author_id"` Message string `json:"message"` @@ -117,7 +117,7 @@ func (q *Queries) GetDashboardVersionRecord(ctx context.Context, arg GetDashboar &i.IsDefault, &i.DashboardCreatedAt, &i.Version, - &i.SpecJson, + &i.SpecJSON, &i.AuthorKind, &i.AuthorID, &i.Message, @@ -141,7 +141,7 @@ type InsertDashboardBelowOwnerLimitParams struct { Name string `json:"name"` Description string `json:"description"` Version int64 `json:"version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` PanelCount int64 `json:"panel_count"` CreatedAt string `json:"created_at"` UpdatedAt string `json:"updated_at"` @@ -154,7 +154,7 @@ func (q *Queries) InsertDashboardBelowOwnerLimit(ctx context.Context, arg Insert arg.Name, arg.Description, arg.Version, - arg.SpecJson, + arg.SpecJSON, arg.PanelCount, arg.CreatedAt, arg.UpdatedAt, @@ -202,7 +202,7 @@ VALUES (?, ?, ?, ?, ?, ?, ?) type InsertDashboardVersionParams struct { DashboardID string `json:"dashboard_id"` Version int64 `json:"version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` AuthorKind string `json:"author_kind"` AuthorID string `json:"author_id"` Message string `json:"message"` @@ -213,7 +213,7 @@ func (q *Queries) InsertDashboardVersion(ctx context.Context, arg InsertDashboar _, err := q.db.ExecContext(ctx, insertDashboardVersion, arg.DashboardID, arg.Version, - arg.SpecJson, + arg.SpecJSON, arg.AuthorKind, arg.AuthorID, arg.Message, @@ -363,7 +363,7 @@ type UpdateDashboardParams struct { Name string `json:"name"` Description string `json:"description"` NextVersion int64 `json:"next_version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` PanelCount int64 `json:"panel_count"` UpdatedAt string `json:"updated_at"` ID string `json:"id"` @@ -376,7 +376,7 @@ func (q *Queries) UpdateDashboard(ctx context.Context, arg UpdateDashboardParams arg.Name, arg.Description, arg.NextVersion, - arg.SpecJson, + arg.SpecJSON, arg.PanelCount, arg.UpdatedAt, arg.ID, diff --git a/internal/db/generated/models.go b/internal/db/generated/models.go index 4aed7e70..f4de3145 100644 --- a/internal/db/generated/models.go +++ b/internal/db/generated/models.go @@ -56,7 +56,7 @@ type AlertRule struct { ForSeconds int64 `json:"for_seconds"` CooldownS int64 `json:"cooldown_s"` RepeatIntervalS int64 `json:"repeat_interval_s"` - WebhookUrl sql.NullString `json:"webhook_url"` + WebhookURL sql.NullString `json:"webhook_url"` WebhookHeaders sql.NullString `json:"webhook_headers"` WebhookTemplate sql.NullString `json:"webhook_template"` NotifyOnResolve int64 `json:"notify_on_resolve"` @@ -84,7 +84,7 @@ type Dashboard struct { Description string `json:"description"` IsDefault int64 `json:"is_default"` Version int64 `json:"version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` PanelCount int64 `json:"panel_count"` CreatedAt string `json:"created_at"` UpdatedAt string `json:"updated_at"` @@ -100,7 +100,7 @@ type DashboardOrigin struct { type DashboardVersion struct { DashboardID string `json:"dashboard_id"` Version int64 `json:"version"` - SpecJson string `json:"spec_json"` + SpecJSON string `json:"spec_json"` AuthorKind string `json:"author_kind"` AuthorID string `json:"author_id"` Message string `json:"message"` diff --git a/internal/db/generated/rules.sql.go b/internal/db/generated/rules.sql.go index 5ea57217..97475c0a 100644 --- a/internal/db/generated/rules.sql.go +++ b/internal/db/generated/rules.sql.go @@ -30,7 +30,7 @@ type CreateRuleParams struct { ForSeconds int64 `json:"for_seconds"` CooldownS int64 `json:"cooldown_s"` RepeatIntervalS int64 `json:"repeat_interval_s"` - WebhookUrl sql.NullString `json:"webhook_url"` + WebhookURL sql.NullString `json:"webhook_url"` WebhookHeaders sql.NullString `json:"webhook_headers"` WebhookTemplate sql.NullString `json:"webhook_template"` NotifyOnResolve int64 `json:"notify_on_resolve"` @@ -48,7 +48,7 @@ func (q *Queries) CreateRule(ctx context.Context, arg CreateRuleParams) (AlertRu arg.ForSeconds, arg.CooldownS, arg.RepeatIntervalS, - arg.WebhookUrl, + arg.WebhookURL, arg.WebhookHeaders, arg.WebhookTemplate, arg.NotifyOnResolve, @@ -65,7 +65,7 @@ func (q *Queries) CreateRule(ctx context.Context, arg CreateRuleParams) (AlertRu &i.ForSeconds, &i.CooldownS, &i.RepeatIntervalS, - &i.WebhookUrl, + &i.WebhookURL, &i.WebhookHeaders, &i.WebhookTemplate, &i.NotifyOnResolve, @@ -102,7 +102,7 @@ func (q *Queries) GetRule(ctx context.Context, id string) (AlertRule, error) { &i.ForSeconds, &i.CooldownS, &i.RepeatIntervalS, - &i.WebhookUrl, + &i.WebhookURL, &i.WebhookHeaders, &i.WebhookTemplate, &i.NotifyOnResolve, @@ -136,7 +136,7 @@ func (q *Queries) ListEnabledRules(ctx context.Context) ([]AlertRule, error) { &i.ForSeconds, &i.CooldownS, &i.RepeatIntervalS, - &i.WebhookUrl, + &i.WebhookURL, &i.WebhookHeaders, &i.WebhookTemplate, &i.NotifyOnResolve, @@ -180,7 +180,7 @@ func (q *Queries) ListRules(ctx context.Context) ([]AlertRule, error) { &i.ForSeconds, &i.CooldownS, &i.RepeatIntervalS, - &i.WebhookUrl, + &i.WebhookURL, &i.WebhookHeaders, &i.WebhookTemplate, &i.NotifyOnResolve, @@ -200,7 +200,7 @@ func (q *Queries) ListRules(ctx context.Context) ([]AlertRule, error) { return items, nil } -const updateRule = `-- name: UpdateRule :one +const replaceRule = `-- name: ReplaceRule :one UPDATE alert_rules SET name = ?, description = ?, enabled = ?, service = ?, namespace = ?, expression = ?, for_seconds = ?, cooldown_s = ?, repeat_interval_s = ?, @@ -210,7 +210,7 @@ WHERE id = ? RETURNING id, name, description, enabled, service, namespace, expression, for_seconds, cooldown_s, repeat_interval_s, webhook_url, webhook_headers, webhook_template, notify_on_resolve, created_at, updated_at ` -type UpdateRuleParams struct { +type ReplaceRuleParams struct { Name string `json:"name"` Description sql.NullString `json:"description"` Enabled int64 `json:"enabled"` @@ -220,15 +220,15 @@ type UpdateRuleParams struct { ForSeconds int64 `json:"for_seconds"` CooldownS int64 `json:"cooldown_s"` RepeatIntervalS int64 `json:"repeat_interval_s"` - WebhookUrl sql.NullString `json:"webhook_url"` + WebhookURL sql.NullString `json:"webhook_url"` WebhookHeaders sql.NullString `json:"webhook_headers"` WebhookTemplate sql.NullString `json:"webhook_template"` NotifyOnResolve int64 `json:"notify_on_resolve"` ID string `json:"id"` } -func (q *Queries) UpdateRule(ctx context.Context, arg UpdateRuleParams) (AlertRule, error) { - row := q.db.QueryRowContext(ctx, updateRule, +func (q *Queries) ReplaceRule(ctx context.Context, arg ReplaceRuleParams) (AlertRule, error) { + row := q.db.QueryRowContext(ctx, replaceRule, arg.Name, arg.Description, arg.Enabled, @@ -238,7 +238,7 @@ func (q *Queries) UpdateRule(ctx context.Context, arg UpdateRuleParams) (AlertRu arg.ForSeconds, arg.CooldownS, arg.RepeatIntervalS, - arg.WebhookUrl, + arg.WebhookURL, arg.WebhookHeaders, arg.WebhookTemplate, arg.NotifyOnResolve, @@ -256,7 +256,7 @@ func (q *Queries) UpdateRule(ctx context.Context, arg UpdateRuleParams) (AlertRu &i.ForSeconds, &i.CooldownS, &i.RepeatIntervalS, - &i.WebhookUrl, + &i.WebhookURL, &i.WebhookHeaders, &i.WebhookTemplate, &i.NotifyOnResolve, diff --git a/internal/db/queries/rules.sql b/internal/db/queries/rules.sql index 7e28a1da..c6813e5f 100644 --- a/internal/db/queries/rules.sql +++ b/internal/db/queries/rules.sql @@ -15,7 +15,7 @@ INSERT INTO alert_rules ( ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING *; --- name: UpdateRule :one +-- name: ReplaceRule :one UPDATE alert_rules SET name = ?, description = ?, enabled = ?, service = ?, namespace = ?, expression = ?, for_seconds = ?, cooldown_s = ?, repeat_interval_s = ?, diff --git a/internal/db/sqlc.yaml b/internal/db/sqlc.yaml index 96b6ad97..3b433b3a 100644 --- a/internal/db/sqlc.yaml +++ b/internal/db/sqlc.yaml @@ -9,3 +9,6 @@ sql: out: "generated" emit_json_tags: true emit_empty_slices: true + rename: + webhook_url: WebhookURL + spec_json: SpecJSON diff --git a/internal/intelligence/annotations_test.go b/internal/intelligence/annotations_test.go index f62603be..e3208823 100644 --- a/internal/intelligence/annotations_test.go +++ b/internal/intelligence/annotations_test.go @@ -23,7 +23,7 @@ func TestDetectorPublishesWhilePersistenceBlocked(t *testing.T) { t.Fatal(err) } defer duck.Close() - unlock := duck.WriteGate().Lock(writegate.WriteRollupService) + unlock := duck.WriteGate().Lock(writegate.RollupService) defer unlock() d := NewDetector(duck, DefaultDetectorConfig()) ctx, cancel := context.WithCancel(t.Context()) @@ -63,7 +63,7 @@ func TestDetectorCoalescesPendingPersistence(t *testing.T) { }() defer func() { cancel(); <-done }() now := time.Now().UTC() - publish := func(at time.Time) { d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: at}) } + publish := func(at time.Time) { d.publishSnapshot(Snapshot{GeneratedAt: at}) } publish(now) select { case at := <-entered: @@ -115,7 +115,7 @@ func TestDetectorAnnotationMapping(t *testing.T) { } defer duck.Close() at := time.Now().UTC().Truncate(time.Minute) - snapshot := IntelligenceSnapshot{GeneratedAt: at, Anomalies: []Anomaly{{ServiceName: "checkout", Type: AnomalyLatencyDegradation, ZScore: -4, DetectedAt: at, WindowStart: at.Add(-5 * time.Minute), WindowEnd: at, Description: "Slow"}, {ServiceName: "payment", Type: AnomalyVolumeChange, ZScore: 3, DetectedAt: at, WindowStart: at.Add(-5 * time.Minute), WindowEnd: at, Description: "Low volume"}}} + snapshot := Snapshot{GeneratedAt: at, Anomalies: []Anomaly{{ServiceName: "checkout", Type: AnomalyLatencyDegradation, ZScore: -4, DetectedAt: at, WindowStart: at.Add(-5 * time.Minute), WindowEnd: at, Description: "Slow"}, {ServiceName: "payment", Type: AnomalyVolumeChange, ZScore: 3, DetectedAt: at, WindowStart: at.Add(-5 * time.Minute), WindowEnd: at, Description: "Low volume"}}} findings := detectorAnnotations(snapshot, "shop") if err := duck.RecordAnomalies(t.Context(), findings, at); err != nil { t.Fatal(err) @@ -192,17 +192,17 @@ func TestDetectorKeepsDisplacedPendingIncidentAndItsInterval(t *testing.T) { }() defer func() { cancel(); <-done }() now := time.Now().UTC() - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now}) + d.publishSnapshot(Snapshot{GeneratedAt: now}) select { case <-entered: case <-time.After(time.Second): t.Fatal("persistence did not start") } incident := Anomaly{ServiceName: "api", Type: AnomalyLatencyDegradation, ZScore: 4, WindowStart: now.Add(-time.Minute), WindowEnd: now} - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(time.Minute), Anomalies: []Anomaly{incident}}) + d.publishSnapshot(Snapshot{GeneratedAt: now.Add(time.Minute), Anomalies: []Anomaly{incident}}) // A repeated live pattern refresh must not duplicate this finding. - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(2 * time.Minute), Anomalies: []Anomaly{incident}}) - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(3 * time.Minute)}) + d.publishSnapshot(Snapshot{GeneratedAt: now.Add(2 * time.Minute), Anomalies: []Anomaly{incident}}) + d.publishSnapshot(Snapshot{GeneratedAt: now.Add(3 * time.Minute)}) close(release) select { case batch := <-entered: @@ -222,7 +222,7 @@ func TestDetectorBoundsPendingFindings(t *testing.T) { for i := range findings { findings[i] = Anomaly{Type: AnomalyLatencyDegradation, WindowStart: now.Add(time.Duration(i) * time.Minute), WindowEnd: now.Add(time.Duration(i+1) * time.Minute)} } - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now, Anomalies: findings}) + d.publishSnapshot(Snapshot{GeneratedAt: now, Anomalies: findings}) } batch := <-d.annotationBatches if len(batch.findings) != maxCarriedAnomalies || !batch.findings[len(batch.findings)-1].To.Equal(now.Add(time.Minute)) { diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go index c14f02bf..60a244be 100644 --- a/internal/intelligence/detector.go +++ b/internal/intelligence/detector.go @@ -28,7 +28,7 @@ type Detector struct { duck *query.Duck config DetectorConfig mu sync.RWMutex - snapshot *IntelligenceSnapshot + snapshot *Snapshot evaluationMu sync.Mutex evaluation *anomalyEvaluation annotationMu sync.Mutex @@ -111,7 +111,7 @@ func (d *Detector) runCheck(ctx context.Context) { } } -func (d *Detector) publishSnapshot(snapshot IntelligenceSnapshot) { +func (d *Detector) publishSnapshot(snapshot Snapshot) { d.mu.Lock() d.snapshot = &snapshot d.mu.Unlock() @@ -205,19 +205,19 @@ func (d *Detector) persistAnnotations(ctx context.Context, record func(context.C } // LatestSnapshot returns the most recent intelligence snapshot, or nil if none. -func (d *Detector) LatestSnapshot() *IntelligenceSnapshot { +func (d *Detector) LatestSnapshot() *Snapshot { d.mu.RLock() defer d.mu.RUnlock() return d.snapshot } // GenerateSnapshot generates a complete intelligence snapshot -func (d *Detector) GenerateSnapshot(ctx context.Context) (IntelligenceSnapshot, error) { +func (d *Detector) GenerateSnapshot(ctx context.Context) (Snapshot, error) { d.evaluationMu.Lock() defer d.evaluationMu.Unlock() completed, err := d.duck.CompletedServiceRollupMinute(ctx) if err != nil { - return IntelligenceSnapshot{}, fmt.Errorf("read detector service watermark: %w", err) + return Snapshot{}, fmt.Errorf("read detector service watermark: %w", err) } now := time.Now() lookback := now.Add(-d.config.LookbackWindow) @@ -227,7 +227,7 @@ func (d *Detector) GenerateSnapshot(ctx context.Context) (IntelligenceSnapshot, if d.evaluation == nil || !completed.Equal(d.evaluation.through) { anomalies, err := d.detectAnomalies(ctx, completed.Add(-d.config.LookbackWindow), completed) if err != nil { - return IntelligenceSnapshot{}, err + return Snapshot{}, err } d.evaluation = &anomalyEvaluation{through: completed, findings: anomalies} } @@ -239,9 +239,9 @@ func (d *Detector) GenerateSnapshot(ctx context.Context) (IntelligenceSnapshot, // A later watermark may replace this evaluation before patterns recover. // Keep its original intervals until a successful annotation publication. d.annotationMu.Lock() - d.unpublishedFindings = mergeAnnotationFindings(d.unpublishedFindings, detectorAnnotations(IntelligenceSnapshot{Anomalies: anomalies}, d.duck.DefaultNamespace())) + d.unpublishedFindings = mergeAnnotationFindings(d.unpublishedFindings, detectorAnnotations(Snapshot{Anomalies: anomalies}, d.duck.DefaultNamespace())) d.annotationMu.Unlock() - return IntelligenceSnapshot{}, err + return Snapshot{}, err } // Calculate health score @@ -250,7 +250,7 @@ func (d *Detector) GenerateSnapshot(ctx context.Context) (IntelligenceSnapshot, // Generate insights from anomalies insights := d.generateInsights(anomalies, patterns) - return IntelligenceSnapshot{ + return Snapshot{ evaluatedThrough: completed, GeneratedAt: now, Timeframe: fmt.Sprintf("last_%dm", int(d.config.LookbackWindow.Minutes())), @@ -812,7 +812,7 @@ func toInt(v interface{}) int { return int(toInt64(v)) } -func detectorAnnotations(snapshot IntelligenceSnapshot, namespace string) []annotations.Anomaly { +func detectorAnnotations(snapshot Snapshot, namespace string) []annotations.Anomaly { findings := make([]annotations.Anomaly, 0, len(snapshot.Anomalies)) for _, a := range snapshot.Anomalies { severity := "warn" diff --git a/internal/intelligence/detector_recovery_test.go b/internal/intelligence/detector_recovery_test.go index c7b25ed3..65e3925c 100644 --- a/internal/intelligence/detector_recovery_test.go +++ b/internal/intelligence/detector_recovery_test.go @@ -84,7 +84,7 @@ func TestDetectorRetriesFailedQueriesAtUnchangedWatermark(t *testing.T) { seedDetectorRecoveryRollup(t, db, end) d := NewDetector(&query.Duck{DB: db}, DefaultDetectorConfig()) if withPrevious { - d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: time.Now().Add(-time.Minute), evaluatedThrough: end.Add(-time.Minute), HealthScore: 85}) + d.publishSnapshot(Snapshot{GeneratedAt: time.Now().Add(-time.Minute), evaluatedThrough: end.Add(-time.Minute), HealthScore: 85}) <-d.annotationBatches } previous := d.LatestSnapshot() @@ -138,7 +138,7 @@ func TestDetectorRetriesPersistenceAtUnchangedWatermark(t *testing.T) { d := NewDetector(duck, DefaultDetectorConfig()) d.runCheck(t.Context()) previous := d.LatestSnapshot() - unlock := duck.WriteGate().Lock(writegate.WriteRollupService) + unlock := duck.WriteGate().Lock(writegate.RollupService) defer func() { if unlock != nil { unlock() @@ -280,7 +280,7 @@ func TestDetectorKeepsUnpublishedFindingsWhenWatermarkAdvances(t *testing.T) { if d.evaluation == nil || len(d.evaluation.findings) != 3 { t.Fatalf("missing successful anomaly evaluation: %+v", d.evaluation) } - want := detectorAnnotations(IntelligenceSnapshot{Anomalies: d.evaluation.findings}, d.duck.DefaultNamespace()) + want := detectorAnnotations(Snapshot{Anomalies: d.evaluation.findings}, d.duck.DefaultNamespace()) // Replace the failed-publication evaluation with a healthy, later window // while patterns still fail, then recover at that new watermark. diff --git a/internal/intelligence/effect_size_test.go b/internal/intelligence/effect_size_test.go index 5e094f75..25deb8a9 100644 --- a/internal/intelligence/effect_size_test.go +++ b/internal/intelligence/effect_size_test.go @@ -189,7 +189,7 @@ func assertEffectSizeFinding(t *testing.T, d *Detector, anomalies []Anomaly, ser if score := d.calculateHealthScore(found, nil); score != 100 { t.Fatalf("noise lowered health to %g", score) } - if len(detectorAnnotations(IntelligenceSnapshot{Anomalies: found}, "default")) != 0 || len(d.generateInsights(found, nil)) != 0 { + if len(detectorAnnotations(Snapshot{Anomalies: found}, "default")) != 0 || len(d.generateInsights(found, nil)) != 0 { t.Fatal("noise emitted annotations or insights") } return @@ -200,7 +200,7 @@ func assertEffectSizeFinding(t *testing.T, d *Detector, anomalies []Anomaly, ser if score := d.calculateHealthScore(found, nil); score >= 100 { t.Fatalf("qualified finding did not lower health: %g", score) } - annotations := detectorAnnotations(IntelligenceSnapshot{Anomalies: found}, "default") + annotations := detectorAnnotations(Snapshot{Anomalies: found}, "default") severity := "warn" if math.Abs(found[0].ZScore) >= 4 { severity = "bad" diff --git a/internal/intelligence/types.go b/internal/intelligence/types.go index 775725b5..839afbeb 100644 --- a/internal/intelligence/types.go +++ b/internal/intelligence/types.go @@ -52,8 +52,8 @@ type Insight struct { Message string `json:"message"` // Description } -// IntelligenceSnapshot represents the current system intelligence state -type IntelligenceSnapshot struct { +// Snapshot represents the current system intelligence state +type Snapshot struct { evaluatedThrough time.Time GeneratedAt time.Time `json:"generated_at"` Timeframe string `json:"timeframe"` // e.g., "last_15m" diff --git a/internal/intelligence/types_test.go b/internal/intelligence/types_test.go index 26e1af2d..500fcac1 100644 --- a/internal/intelligence/types_test.go +++ b/internal/intelligence/types_test.go @@ -144,9 +144,9 @@ func TestInsight(t *testing.T) { } } -func TestIntelligenceSnapshot(t *testing.T) { +func TestSnapshot(t *testing.T) { now := time.Now() - snap := IntelligenceSnapshot{ + snap := Snapshot{ GeneratedAt: now, Timeframe: "last_15m", Insights: []Insight{ diff --git a/internal/mcp/describe.go b/internal/mcp/describe.go index cc247005..c644115c 100644 --- a/internal/mcp/describe.go +++ b/internal/mcp/describe.go @@ -63,7 +63,7 @@ type ToolInput struct { type describeIntelligence struct{} -func (describeIntelligence) LatestSnapshot() *intelligence.IntelligenceSnapshot { return nil } +func (describeIntelligence) LatestSnapshot() *intelligence.Snapshot { return nil } // DescribeTools reports every tool this server exposes, as the server itself // reports them over MCP. diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 0abd4174..d0a5e28f 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -39,7 +39,7 @@ type Observability interface { } type IntelligenceSnapshots interface { - LatestSnapshot() *intelligence.IntelligenceSnapshot + LatestSnapshot() *intelligence.Snapshot } type QueryInput struct { @@ -448,10 +448,10 @@ func (s *Server) dependencies(ctx context.Context, _ *mcp.CallToolRequest, input return summary(output.Summary), output, nil } -func (s *Server) intelligenceSnapshot(_ context.Context, _ *mcp.CallToolRequest, _ struct{}) (*mcp.CallToolResult, intelligence.IntelligenceSnapshot, error) { +func (s *Server) intelligenceSnapshot(_ context.Context, _ *mcp.CallToolRequest, _ struct{}) (*mcp.CallToolResult, intelligence.Snapshot, error) { snapshot := s.intelligence.LatestSnapshot() if snapshot == nil { - return nil, intelligence.IntelligenceSnapshot{}, fmt.Errorf("intelligence snapshot is not ready") + return nil, intelligence.Snapshot{}, fmt.Errorf("intelligence snapshot is not ready") } return summary(snapshot.Summary), *snapshot, nil } diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go index 44802a8b..2ac57c7c 100644 --- a/internal/mcp/server_test.go +++ b/internal/mcp/server_test.go @@ -37,10 +37,10 @@ func withName(d panel.Dashboard, name string) panel.Dashboard { } type fakeIntelligence struct { - snapshot *intelligence.IntelligenceSnapshot + snapshot *intelligence.Snapshot } -func (f fakeIntelligence) LatestSnapshot() *intelligence.IntelligenceSnapshot { +func (f fakeIntelligence) LatestSnapshot() *intelligence.Snapshot { return f.snapshot } @@ -188,7 +188,7 @@ func TestDependencyToolForwardsScopeAndBounds(t *testing.T) { } func TestIntelligenceSnapshotReturnsStructuredOutput(t *testing.T) { - want := &intelligence.IntelligenceSnapshot{ + want := &intelligence.Snapshot{ GeneratedAt: time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC), Timeframe: "last_15m", Summary: "One anomaly requires attention.", diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index cedc1671..a7623ec4 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -461,7 +461,7 @@ func RecordParquetRead(reader string, waitSec float64, admitted bool) { } // RecordWriteGate records one complete Telemetry catalog write critical section. -// Callers constrain operation to the fixed writegate.WriteOperation set so this +// Callers constrain operation to the fixed writegate.Operation set so this // metric cannot grow with tenant or telemetry cardinality. func RecordWriteGate(operation string, waitSec, holdSec float64) { WriteGateWait.WithLabelValues(operation).Observe(waitSec) diff --git a/internal/query/analytical_backlog_test.go b/internal/query/analytical_backlog_test.go index 1dc53684..24856c72 100644 --- a/internal/query/analytical_backlog_test.go +++ b/internal/query/analytical_backlog_test.go @@ -97,7 +97,7 @@ FROM rollup_state`, edgeRollupStateKey, edgeRollupSubCursorKey).Scan(&watermark, // The completed-batch writer still works with an analytical transaction // occupying its gate and pool slot. - unlock := d.writeGate.Lock(writegate.WriteRollupEdge) + unlock := d.writeGate.Lock(writegate.RollupEdge) tx, err := d.writer().BeginTx(ctx, nil) if err != nil { unlock() diff --git a/internal/query/annotation_maintenance_test.go b/internal/query/annotation_maintenance_test.go index 132c9983..d08d99d5 100644 --- a/internal/query/annotation_maintenance_test.go +++ b/internal/query/annotation_maintenance_test.go @@ -116,7 +116,7 @@ func TestVersionFailureBackoffCapsAtThirtyMinutes(t *testing.T) { func TestVersionPassTimeoutWarnAndMetric(t *testing.T) { d, _ := versionEngine(t) - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) defer unlock() counter := metrics.RollupComponentTotal.WithLabelValues("version", "error") before := testutil.ToFloat64(counter) @@ -265,7 +265,7 @@ func TestSlowVersionPassCommits(t *testing.T) { if err := repo.Commit(t.Context(), telemetrystore.Batch{ID: "slow", Logs: []telemetry.Log{{ServiceName: "svc", Resource: map[string]any{"service.version": "v1"}, EventUnixNanos: now, IngestedAt: now}}}); err != nil { t.Fatal(err) } - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) timer := time.AfterFunc(2200*time.Millisecond, unlock) defer func() { if timer.Stop() { diff --git a/internal/query/annotations.go b/internal/query/annotations.go index 8885b446..736c5473 100644 --- a/internal/query/annotations.go +++ b/internal/query/annotations.go @@ -139,7 +139,7 @@ func (d *Duck) refreshVersionRollup(ctx context.Context, waitBudget time.Duratio }() waitCtx, cancelWait := context.WithTimeout(ctx, waitBudget) defer cancelWait() - unlock, err := d.writeGate.LockContext(waitCtx, writegate.WriteRollupVersion) + unlock, err := d.writeGate.LockContext(waitCtx, writegate.RollupVersion) if err != nil { return 0, fmt.Errorf("version_rollup/write_gate: %w", err) } @@ -419,7 +419,7 @@ func mergeAnomalies(existing, incoming []annotations.Anomaly) []annotations.Anom func (d *Duck) RecordAnomalies(ctx context.Context, findings []annotations.Anomaly, now time.Time) error { ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() - unlock, err := d.writeGate.LockContext(ctx, writegate.WriteAnomalyLog) + unlock, err := d.writeGate.LockContext(ctx, writegate.AnomalyLog) if err != nil { return fmt.Errorf("anomaly_log/write_gate: %w", err) } @@ -519,7 +519,7 @@ func drainVersionRollup(ctx context.Context, refresh func(context.Context) (int6 } func (d *Duck) transferVersionMarker(ctx context.Context, output telemetry.BatchMetadata, inputs []string) error { - unlock, err := d.writeGate.LockContext(ctx, writegate.WriteRollupVersion) + unlock, err := d.writeGate.LockContext(ctx, writegate.RollupVersion) if err != nil { return err } diff --git a/internal/query/backlog_pressure_test.go b/internal/query/backlog_pressure_test.go index 9b3deadf..f2ba99a9 100644 --- a/internal/query/backlog_pressure_test.go +++ b/internal/query/backlog_pressure_test.go @@ -405,7 +405,7 @@ func TestBacklogAnnotationTimeoutsIdentifyWriteGateAndPreserveDeadline(t *testin for _, label := range []string{"version_rollup/write_gate", "anomaly_log/write_gate"} { t.Run(label, func(t *testing.T) { d := &Duck{repository: &telemetrystore.Repository{}} - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) defer unlock() ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) defer cancel() diff --git a/internal/query/batch_cache.go b/internal/query/batch_cache.go index c74341ff..9578f2fd 100644 --- a/internal/query/batch_cache.go +++ b/internal/query/batch_cache.go @@ -134,7 +134,7 @@ func (d *Duck) RefreshReadCaches(ctx context.Context) (rows int64, err error) { if d.repository == nil { return 0, nil } - unlock, err := d.cacheGate.LockContext(ctx, writegate.WriteReadCache) + unlock, err := d.cacheGate.LockContext(ctx, writegate.ReadCache) if err != nil { return 0, err } @@ -414,7 +414,7 @@ func (d *Duck) PublishParquetReplacement(ctx context.Context, output telemetry.B if len(inputs) == 0 { return d.PublishParquet(ctx, publish) } - unlock, err := d.cacheGate.LockContext(ctx, writegate.WriteReadCache) + unlock, err := d.cacheGate.LockContext(ctx, writegate.ReadCache) if err != nil { return err } diff --git a/internal/query/duck.go b/internal/query/duck.go index efe60f49..18162a50 100644 --- a/internal/query/duck.go +++ b/internal/query/duck.go @@ -59,8 +59,8 @@ type Duck struct { // Cached under writeGate so an idle pass executes no SQL at all. rollupWatermarks map[string]int64 // writeGate serializes writes to the rebuildable DuckDB rollup cache. - writeGate writegate.WriteGate - cacheGate writegate.WriteGate + writeGate writegate.Gate + cacheGate writegate.Gate // Protected by writeGate; controls periodic version-history retirement. versionRollupPasses uint64 versionRollupFailures map[string]versionRollupFailure @@ -127,7 +127,7 @@ var parquetStatsWait = 5 * time.Second // WriteGate returns the gate that serializes writes to DuckDB's rebuildable // rollup cache. -func (d *Duck) WriteGate() *writegate.WriteGate { return &d.writeGate } +func (d *Duck) WriteGate() *writegate.Gate { return &d.writeGate } // duckDBPoolSize is the effective connection-pool size: the configured value, // floored at 1. @@ -306,7 +306,7 @@ func NewDuck(ctx context.Context, cfg config.Config, repository *telemetrystore. // historical dataset. Runs once at boot before RunRollups, so taking the write // gate here is uncontended. func (d *Duck) skipRollupToLatest(ctx context.Context) error { - unlock := d.writeGate.Lock(writegate.WriteRollupSkip) + unlock := d.writeGate.Lock(writegate.RollupSkip) defer unlock() tx, err := d.writer().BeginTx(ctx, nil) @@ -751,12 +751,12 @@ func (d *Duck) runRepositoryMaintenance(ctx context.Context) error { pruneErr = errors.Join(pruneErr, recoveryErr, cleanupErr, parquetErr, compactErr) } cacheErr := func() error { - unlock, err := d.writeGate.LockContext(ctx, writegate.WriteMaintenance) + unlock, err := d.writeGate.LockContext(ctx, writegate.Maintenance) if err != nil { return err } defer unlock() - unlockCache, err := d.cacheGate.LockContext(ctx, writegate.WriteMaintenance) + unlockCache, err := d.cacheGate.LockContext(ctx, writegate.Maintenance) if err != nil { return err } @@ -812,7 +812,7 @@ func (d *Duck) refreshServiceRollup(ctx context.Context) (int64, error) { // Serialize against other DuckDB writers (edge rollup and maintenance). // The write gate is always acquired before a connection to keep lock ordering // consistent and deadlock-free. - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) defer unlock() if err := ctx.Err(); err != nil { return 0, err @@ -925,7 +925,7 @@ func (d *Duck) refreshEdgeRollupPass(ctx context.Context) (int64, bool, error) { updateRollupProgress(metrics.RollupEdge, true, watermark, sourceMax) } }() - unlock := d.writeGate.Lock(writegate.WriteRollupEdge) + unlock := d.writeGate.Lock(writegate.RollupEdge) defer unlock() if err := ctx.Err(); err != nil { return 0, false, err diff --git a/internal/query/duck_test.go b/internal/query/duck_test.go index 3a0319d5..790e560c 100644 --- a/internal/query/duck_test.go +++ b/internal/query/duck_test.go @@ -390,7 +390,7 @@ func TestPublishParquetHonorsContext(t *testing.T) { func TestParquetPublicationDoesNotWaitForDuckDBWriteGate(t *testing.T) { d := &Duck{} - release := d.writeGate.Lock(writegate.WriteRollupService) + release := d.writeGate.Lock(writegate.RollupService) defer release() ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) defer cancel() @@ -559,7 +559,7 @@ func TestRepositoryPublicationDoesNotWaitForDuckDBWrites(t *testing.T) { mock.ExpectExec("CHECKPOINT").WillReturnResult(sqlmock.NewResult(0, 0)) d := &Duck{DB: db, writeDB: db, repository: repository, cfg: config.Config{MaintenanceInterval: time.Nanosecond, RetentionDays: 1}} mustRLock(t, &d.parquetMu) - release := d.writeGate.Lock(writegate.WriteRollupService) + release := d.writeGate.Lock(writegate.RollupService) done := make(chan error, 1) go func() { done <- d.runRepositoryMaintenance(context.Background()) }() deadline := time.Now().Add(time.Second) diff --git a/internal/query/file_snapshot_test.go b/internal/query/file_snapshot_test.go index 43b1a8f3..8162d892 100644 --- a/internal/query/file_snapshot_test.go +++ b/internal/query/file_snapshot_test.go @@ -265,7 +265,7 @@ func TestFailedBatchCacheTransactionPublishesNoContributions(t *testing.T) { func TestReadCacheWorkerAcknowledgesNewFilesWithoutRollupTick(t *testing.T) { d, repo := newBatchCacheTest(t) - release := d.writeGate.Lock(writegate.WriteRollupService) + release := d.writeGate.Lock(writegate.RollupService) defer release() ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) @@ -465,7 +465,7 @@ func TestPreparedReplacementDoesNotHoldCacheWriteGate(t *testing.T) { waitForQueuedWriter(t, &d.parquetMu) passCtx, stop := context.WithTimeout(t.Context(), time.Second) defer stop() - unlock, err := d.cacheGate.LockContext(passCtx, writegate.WriteReadCache) + unlock, err := d.cacheGate.LockContext(passCtx, writegate.ReadCache) if err != nil { t.Fatalf("preparation held cache gate behind publication: %v", err) } @@ -576,7 +576,7 @@ func TestCompletedBatchWriterProgressesDuringAnalyticalTransaction(t *testing.T) if err := repo.Commit(t.Context(), telemetrystore.Batch{ID: "independent", Spans: []telemetry.Span{{TraceID: "trace", StartUnixNanos: 1, EndUnixNanos: 2, IngestedAt: 1}}}); err != nil { t.Fatal(err) } - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) defer unlock() tx, err := d.writer().BeginTx(t.Context(), nil) if err != nil { diff --git a/internal/query/version_gate_test.go b/internal/query/version_gate_test.go index ece97cec..e9d21e93 100644 --- a/internal/query/version_gate_test.go +++ b/internal/query/version_gate_test.go @@ -18,7 +18,7 @@ import ( func TestVersionRollupBudgetStartsAfterGate(t *testing.T) { d, repo := versionEngine(t) commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) timer := time.AfterFunc(10100*time.Millisecond, unlock) defer func() { if timer.Stop() { @@ -35,7 +35,7 @@ func TestVersionRollupBudgetStartsAfterGate(t *testing.T) { func TestGateTimeoutDoesNotBackOff(t *testing.T) { d, repo := versionEngine(t) commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) defer unlock() ctx, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) defer cancel() @@ -50,7 +50,7 @@ func TestGateTimeoutDoesNotBackOff(t *testing.T) { func TestVersionRollupWaitIsBoundedWithoutCallerDeadline(t *testing.T) { d, repo := versionEngine(t) commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) var releaseOnce sync.Once release := func() { releaseOnce.Do(unlock) } defer release() @@ -86,7 +86,7 @@ func TestVersionRollupWaitIsBoundedWithoutCallerDeadline(t *testing.T) { func TestVersionRollupWaitHonorsCallerCancellation(t *testing.T) { d, repo := versionEngine(t) commitVersionLogs(t, repo, "healthy", time.Now().UTC().Add(-time.Hour), 1) - unlock := d.writeGate.Lock(writegate.WriteRollupService) + unlock := d.writeGate.Lock(writegate.RollupService) var once sync.Once release := func() { once.Do(unlock) } defer release() @@ -126,7 +126,7 @@ func TestVersionRollupWaitHonorsCallerCancellation(t *testing.T) { // the write gate. // This test-only probe avoids adding instrumentation to the production gates. func signalVersionGateWait(blocked chan<- struct{}, stop <-chan struct{}) { - frame := "writegate.(*WriteGate).LockContext" + frame := "writegate.(*Gate).LockContext" ticker := time.NewTicker(time.Millisecond) defer ticker.Stop() buffer := make([]byte, 1<<20) diff --git a/internal/query/writegate/write_gate.go b/internal/query/writegate/write_gate.go index 8922c5f7..7f2ab5e9 100644 --- a/internal/query/writegate/write_gate.go +++ b/internal/query/writegate/write_gate.go @@ -10,29 +10,29 @@ import ( "github.com/labstack/fanout/internal/metrics" ) -// WriteOperation is a bounded metric label for a rollup-cache write. +// Operation is a bounded metric label for a rollup-cache write. // Keep this list exhaustive: arbitrary strings would create an unbounded // Prometheus label surface. -type WriteOperation string +type Operation string const ( - WriteRollupSkip WriteOperation = "rollup_skip_to_latest" - WriteRollupService WriteOperation = "rollup_service" - WriteReadCache WriteOperation = "read_cache" - WriteRollupEdge WriteOperation = "rollup_edge" - WriteRollupVersion WriteOperation = "rollup_version" - WriteAnomalyLog WriteOperation = "anomaly_log" - WriteMaintenance WriteOperation = "maintenance" + RollupSkip Operation = "rollup_skip_to_latest" + RollupService Operation = "rollup_service" + ReadCache Operation = "read_cache" + RollupEdge Operation = "rollup_edge" + RollupVersion Operation = "rollup_version" + AnomalyLog Operation = "anomaly_log" + Maintenance Operation = "maintenance" ) -// WriteGate serializes DuckDB writes while measuring how long each operation +// Gate serializes DuckDB writes while measuring how long each operation // waits for and holds the critical section. The zero value is ready for use. -type WriteGate struct { +type Gate struct { once sync.Once token chan struct{} } -func (g *WriteGate) init() { +func (g *Gate) init() { g.once.Do(func() { g.token = make(chan struct{}, 1) g.token <- struct{}{} @@ -48,7 +48,7 @@ var observe = metrics.RecordWriteGate // Lock acquires the cache write gate and returns its release function. // Callers must defer the returned function before acquiring a database // connection, transaction, or appender, and must call it exactly once. -func (g *WriteGate) Lock(operation WriteOperation) func() { +func (g *Gate) Lock(operation Operation) func() { unlock, err := g.LockContext(context.Background(), operation) if err != nil { panic(err) @@ -57,7 +57,7 @@ func (g *WriteGate) Lock(operation WriteOperation) func() { } // LockContext acquires the cache write gate, or returns when ctx expires. -func (g *WriteGate) LockContext(ctx context.Context, operation WriteOperation) (func(), error) { +func (g *Gate) LockContext(ctx context.Context, operation Operation) (func(), error) { g.init() waitStarted := time.Now() select { diff --git a/internal/query/writegate/write_gate_test.go b/internal/query/writegate/write_gate_test.go index e0929fef..46020aa7 100644 --- a/internal/query/writegate/write_gate_test.go +++ b/internal/query/writegate/write_gate_test.go @@ -13,22 +13,22 @@ import ( dto "github.com/prometheus/client_model/go" ) -func TestWriteGateLockContextHonorsCancellation(t *testing.T) { - var gate WriteGate - unlock := gate.Lock(WriteMaintenance) +func TestGateLockContextHonorsCancellation(t *testing.T) { + var gate Gate + unlock := gate.Lock(Maintenance) ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) defer cancel() - if _, err := gate.LockContext(ctx, WriteMaintenance); !errors.Is(err, context.DeadlineExceeded) { + if _, err := gate.LockContext(ctx, Maintenance); !errors.Is(err, context.DeadlineExceeded) { t.Fatalf("LockContext error = %v, want deadline exceeded", err) } unlock() - gate.Lock(WriteMaintenance)() + gate.Lock(Maintenance)() } -func TestWriteGateSerializesHoldersInAcquisitionOrder(t *testing.T) { +func TestGateSerializesHoldersInAcquisitionOrder(t *testing.T) { t.Parallel() - var gate WriteGate + var gate Gate firstEntered := make(chan struct{}) releaseFirst := make(chan struct{}) secondStarted := make(chan struct{}) @@ -37,7 +37,7 @@ func TestWriteGateSerializesHoldersInAcquisitionOrder(t *testing.T) { go func() { defer func() { done <- struct{}{} }() - unlock := gate.Lock(WriteMaintenance) + unlock := gate.Lock(Maintenance) close(firstEntered) <-releaseFirst unlock() @@ -47,7 +47,7 @@ func TestWriteGateSerializesHoldersInAcquisitionOrder(t *testing.T) { go func() { defer func() { done <- struct{}{} }() close(secondStarted) - defer gate.Lock(WriteMaintenance)() + defer gate.Lock(Maintenance)() close(secondEntered) }() <-secondStarted @@ -70,23 +70,23 @@ func TestWriteGateSerializesHoldersInAcquisitionOrder(t *testing.T) { // A panicking holder must not strand the gate. Recovering and re-acquiring in // the same test proves the mutex is free, which a leaked lock would deadlock on. -func TestWriteGateReleasesAfterPanic(t *testing.T) { +func TestGateReleasesAfterPanic(t *testing.T) { t.Parallel() - var gate WriteGate + var gate Gate func() { defer func() { if recover() == nil { t.Error("expected panic to propagate") } }() - defer gate.Lock(WriteMaintenance)() + defer gate.Lock(Maintenance)() panic("boom") }() acquired := make(chan struct{}) go func() { - defer gate.Lock(WriteMaintenance)() + defer gate.Lock(Maintenance)() close(acquired) }() select { @@ -101,8 +101,8 @@ func TestWriteGateReleasesAfterPanic(t *testing.T) { // catalog write behind the metrics registry. Asserting the gate is acquirable // from inside the observer is what makes the ordering detectable — timing the // release cannot, because a Prometheus write is only microseconds. -func TestWriteGateObservesOutsideTheCriticalSection(t *testing.T) { - var gate WriteGate +func TestGateObservesOutsideTheCriticalSection(t *testing.T) { + var gate Gate var freeDuringObserve bool restore := swapObserver(t, func(string, float64, float64) { @@ -115,7 +115,7 @@ func TestWriteGateObservesOutsideTheCriticalSection(t *testing.T) { }) defer restore() - gate.Lock(WriteMaintenance)() + gate.Lock(Maintenance)() if !freeDuringObserve { t.Fatal("gate was still held while the observation ran — move the Unlock above observe()") @@ -125,17 +125,17 @@ func TestWriteGateObservesOutsideTheCriticalSection(t *testing.T) { // Both halves must reach Prometheus: wait time without hold time cannot answer // whether ingest is stalling behind rollups, which is the question this // instrumentation exists to answer. -func TestWriteGateRecordsBothWaitAndHoldHistograms(t *testing.T) { - beforeWait := histogramCount(t, "fanout_write_gate_wait_seconds", WriteMaintenance) - beforeHold := histogramCount(t, "fanout_write_gate_hold_seconds", WriteMaintenance) +func TestGateRecordsBothWaitAndHoldHistograms(t *testing.T) { + beforeWait := histogramCount(t, "fanout_write_gate_wait_seconds", Maintenance) + beforeHold := histogramCount(t, "fanout_write_gate_hold_seconds", Maintenance) - var gate WriteGate - gate.Lock(WriteMaintenance)() + var gate Gate + gate.Lock(Maintenance)() - if after := histogramCount(t, "fanout_write_gate_wait_seconds", WriteMaintenance); after != beforeWait+1 { + if after := histogramCount(t, "fanout_write_gate_wait_seconds", Maintenance); after != beforeWait+1 { t.Errorf("wait sample count = %d, want %d", after, beforeWait+1) } - if after := histogramCount(t, "fanout_write_gate_hold_seconds", WriteMaintenance); after != beforeHold+1 { + if after := histogramCount(t, "fanout_write_gate_hold_seconds", Maintenance); after != beforeHold+1 { t.Errorf("hold sample count = %d, want %d", after, beforeHold+1) } } @@ -151,7 +151,7 @@ func swapObserver(t *testing.T, fn func(string, float64, float64)) func() { // histogramCount reports how many samples one write-gate histogram holds for an // operation. Deltas rather than Reset() keep this immune to test order. -func histogramCount(t *testing.T, name string, operation WriteOperation) uint64 { +func histogramCount(t *testing.T, name string, operation Operation) uint64 { t.Helper() families, err := prometheus.DefaultGatherer.Gather()