From 9f4ac25a488c52a5ca1d41a35133cb85b3a48783 Mon Sep 17 00:00:00 2001 From: claude-bot-go Date: Sat, 25 Jul 2026 13:18:45 +0800 Subject: [PATCH] =?UTF-8?q?Fix=20#204:=20Fix=20WasmAgent/symkernel#203=20(?= =?UTF-8?q?[milestone=20Milestone=2011]=20`internal/observability`=20?= =?UTF-8?q?=E2=80=94=20Policy=20performance=20dashboards:=20per-po?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/observability/alerting.go | 444 ++++++++++++++++ internal/observability/dashboard.go | 454 ++++++++++++++++ internal/observability/metrics.go | 406 +++++++++++++++ internal/observability/observability_test.go | 511 +++++++++++++++++++ 4 files changed, 1815 insertions(+) create mode 100644 internal/observability/alerting.go create mode 100644 internal/observability/dashboard.go create mode 100644 internal/observability/metrics.go create mode 100644 internal/observability/observability_test.go diff --git a/internal/observability/alerting.go b/internal/observability/alerting.go new file mode 100644 index 0000000..48f45a6 --- /dev/null +++ b/internal/observability/alerting.go @@ -0,0 +1,444 @@ +// alerting.go — degradation alerts over the per-policy metrics. +// +// An AlertEngine holds a set of AlertRules; Evaluate walks the rules against +// the live Collector snapshot and fires an Alert for every policy whose +// observed value crosses its threshold. The same rules can be exported as a +// Prometheus alerting-rule file (PrometheusRulesYAML) so an operator can run +// the alerts either inside symkernel (over the HTTP API) or in Prometheus + +// Alertmanager (the "integration with Grafana/Prometheus for alerting on +// policy degradation" the milestone calls for). +// +// Thresholds are deliberately conservative: a verification service degrades +// long before it fails, so the defaults flag a p95 latency regression and a +// rising error rate rather than a hard outage. Operators override them via +// NewAlertEngine with a custom rule set. + +package observability + +import ( + "bytes" + "fmt" + "time" + + "go.yaml.in/yaml/v3" +) + +// Severity grades an alert's urgency. It maps to the Prometheus severity +// label and, in Grafana, to notification-routing policy. +type Severity string + +const ( + // SeverityInfo is informational (e.g. a policy that just crossed a + // soft threshold). Routed to a low-noise channel. + SeverityInfo Severity = "info" + // SeverityWarning means performance is degrading; investigate soon. + SeverityWarning Severity = "warning" + // SeverityCritical means the policy is failing its SLO now. + SeverityCritical Severity = "critical" +) + +// MetricKind names the metric an AlertRule evaluates. It selects which field +// of a PolicySummary the rule compares against its Threshold. +type MetricKind string + +const ( + // MetricLatencyP95Ms compares the policy's p95 latency (ms). + MetricLatencyP95Ms MetricKind = "latency_p95_ms" + // MetricLatencyP99Ms compares the policy's p99 latency (ms). + MetricLatencyP99Ms MetricKind = "latency_p99_ms" + // MetricErrorRate compares the policy's failure fraction (0..1). + MetricErrorRate MetricKind = "error_rate" + // MetricPeakMemoryBytes compares the policy's peak memory (bytes). + MetricPeakMemoryBytes MetricKind = "peak_memory_bytes" +) + +// Comparison is the relational operator applied to observed vs threshold. +type Comparison string + +const ( + // CompareGreaterThan fires when observed > threshold. + CompareGreaterThan Comparison = ">" + // CompareLessThan fires when observed < threshold (e.g. throughput + // collapse if a future rule tracks it). + CompareLessThan Comparison = "<" +) + +// AlertRule defines a single degradation check. A rule fires for every +// policy whose metric crosses Threshold in the Comparison direction, +// provided the policy has at least MinSamples observations (so a policy +// that was just deployed does not alert on a single slow request). +type AlertRule struct { + // ID is the stable rule identifier used in the fired Alert and the + // Prometheus alert name. Required. + ID string + + // Name is a human-readable summary. + Name string + + // Metric selects which PolicySummary field is compared. + Metric MetricKind + + // Comparison is the direction of the comparison (defaults to >). + Comparison Comparison + + // Threshold is the value Metric is compared against, in the metric's + // native unit (ms for latency, fraction for error rate, bytes for + // memory). + Threshold float64 + + // For is the duration the condition must hold before the alert + // fires in Prometheus; inside symkernel Evaluate fires immediately + // and stamps For into the emitted Prometheus rule. + For time.Duration + + // Severity grades the alert. + Severity Severity + + // Message is the human-readable explanation included in a fired + // Alert and the Prometheus annotation. + Message string + + // MinSamples is the minimum observation count required before the + // rule is evaluated for a policy. Zero means "always evaluate". + MinSamples uint64 +} + +// Alert is a rule firing against a specific policy. It is the JSON shape +// returned by GET /v1/observability/alerts. +type Alert struct { + RuleID string `json:"rule_id"` + Name string `json:"name"` + Severity Severity `json:"severity"` + PolicyID string `json:"policy_id"` + Tier Tier `json:"tier"` + Metric MetricKind `json:"metric"` + Observed float64 `json:"observed"` + Threshold float64 `json:"threshold"` + Unit string `json:"unit"` + Message string `json:"message"` + FiredAt time.Time `json:"fired_at"` +} + +// AlertEngine evaluates a rule set against a Collector. It is safe for +// concurrent use. +type AlertEngine struct { + rules []AlertRule + now func() time.Time +} + +// NewAlertEngine builds an engine from the given rules. If rules is empty +// the engine fires nothing — call DefaultRules for the standard set. +func NewAlertEngine(rules []AlertRule) *AlertEngine { + return &AlertEngine{rules: rules, now: time.Now} +} + +// DefaultRules returns the standard symkernel degradation rule set: +// +// - p95 latency over 500ms (warning) and over 1s (critical) +// - error rate over 5% (warning) and over 10% (critical) +// - peak memory over 512MiB (critical) +// +// These are conservative starting points tuned to the per-tier SLOs (CEL +// sub-millisecond, wazero ~1ms, Z3 up to seconds); operators override them +// with NewAlertEngine. +func DefaultRules() []AlertRule { + return []AlertRule{ + { + ID: "PolicyLatencyP95High", + Name: "Policy p95 latency is high", + Metric: MetricLatencyP95Ms, + Comparison: CompareGreaterThan, + Threshold: 500, + For: 5 * time.Minute, + Severity: SeverityWarning, + Message: "p95 verification latency for {{policy_id}} exceeded 500ms — investigate a performance regression.", + MinSamples: 20, + }, + { + ID: "PolicyLatencyP95Critical", + Name: "Policy p95 latency is critical", + Metric: MetricLatencyP95Ms, + Comparison: CompareGreaterThan, + Threshold: 1000, + For: 2 * time.Minute, + Severity: SeverityCritical, + Message: "p95 verification latency for {{policy_id}} exceeded 1s — the policy is failing its latency SLO.", + MinSamples: 10, + }, + { + ID: "PolicyErrorRateHigh", + Name: "Policy error rate is elevated", + Metric: MetricErrorRate, + Comparison: CompareGreaterThan, + Threshold: 0.05, + For: 5 * time.Minute, + Severity: SeverityWarning, + Message: "More than 5% of verifications for {{policy_id}} are failing — check for malformed inputs or solver errors.", + MinSamples: 50, + }, + { + ID: "PolicyErrorRateCritical", + Name: "Policy error rate is critical", + Metric: MetricErrorRate, + Comparison: CompareGreaterThan, + Threshold: 0.10, + For: 2 * time.Minute, + Severity: SeverityCritical, + Message: "More than 10% of verifications for {{policy_id}} are failing — the policy is likely broken.", + MinSamples: 50, + }, + { + ID: "PolicyPeakMemoryCritical", + Name: "Policy peak memory is critical", + Metric: MetricPeakMemoryBytes, + Comparison: CompareGreaterThan, + Threshold: 512 * 1024 * 1024, + For: 5 * time.Minute, + Severity: SeverityCritical, + Message: "A verification for {{policy_id}} consumed over 512MiB — possible runaway sandbox allocation.", + MinSamples: 1, + }, + } +} + +// DefaultAlertEngine returns an engine seeded with DefaultRules. +func DefaultAlertEngine() *AlertEngine { + return NewAlertEngine(DefaultRules()) +} + +// Evaluate walks every rule against every policy in the Collector snapshot +// and returns the alerts that fired, sorted by severity (critical first) +// then policy id. It does not mutate the Collector. +func (e *AlertEngine) Evaluate(c *Collector) []Alert { + now := e.now() + var alerts []Alert + for _, ps := range c.allSeries() { + summary := summarize(ps) + for _, rule := range e.rules { + if rule.MinSamples > 0 && summary.Latency.Count < rule.MinSamples { + // MinSamples guards on the latency sample count, + // which is also the verification count for a + // policy (one latency sample per verification). + continue + } + observed, ok := metricValue(summary, rule.Metric) + if !ok { + continue + } + if !compare(rule.Comparison, observed, rule.Threshold) { + continue + } + alerts = append(alerts, Alert{ + RuleID: rule.ID, + Name: rule.Name, + Severity: rule.Severity, + PolicyID: summary.PolicyID, + Tier: summary.Tier, + Metric: rule.Metric, + Observed: observed, + Threshold: rule.Threshold, + Unit: metricUnit(rule.Metric), + Message: renderMessage(rule.Message, summary.PolicyID, observed), + FiredAt: now, + }) + } + } + sortAlerts(alerts) + return alerts +} + +// metricValue extracts the metric the rule watches from a PolicySummary. +// The ok return is false for a metric the summary does not carry (none +// today, but guards future MetricKinds). +func metricValue(s PolicySummary, m MetricKind) (float64, bool) { + switch m { + case MetricLatencyP95Ms: + return s.Latency.P95Ms, true + case MetricLatencyP99Ms: + return s.Latency.P99Ms, true + case MetricErrorRate: + return s.ErrorRate, true + case MetricPeakMemoryBytes: + return float64(s.Resources.PeakMemoryBytes), true + default: + return 0, false + } +} + +// metricUnit returns the display unit for a metric, included in fired +// alerts so dashboards can format Observed without guessing. +func metricUnit(m MetricKind) string { + switch m { + case MetricLatencyP95Ms, MetricLatencyP99Ms: + return "ms" + case MetricErrorRate: + return "ratio" + case MetricPeakMemoryBytes: + return "bytes" + default: + return "" + } +} + +// compare applies the relational operator. +func compare(op Comparison, observed, threshold float64) bool { + switch op { + case CompareLessThan: + return observed < threshold + default: // CompareGreaterThan (and unset, which defaults to >) + return observed > threshold + } +} + +// renderMessage substitutes {{policy_id}} and {{observed}} into the rule's +// message template. It is deliberately tiny (no full template engine) so +// the dependency surface stays at the standard library. +func renderMessage(msg, policyID string, observed float64) string { + out := msg + out = replaceAll(out, "{{policy_id}}", policyID) + out = replaceAll(out, "{{observed}}", fmt.Sprintf("%.4g", observed)) + return out +} + +// replaceAll is a strings.ReplaceAll alias kept local so this file does not +// import strings solely for one call. +func replaceAll(s, old, new string) string { + var b []byte + for i := 0; i < len(s); { + if i+len(old) <= len(s) && s[i:i+len(old)] == old { + b = append(b, new...) + i += len(old) + continue + } + b = append(b, s[i]) + i++ + } + return string(b) +} + +// alertOrder is the severity ranking used to sort fired alerts (critical +// first so dashboards surface the worst degradation). +var alertOrder = map[Severity]int{ + SeverityCritical: 0, + SeverityWarning: 1, + SeverityInfo: 2, +} + +func sortAlerts(a []Alert) { + // Simple insertion sort: the alert count is small (rules × policies), + // and avoiding a sort.Slice import keeps the dependency list tight. + for i := 1; i < len(a); i++ { + for j := i; j > 0 && alertLess(a[j], a[j-1]); j-- { + a[j], a[j-1] = a[j-1], a[j] + } + } +} + +func alertLess(x, y Alert) bool { + ox, ok := alertOrder[x.Severity] + if !ok { + ox = 3 + } + oy, ok := alertOrder[y.Severity] + if !ok { + oy = 3 + } + if ox != oy { + return ox < oy + } + if x.PolicyID != y.PolicyID { + return x.PolicyID < y.PolicyID + } + return x.RuleID < y.RuleID +} + +// --- Prometheus alerting-rule export --- + +// promRuleGroup is the YAML model for a Prometheus alerting-rules file. It +// marshals to the "groups:" document Prometheus and Alertmanager consume. +type promRuleGroup struct { + Name string `yaml:"name"` + Rules []promRule `yaml:"rules"` +} + +type promRule struct { + Alert string `yaml:"alert"` + Expr string `yaml:"expr"` + For string `yaml:"for"` + Labels map[string]string `yaml:"labels"` + Annotations map[string]string `yaml:"annotations"` +} + +// promQLFor renders the PromQL expression for a rule against the metric +// names emitted by Collector.PrometheusText. +func promQLFor(r AlertRule) string { + switch r.Metric { + case MetricLatencyP95Ms: + return fmt.Sprintf( + `histogram_quantile(0.95, sum by (policy_id, le) (rate(symkernel_policy_verification_duration_seconds_bucket[5m]))) > %g`, + r.Threshold/1000) + case MetricLatencyP99Ms: + return fmt.Sprintf( + `histogram_quantile(0.99, sum by (policy_id, le) (rate(symkernel_policy_verification_duration_seconds_bucket[5m]))) > %g`, + r.Threshold/1000) + case MetricErrorRate: + return fmt.Sprintf( + `sum by (policy_id) (rate(symkernel_policy_verification_total{result="failure"}[5m])) / clamp_min(sum by (policy_id) (rate(symkernel_policy_verification_total[5m])), 1) > %g`, + r.Threshold) + case MetricPeakMemoryBytes: + return fmt.Sprintf( + `max by (policy_id) (symkernel_policy_verification_memory_bytes_max) > %g`, + r.Threshold) + default: + return "" + } +} + +// PrometheusRulesYAML renders the engine's rules as a Prometheus alerting- +// rules file, ready to drop into a Prometheus server or hand to +// `promtool check rules`. The returned YAML has one group, +// "symkernel_policy_degradation", containing one alert per rule. +func (e *AlertEngine) PrometheusRulesYAML() (string, error) { + group := promRuleGroup{Name: "symkernel_policy_degradation"} + for _, r := range e.rules { + expr := promQLFor(r) + if expr == "" { + continue + } + rule := promRule{ + Alert: r.ID, + Expr: expr, + For: promFor(r.For), + Labels: map[string]string{ + "severity": string(r.Severity), + }, + Annotations: map[string]string{ + "summary": r.Name, + "message": renderMessage(r.Message, "{{policy_id}}", r.Threshold), + "metric": string(r.Metric), + "threshold": fmt.Sprintf("%g", r.Threshold), + }, + } + group.Rules = append(group.Rules, rule) + } + + var buf bytes.Buffer + enc := yaml.NewEncoder(&buf) + enc.SetIndent(2) + if err := enc.Encode([]promRuleGroup{group}); err != nil { + return "", fmt.Errorf("observability: encode prometheus rules: %w", err) + } + if err := enc.Close(); err != nil { + return "", fmt.Errorf("observability: close prometheus rules encoder: %w", err) + } + return buf.String(), nil +} + +// promFor formats a duration as Prometheus' "for" field (e.g. "5m", "30s"), +// clamping zero to "0s". +func promFor(d time.Duration) string { + if d <= 0 { + return "0s" + } + return d.String() +} diff --git a/internal/observability/dashboard.go b/internal/observability/dashboard.go new file mode 100644 index 0000000..86fc305 --- /dev/null +++ b/internal/observability/dashboard.go @@ -0,0 +1,454 @@ +// dashboard.go — the per-policy performance dashboard. +// +// The dashboard is the human-facing summary of a Collector: it rolls the +// raw observation stream up into one PolicySummary per policy (latency +// distribution, error rate, resource usage) and renders both a JSON model +// for the symkernel HTTP API and a Grafana provisioning model that an +// operator can drop into a Grafana instance. The HTTP wiring lives in the +// Service type at the bottom of this file, which mounts the observability +// endpoints on a ServeMux the same way diagnostics/cache/audit do. + +package observability + +import ( + "encoding/json" + "net/http" + "time" +) + +// LatencyStats is the latency rollup shown on the dashboard. All millisecond +// fields are rounded to three decimals; percentiles are approximated from +// the latency histogram. +type LatencyStats struct { + Count uint64 `json:"count"` + MeanMs float64 `json:"mean_ms"` + MinMs float64 `json:"min_ms"` + MaxMs float64 `json:"max_ms"` + P50Ms float64 `json:"p50_ms"` + P95Ms float64 `json:"p95_ms"` + P99Ms float64 `json:"p99_ms"` + Buckets []Bucket `json:"buckets"` +} + +// Bucket is one latency-histogram bucket, exposed cumulatively (matching +// Prometheus convention) for direct plotting in Grafana. +type Bucket struct { + // UpperBound is the bucket's upper latency bound in seconds; the + // final bucket uses +Inf semantics (UpperBoundSeconds = -1 sentinel + // rendered as "+Inf" in Prometheus text). + UpperBoundSeconds float64 `json:"upper_bound_seconds"` + + // CumulativeCount is the number of observations at or below this + // bound. + CumulativeCount uint64 `json:"cumulative_count"` +} + +// ResourceSummary is the per-policy resource rollup. +type ResourceSummary struct { + // PeakMemoryBytes is the highest single-verification memory observed. + PeakMemoryBytes uint64 `json:"peak_memory_bytes"` + + // TotalInstructions is the cumulative instruction count across all + // observations of this policy. + TotalInstructions uint64 `json:"total_instructions"` +} + +// PolicySummary is the per-policy rollup rendered on the dashboard. It is +// the JSON shape returned by GET /v1/observability/dashboard. +type PolicySummary struct { + PolicyID string `json:"policy_id"` + Tier Tier `json:"tier"` + Total uint64 `json:"total"` + Successes uint64 `json:"successes"` + Failures uint64 `json:"failures"` + ErrorRate float64 `json:"error_rate"` + Latency LatencyStats `json:"latency"` + Resources ResourceSummary `json:"resources"` + // ErrorsByCode breaks failures down by error code; omitted when empty. + ErrorsByCode map[string]uint64 `json:"errors_by_code,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Dashboard is the full dashboard model: every policy the Collector has +// observed, plus an overall rollup. It is the JSON shape returned by +// GET /v1/observability/dashboard (under the "policies" key). +type Dashboard struct { + GeneratedAt time.Time `json:"generated_at"` + PolicyCount int `json:"policy_count"` + Overall PolicySummary `json:"overall"` + Policies []PolicySummary `json:"policies"` +} + +// summarize converts a raw policySeries into a PolicySummary. The latency +// distribution is copied out defensively so a later Record cannot mutate a +// summary the caller is holding. +func summarize(ps policySeries) PolicySummary { + out := PolicySummary{ + PolicyID: ps.policyID, + Tier: ps.tier, + Total: ps.total, + Successes: ps.successes, + Failures: ps.failures, + ErrorRate: errorRate(ps.total, ps.failures), + Latency: LatencyStats{ + Count: ps.lat.count, + MeanMs: ms(ps.lat.mean()), + MinMs: ms(ps.lat.min), + MaxMs: ms(ps.lat.max), + P50Ms: ms(ps.lat.percentile(0.50)), + P95Ms: ms(ps.lat.percentile(0.95)), + P99Ms: ms(ps.lat.percentile(0.99)), + Buckets: cumulativeBuckets(ps.lat.buckets), + }, + Resources: ResourceSummary{ + PeakMemoryBytes: ps.maxMemory, + TotalInstructions: ps.totalInstructions, + }, + ErrorsByCode: copyErrorCodes(ps.errorsByCode), + UpdatedAt: ps.lastUpdated, + } + return out +} + +// mean returns the average latency, or zero if there are no samples. +func (a latencyAgg) mean() time.Duration { + if a.count == 0 { + return 0 + } + return a.sum / time.Duration(a.count) +} + +// errorRate returns failures/total as a fraction in [0,1], guarding divide +// by zero. +func errorRate(total, failures uint64) float64 { + if total == 0 { + return 0 + } + return float64(failures) / float64(total) +} + +// cumulativeBuckets folds the non-cumulative bucket counts into the +// cumulative form the dashboard exposes. +func cumulativeBuckets(nonCum []uint64) []Bucket { + out := make([]Bucket, 0, len(latencyBucketsSeconds)+1) + cum := uint64(0) + for i, bound := range latencyBucketsSeconds { + cum += nonCum[i] + out = append(out, Bucket{ + UpperBoundSeconds: bound, + CumulativeCount: cum, + }) + } + // +Inf bucket carries the full count. + out = append(out, Bucket{ + UpperBoundSeconds: -1, + CumulativeCount: cum + nonCum[len(nonCum)-1], + }) + return out +} + +// copyErrorCodes returns a defensive copy of m, or nil if empty so the JSON +// field is omitted. +func copyErrorCodes(m map[string]uint64) map[string]uint64 { + if len(m) == 0 { + return nil + } + cp := make(map[string]uint64, len(m)) + for k, v := range m { + cp[k] = v + } + return cp +} + +// ms converts a duration to milliseconds rounded to three decimals. +func ms(d time.Duration) float64 { + return float64(d.Microseconds()) / 1000.0 +} + +// Render builds a Dashboard from the Collector. The returned model is a +// point-in-time snapshot; later observations do not affect it. +func Render(c *Collector) Dashboard { + series := c.allSeries() + policies := make([]PolicySummary, 0, len(series)) + // overall is pre-initialized with an allocated latency bucket slice so + // the empty-collector case still renders cleanly (cumulativeBuckets + // indexes the slice unconditionally). + overall := policySeries{lat: newLatencyAgg(), errorsByCode: make(map[string]uint64)} + for _, ps := range series { + policies = append(policies, summarize(ps)) + overall = mergeSeries(overall, ps) + } + overall.policyID = "__overall__" + overall.tier = TierUnknown + return Dashboard{ + GeneratedAt: time.Now().UTC(), + PolicyCount: len(policies), + Overall: summarize(overall), + Policies: policies, + } +} + +// mergeSeries folds src into dst, producing the combined aggregate used for +// the dashboard's overall rollup. Latency is merged by re-adding the bucket +// counts and recomputing min/max/sum/count. +func mergeSeries(dst, src policySeries) policySeries { + if dst.lat.buckets == nil { + dst.lat = newLatencyAgg() + } + if dst.errorsByCode == nil { + dst.errorsByCode = make(map[string]uint64) + } + if src.lat.count > 0 { + if dst.lat.count == 0 || src.lat.min < dst.lat.min { + dst.lat.min = src.lat.min + } + if src.lat.max > dst.lat.max { + dst.lat.max = src.lat.max + } + dst.lat.count += src.lat.count + dst.lat.sum += src.lat.sum + for i := range dst.lat.buckets { + dst.lat.buckets[i] += src.lat.buckets[i] + } + } + dst.total += src.total + dst.successes += src.successes + dst.failures += src.failures + for code, n := range src.errorsByCode { + dst.errorsByCode[code] += n + } + if src.maxMemory > dst.maxMemory { + dst.maxMemory = src.maxMemory + } + dst.totalInstructions += src.totalInstructions + if src.lastUpdated.After(dst.lastUpdated) { + dst.lastUpdated = src.lastUpdated + } + return dst +} + +// --- Grafana provisioning model --- + +// GrafanaDashboard is a minimal Grafana dashboard provisioning model. It is +// not the full Grafana schema — only enough to render the symkernel policy +// panels and a policy-id template variable. Operators paste the marshalled +// JSON into Grafana's "Import" flow or drop it under provisioning/dashboards. +type GrafanaDashboard struct { + UID string `json:"uid"` + Title string `json:"title"` + Tags []string `json:"tags"` + Timezone string `json:"timezone"` + SchemaVersion int `json:"schema_version"` + Refresh string `json:"refresh"` + Time grafanaTime `json:"time"` + Templating grafanaTemplating `json:"templating"` + Panels []GrafanaPanel `json:"panels"` +} + +type grafanaTime struct { + From string `json:"from"` + To string `json:"to"` +} + +type grafanaTemplating struct { + List []grafanaVariable `json:"list"` +} + +// grafanaVariable is a query variable backed by the policy_id label. +type grafanaVariable struct { + Name string `json:"name"` + Type string `json:"type"` + Label string `json:"label"` + IncludeAll bool `json:"includeAll"` + Multi bool `json:"multi"` + Query grafanaVarQuery `json:"query"` + Current map[string]string `json:"current"` +} + +type grafanaVarQuery struct { + Query string `json:"query"` + Format string `json:"format"` +} + +// GrafanaPanel is one dashboard panel. The PromQL expressions reference the +// Prometheus metric names emitted by Collector.PrometheusText and are +// scoped to the selected policy via the $policy_id variable. +type GrafanaPanel struct { + ID int `json:"id"` + Title string `json:"title"` + Type string `json:"type"` + GridPos grafanaGrid `json:"gridPos"` + Targets []grafanaTarget `json:"targets"` +} + +type grafanaGrid struct { + H int `json:"h"` + W int `json:"w"` + X int `json:"x"` + Y int `json:"y"` +} + +type grafanaTarget struct { + Expr string `json:"expr"` + LegendFmt string `json:"legendFormat"` + RefID string `json:"refId"` +} + +// GrafanaDashboardJSON builds a Grafana provisioning dashboard that graphs +// per-policy latency, error rate, and resource usage from the Prometheus +// metrics this package emits. title overrides the dashboard title; pass "" +// for the default. +func GrafanaDashboardJSON(title string) GrafanaDashboard { + if title == "" { + title = "symkernel — Policy Performance" + } + return GrafanaDashboard{ + UID: "symkernel-policy-perf", + Title: title, + Tags: []string{"symkernel", "verification"}, + Timezone: "browser", + SchemaVersion: 39, + Refresh: "10s", + Time: grafanaTime{From: "now-1h", To: "now"}, + Templating: grafanaTemplating{ + List: []grafanaVariable{{ + Name: "policy_id", + Type: "query", + Label: "Policy", + IncludeAll: true, + Multi: true, + Query: grafanaVarQuery{ + Query: `label_values(symkernel_policy_verification_duration_seconds_count, policy_id)`, + Format: "string", + }, + Current: map[string]string{"text": "All", "value": "$__all"}, + }}, + }, + Panels: []GrafanaPanel{ + { + ID: 1, Title: "Verification Latency (p50 / p95 / p99)", Type: "timeseries", + GridPos: grafanaGrid{H: 8, W: 12, X: 0, Y: 0}, + Targets: []grafanaTarget{ + {Expr: `histogram_quantile(0.50, sum by (le) (rate(symkernel_policy_verification_duration_seconds_bucket{policy_id=~"$policy_id"}[5m])))`, LegendFmt: "p50", RefID: "A"}, + {Expr: `histogram_quantile(0.95, sum by (le) (rate(symkernel_policy_verification_duration_seconds_bucket{policy_id=~"$policy_id"}[5m])))`, LegendFmt: "p95", RefID: "B"}, + {Expr: `histogram_quantile(0.99, sum by (le) (rate(symkernel_policy_verification_duration_seconds_bucket{policy_id=~"$policy_id"}[5m])))`, LegendFmt: "p99", RefID: "C"}, + }, + }, + { + ID: 2, Title: "Error Rate", Type: "timeseries", + GridPos: grafanaGrid{H: 8, W: 12, X: 12, Y: 0}, + Targets: []grafanaTarget{ + {Expr: `sum by (policy_id) (rate(symkernel_policy_verification_total{policy_id=~"$policy_id",result="failure"}[5m])) / clamp_min(sum by (policy_id) (rate(symkernel_policy_verification_total{policy_id=~"$policy_id"}[5m])), 1)`, LegendFmt: "{{policy_id}}", RefID: "A"}, + }, + }, + { + ID: 3, Title: "Peak Memory (bytes)", Type: "timeseries", + GridPos: grafanaGrid{H: 8, W: 12, X: 0, Y: 8}, + Targets: []grafanaTarget{ + {Expr: `max by (policy_id) (symkernel_policy_verification_memory_bytes_max{policy_id=~"$policy_id"})`, LegendFmt: "{{policy_id}}", RefID: "A"}, + }, + }, + { + ID: 4, Title: "Throughput (ops/s)", Type: "timeseries", + GridPos: grafanaGrid{H: 8, W: 12, X: 12, Y: 8}, + Targets: []grafanaTarget{ + {Expr: `sum by (policy_id) (rate(symkernel_policy_verification_duration_seconds_count{policy_id=~"$policy_id"}[5m]))`, LegendFmt: "{{policy_id}}", RefID: "A"}, + }, + }, + }, + } +} + +// --- HTTP service --- + +// Service is the HTTP façade over a Collector and AlertEngine. It mounts the +// observability read endpoints on a ServeMux, mirroring the RegisterRoutes +// pattern used by diagnostics, cache, and audit. It performs no mutation; +// callers feed the Collector out-of-band (typically from the verify +// handlers) and read the dashboard, metrics, and alerts over HTTP. +type Service struct { + collector *Collector + alerts *AlertEngine +} + +// NewService wires a Collector to the default AlertEngine and returns a +// Service ready to have its routes registered. Pass nil for c to get a +// fresh Collector. +func NewService(c *Collector) *Service { + if c == nil { + c = NewCollector() + } + return &Service{collector: c, alerts: DefaultAlertEngine()} +} + +// NewServiceWithAlerts is like NewService but with a custom AlertEngine, +// for operators that want different degradation thresholds. +func NewServiceWithAlerts(c *Collector, e *AlertEngine) *Service { + if c == nil { + c = NewCollector() + } + if e == nil { + e = DefaultAlertEngine() + } + return &Service{collector: c, alerts: e} +} + +// Collector exposes the underlying Collector so callers (e.g. a verify +// handler) can Record observations. +func (s *Service) Collector() *Collector { return s.collector } + +// RegisterRoutes mounts the observability endpoints on the given ServeMux: +// +// GET /v1/observability/dashboard — per-policy performance summary (JSON) +// GET /v1/observability/metrics — Prometheus text exposition +// GET /v1/observability/alerts — currently-firing degradation alerts +// GET /v1/observability/grafana — Grafana dashboard provisioning model +func (s *Service) RegisterRoutes(mux *http.ServeMux) { + mux.Handle("GET /v1/observability/dashboard", s.dashboardHandler()) + mux.Handle("GET /v1/observability/metrics", s.metricsHandler()) + mux.Handle("GET /v1/observability/alerts", s.alertsHandler()) + mux.Handle("GET /v1/observability/grafana", s.grafanaHandler()) +} + +func (s *Service) dashboardHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, Render(s.collector)) + } +} + +func (s *Service) metricsHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + _, _ = w.Write([]byte(s.collector.PrometheusText())) + } +} + +func (s *Service) alertsHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + alerts := s.alerts.Evaluate(s.collector) + writeJSON(w, http.StatusOK, alertsResponse{Alerts: alerts}) + } +} + +func (s *Service) grafanaHandler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, GrafanaDashboardJSON("")) + } +} + +// alertsResponse wraps the alerts slice so the JSON is an object, allowing +// future fields to be added without breaking the response shape. +type alertsResponse struct { + Alerts []Alert `json:"alerts"` +} + +// writeJSON encodes v as JSON with a 2-space indent and the given status. +// It is best-effort on the final encode error (a partial write to a +// ResponseWriter is unrecoverable, matching the diagnostics handler). +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + _ = enc.Encode(v) +} diff --git a/internal/observability/metrics.go b/internal/observability/metrics.go new file mode 100644 index 0000000..35a1c0b --- /dev/null +++ b/internal/observability/metrics.go @@ -0,0 +1,406 @@ +// Package observability provides policy performance dashboards and alerting +// for the symkernel verification service, the observability tier called out +// by Milestone 11. +// +// A Collector records per-policy verification observations — latency, +// outcome (pass/fail), and resource usage (memory and instructions) — and +// exposes them three ways: +// +// - as a Prometheus text exposition (PrometheusText) suitable for scraping +// at /metrics, +// - as a summarized dashboard (dashboard.go) consumed by operators and +// Grafana provisioning, +// - as degradation alerts (alerting.go) when latency, error rate, or +// resource usage cross configured thresholds. +// +// The package is transport agnostic: any verifier that knows its policy id, +// tier, and outcome can feed it an Observation. Latency is bucketed into a +// fixed Prometheus-style histogram so memory use stays bounded regardless of +// request volume; percentiles (p50/p95/p99) are approximated from the +// histogram, matching Prometheus' histogram_quantile semantics. +package observability + +import ( + "fmt" + "sort" + "strings" + "sync" + "time" +) + +// Tier labels the verification tier an observation originated from. It +// mirrors the three-tier model (CEL rules, wazero sandbox, Z3 SMT) that +// symkernel exposes. +type Tier string + +const ( + // TierCEL is the lightweight rules tier (cel-go). + TierCEL Tier = "cel" + // TierWasm is the hard-isolation tier (wazero sandbox). + TierWasm Tier = "wasm" + // TierZ3 is the formal-proof tier (Z3 SMT solver). + TierZ3 Tier = "z3" + // TierUnknown is used when the tier is not known, so a stray empty + // string still produces a valid (if unhelpful) label. + TierUnknown Tier = "unknown" +) + +// String returns the canonical lower-case label for the tier, normalizing +// the empty string to "unknown". +func (t Tier) String() string { + if t == "" { + return string(TierUnknown) + } + return string(t) +} + +// ResourceUsage captures the resource cost of a single verification. Fields +// are optional: a CEL evaluation records no instructions, while a wazero +// sandbox run records both memory and instructions. +type ResourceUsage struct { + // MemoryBytes is the peak memory consumed during verification, in + // bytes (e.g. the wazero guest's high-water mark). + MemoryBytes uint64 + + // InstructionCount is the number of instructions executed (e.g. the + // wazero metering counter). Zero means "not measured". + InstructionCount uint64 +} + +// Observation is a single verification event recorded by the Collector. It +// is the unit of input the dashboard, metrics, and alerts are built from. +type Observation struct { + // PolicyID identifies the policy that was evaluated (e.g. + // "rate-limit-v2"). Required. + PolicyID string + + // Tier is the verification tier that produced the observation. + Tier Tier + + // Latency is the wall-clock duration the verification took. + Latency time.Duration + + // Success reports whether the verification completed without error. + // A failed verification (policy rejected the input) is still a + // Success if the verifier ran cleanly; set Success=false only when + // the verifier itself erred or timed out. + Success bool + + // ErrorCode is an optional machine-readable failure code (e.g. + // "timeout", "trap", "malformed"). Populated only when Success is + // false. It drives the errors-by-code breakdown. + ErrorCode string + + // Resources is the resource cost of the verification. + Resources ResourceUsage + + // Timestamp is when the observation was produced. If zero, Record + // stamps it with the current time. + Timestamp time.Time +} + +// latencyBucketsSeconds are the histogram bucket upper bounds for +// verification latency, in seconds. They are the Prometheus default +// latency buckets (0.005s … 10s), which span the full symkernel SLO range: +// a CEL hit is sub-millisecond, a Z3 proof can run for seconds. +var latencyBucketsSeconds = []float64{ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, +} + +// numLatencyBuckets is the per-policy bucket array length: one slot per +// declared upper bound plus a final +Inf overflow bucket. It is a package +// variable (not a const) because len over a slice is not a constant +// expression. +var numLatencyBuckets = len(latencyBucketsSeconds) + 1 + +// latencyAgg accumulates the latency distribution for a single policy. The +// buckets slice holds non-cumulative counts: bucket[i] is the number of +// observations with latency in (bounds[i-1], bounds[i]], with bucket[len-1] +// the +Inf bucket capturing everything above the largest bound. +type latencyAgg struct { + count uint64 + sum time.Duration + min time.Duration + max time.Duration + buckets []uint64 // len == numLatencyBuckets +} + +// newLatencyAgg returns a zeroed latencyAgg with an allocated bucket slice. +func newLatencyAgg() latencyAgg { + return latencyAgg{ + min: 0, + max: 0, + buckets: make([]uint64, numLatencyBuckets), + } +} + +// record adds a single latency sample to the aggregate. +func (a *latencyAgg) record(d time.Duration) { + if a.count == 0 || d < a.min { + a.min = d + } + if d > a.max { + a.max = d + } + a.count++ + a.sum += d + a.buckets[latencyBucketIndex(d)]++ +} + +// latencyBucketIndex returns the non-cumulative bucket index for a duration. +func latencyBucketIndex(d time.Duration) int { + s := d.Seconds() + for i, b := range latencyBucketsSeconds { + if s <= b { + return i + } + } + return numLatencyBuckets - 1 // +Inf overflow bucket +} + +// percentile approximates the q-quantile (0 <= q <= 1) of the latency +// distribution via linear interpolation between bucket boundaries, matching +// Prometheus' histogram_quantile behavior. Returns 0 if there are no +// samples. +func (a latencyAgg) percentile(q float64) time.Duration { + if a.count == 0 { + return 0 + } + if q <= 0 { + return a.min + } + if q >= 1 { + return a.max + } + target := q * float64(a.count) + cum := uint64(0) + prevBound := 0.0 + for i, bound := range latencyBucketsSeconds { + bucketCount := a.buckets[i] + cumPrev := cum + cum += bucketCount + if float64(cum) >= target { + if bucketCount == 0 { + return secondsToDuration(bound) + } + // Linearly interpolate within this bucket. + frac := (target - float64(cumPrev)) / float64(bucketCount) + if frac < 0 { + frac = 0 + } else if frac > 1 { + frac = 1 + } + return secondsToDuration(prevBound + frac*(bound-prevBound)) + } + prevBound = bound + } + // Target falls in the +Inf bucket: return the observed max. + return a.max +} + +// secondsToDuration converts a float second value to a time.Duration, +// clamping negatives to zero. +func secondsToDuration(s float64) time.Duration { + if s <= 0 { + return 0 + } + return time.Duration(s * float64(time.Second)) +} + +// policySeries is the full per-policy metric set maintained by the Collector. +type policySeries struct { + policyID string + tier Tier + + lat latencyAgg + + total uint64 + successes uint64 + failures uint64 + + errorsByCode map[string]uint64 + + maxMemory uint64 + totalInstructions uint64 + + lastUpdated time.Time +} + +// Collector aggregates per-policy verification metrics. It is safe for +// concurrent use: Record may be called from many goroutines (one per +// in-flight verification) while Snapshot/PrometheusText are read concurrently. +type Collector struct { + mu sync.RWMutex + series map[string]*policySeries +} + +// NewCollector creates an empty Collector. +func NewCollector() *Collector { + return &Collector{series: make(map[string]*policySeries)} +} + +// Record ingests a single Observation. It is safe for concurrent use. An +// observation with an empty PolicyID is dropped — there is no series to +// attribute it to — because the dashboard, metrics, and alerts are all +// per-policy. +func (c *Collector) Record(obs Observation) { + if obs.PolicyID == "" { + return + } + if obs.Timestamp.IsZero() { + obs.Timestamp = time.Now().UTC() + } + + c.mu.Lock() + defer c.mu.Unlock() + + ps := c.series[obs.PolicyID] + if ps == nil { + ps = &policySeries{ + policyID: obs.PolicyID, + tier: obs.Tier, + lat: newLatencyAgg(), + errorsByCode: make(map[string]uint64), + } + c.series[obs.PolicyID] = ps + } + // Keep the most-recently-seen tier so the dashboard reflects the + // active policy owner even if a policy is re-keyed across tiers. + ps.tier = obs.Tier + + ps.lat.record(obs.Latency) + ps.total++ + if obs.Success { + ps.successes++ + } else { + ps.failures++ + if obs.ErrorCode != "" { + ps.errorsByCode[obs.ErrorCode]++ + } else { + ps.errorsByCode["unknown"]++ + } + } + if obs.Resources.MemoryBytes > ps.maxMemory { + ps.maxMemory = obs.Resources.MemoryBytes + } + ps.totalInstructions += obs.Resources.InstructionCount + ps.lastUpdated = obs.Timestamp +} + +// snapshotSeries returns a defensive copy of the named series. The boolean +// reports whether the policy was known. +func (c *Collector) snapshotSeries(policyID string) (policySeries, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + ps, ok := c.series[policyID] + if !ok { + return policySeries{}, false + } + return *ps, true +} + +// allSeries returns a stable snapshot of every series, sorted by policy id +// for deterministic output across exposition and tests. +func (c *Collector) allSeries() []policySeries { + c.mu.RLock() + defer c.mu.RUnlock() + out := make([]policySeries, 0, len(c.series)) + for _, ps := range c.series { + out = append(out, *ps) + } + sort.Slice(out, func(i, j int) bool { return out[i].policyID < out[j].policyID }) + return out +} + +// PolicyIDs returns the sorted set of policy ids the Collector has observed. +func (c *Collector) PolicyIDs() []string { + c.mu.RLock() + defer c.mu.RUnlock() + ids := make([]string, 0, len(c.series)) + for id := range c.series { + ids = append(ids, id) + } + sort.Strings(ids) + return ids +} + +// PrometheusText exposes the collected metrics in the Prometheus text +// exposition format (version 0.0.4), suitable for scraping at /metrics. +// It emits, per policy: +// +// symkernel_policy_verification_duration_seconds_bucket{...} (cumulative) +// symkernel_policy_verification_duration_seconds_sum{...} +// symkernel_policy_verification_duration_seconds_count{...} +// symkernel_policy_verification_total{...,result=success|failure} +// symkernel_policy_verification_errors_total{...,code=...} +// symkernel_policy_verification_memory_bytes_max{...} +// symkernel_policy_verification_instructions_total{...} +func (c *Collector) PrometheusText() string { + series := c.allSeries() + var b strings.Builder + b.WriteString("# HELP symkernel_policy_verification_duration_seconds Verification latency distribution.\n") + b.WriteString("# TYPE symkernel_policy_verification_duration_seconds histogram\n") + b.WriteString("# HELP symkernel_policy_verification_total Total verifications by policy and outcome.\n") + b.WriteString("# TYPE symkernel_policy_verification_total counter\n") + b.WriteString("# HELP symkernel_policy_verification_errors_total Verification errors by policy and code.\n") + b.WriteString("# TYPE symkernel_policy_verification_errors_total counter\n") + b.WriteString("# HELP symkernel_policy_verification_memory_bytes_max Peak memory consumed by a verification, in bytes.\n") + b.WriteString("# TYPE symkernel_policy_verification_memory_bytes_max gauge\n") + b.WriteString("# HELP symkernel_policy_verification_instructions_total Cumulative instructions executed across verifications.\n") + b.WriteString("# TYPE symkernel_policy_verification_instructions_total counter\n") + + for _, ps := range series { + labels := fmt.Sprintf(`policy_id=%q,tier=%q`, ps.policyID, ps.tier.String()) + + // Cumulative histogram buckets. + cum := uint64(0) + for i, bound := range latencyBucketsSeconds { + cum += ps.lat.buckets[i] + fmt.Fprintf(&b, + `symkernel_policy_verification_duration_seconds_bucket{%s,le="%g"} %d`+"\n", + labels, bound, cum) + } + // +Inf bucket equals the total count. + fmt.Fprintf(&b, + `symkernel_policy_verification_duration_seconds_bucket{%s,le="+Inf"} %d`+"\n", + labels, ps.lat.count) + fmt.Fprintf(&b, + `symkernel_policy_verification_duration_seconds_sum{%s} %g`+"\n", + labels, ps.lat.sum.Seconds()) + fmt.Fprintf(&b, + `symkernel_policy_verification_duration_seconds_count{%s} %d`+"\n", + labels, ps.lat.count) + + fmt.Fprintf(&b, + `symkernel_policy_verification_total{%s,result="success"} %d`+"\n", + labels, ps.successes) + fmt.Fprintf(&b, + `symkernel_policy_verification_total{%s,result="failure"} %d`+"\n", + labels, ps.failures) + + // Errors by code, sorted for stable output. + for _, code := range sortedKeys(ps.errorsByCode) { + fmt.Fprintf(&b, + `symkernel_policy_verification_errors_total{%s,code=%q} %d`+"\n", + labels, code, ps.errorsByCode[code]) + } + + fmt.Fprintf(&b, + `symkernel_policy_verification_memory_bytes_max{%s} %d`+"\n", + labels, ps.maxMemory) + fmt.Fprintf(&b, + `symkernel_policy_verification_instructions_total{%s} %d`+"\n", + labels, ps.totalInstructions) + } + return b.String() +} + +// sortedKeys returns the keys of m sorted lexicographically. +func sortedKeys[V any](m map[string]V) []string { + keys := make([]string, 0, len(m)) + for k := range m { + keys = append(keys, k) + } + sort.Strings(keys) + return keys +} diff --git a/internal/observability/observability_test.go b/internal/observability/observability_test.go new file mode 100644 index 0000000..47ae004 --- /dev/null +++ b/internal/observability/observability_test.go @@ -0,0 +1,511 @@ +package observability + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// recordN records n identical observations for quick setup. +func recordN(c *Collector, policyID string, tier Tier, latency time.Duration, success bool, n int) { + for i := 0; i < n; i++ { + c.Record(Observation{ + PolicyID: policyID, + Tier: tier, + Latency: latency, + Success: success, + }) + } +} + +// --- Collector basics --- + +func TestRecordEmptyPolicyIDIsDropped(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "", Latency: 10 * time.Millisecond}) + if got := c.PolicyIDs(); len(got) != 0 { + t.Fatalf("empty policy id should be dropped, got %v", got) + } +} + +func TestRecordStampsTimestampWhenZero(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "p1", Latency: time.Millisecond}) + _, ok := c.snapshotSeries("p1") + if !ok { + t.Fatal("expected series p1") + } +} + +func TestTierStringNormalizesEmpty(t *testing.T) { + if got := Tier("").String(); got != "unknown" { + t.Fatalf(`Tier("").String() = %q, want "unknown"`, got) + } + if got := TierCEL.String(); got != "cel" { + t.Fatalf("TierCEL.String() = %q, want \"cel\"", got) + } +} + +// --- latency histogram + percentiles --- + +func TestLatencyPercentilesMonotonicAndBounded(t *testing.T) { + c := NewCollector() + // A spread of latencies spanning several buckets. + latencies := []time.Duration{ + 1 * time.Millisecond, 2 * time.Millisecond, 3 * time.Millisecond, + 10 * time.Millisecond, 20 * time.Millisecond, 30 * time.Millisecond, + 100 * time.Millisecond, 200 * time.Millisecond, 300 * time.Millisecond, + } + for _, d := range latencies { + c.Record(Observation{PolicyID: "spread", Tier: TierCEL, Latency: d}) + } + ps, ok := c.snapshotSeries("spread") + if !ok { + t.Fatal("missing series") + } + p50 := ps.lat.percentile(0.50) + p95 := ps.lat.percentile(0.95) + p99 := ps.lat.percentile(0.99) + if p50 > p95 || p95 > p99 { + t.Fatalf("percentiles not monotonic: p50=%v p95=%v p99=%v", p50, p95, p99) + } + // Lower bound holds: percentiles are >= the observed minimum. + if p50 < ps.lat.min { + t.Fatalf("p50 below min: p50=%v min=%v", p50, ps.lat.min) + } + // Histogram quantiles (like Prometheus histogram_quantile) approximate + // by interpolating between bucket boundaries, so p99 can overshoot the + // true sample max when the quantile lands in the top occupied bucket. + // It must never exceed the largest declared bucket bound, though. + ceiling := time.Duration(latencyBucketsSeconds[len(latencyBucketsSeconds)-1] * float64(time.Second)) + if p99 > ceiling { + t.Fatalf("p99 above histogram ceiling: p99=%v ceiling=%v", p99, ceiling) + } +} + +func TestLatencyPercentileNoSamples(t *testing.T) { + a := newLatencyAgg() + if got := a.percentile(0.95); got != 0 { + t.Fatalf("percentile of empty agg = %v, want 0", got) + } +} + +func TestLatencyBucketIndexOverflow(t *testing.T) { + // A latency above the largest bucket lands in the +Inf overflow bucket. + idx := latencyBucketIndex(60 * time.Second) // 60s > 10s max bucket + if idx != numLatencyBuckets-1 { + t.Fatalf("overflow bucket index = %d, want %d", idx, numLatencyBuckets-1) + } + // A tiny latency lands in the first bucket. + if got := latencyBucketIndex(time.Microsecond); got != 0 { + t.Fatalf("first bucket index = %d, want 0", got) + } +} + +// --- error rate + resources --- + +func TestErrorRateAndResourceTracking(t *testing.T) { + c := NewCollector() + recordN(c, "policy", TierWasm, 5*time.Millisecond, true, 80) + recordN(c, "policy", TierWasm, 5*time.Millisecond, false, 20) + // Two sandbox runs with known resource costs. + c.Record(Observation{PolicyID: "policy", Tier: TierWasm, Latency: 1 * time.Millisecond, Success: true, + Resources: ResourceUsage{MemoryBytes: 1024, InstructionCount: 100}}) + c.Record(Observation{PolicyID: "policy", Tier: TierWasm, Latency: 1 * time.Millisecond, Success: true, + Resources: ResourceUsage{MemoryBytes: 4096, InstructionCount: 200}}) + + ps, ok := c.snapshotSeries("policy") + if !ok { + t.Fatal("missing series") + } + if ps.total != 102 { + t.Fatalf("total = %d, want 102", ps.total) + } + if ps.successes != 82 || ps.failures != 20 { + t.Fatalf("successes=%d failures=%d, want 82/20", ps.successes, ps.failures) + } + if er := errorRate(ps.total, ps.failures); er < 0.19 || er > 0.21 { + t.Fatalf("error rate = %.3f, want ~0.196", er) + } + if ps.maxMemory != 4096 { + t.Fatalf("max memory = %d, want 4096", ps.maxMemory) + } + if ps.totalInstructions != 300 { + t.Fatalf("total instructions = %d, want 300", ps.totalInstructions) + } +} + +func TestFailedObservationWithoutErrorCodeDefaultsToUnknown(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "p", Tier: TierCEL, Latency: time.Millisecond, Success: false}) + ps, ok := c.snapshotSeries("p") + if !ok { + t.Fatal("missing series") + } + if ps.errorsByCode["unknown"] != 1 { + t.Fatalf("expected unknown error code, got %v", ps.errorsByCode) + } +} + +func TestFailedObservationWithErrorCode(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "p", Tier: TierCEL, Latency: time.Millisecond, Success: false, ErrorCode: "timeout"}) + c.Record(Observation{PolicyID: "p", Tier: TierCEL, Latency: time.Millisecond, Success: false, ErrorCode: "timeout"}) + ps, ok := c.snapshotSeries("p") + if !ok { + t.Fatal("missing series") + } + if ps.errorsByCode["timeout"] != 2 { + t.Fatalf("timeout count = %d, want 2", ps.errorsByCode["timeout"]) + } +} + +// --- Dashboard --- + +func TestRenderSummarizesAndRollsUp(t *testing.T) { + c := NewCollector() + recordN(c, "alpha", TierCEL, 2*time.Millisecond, true, 10) + recordN(c, "beta", TierZ3, 200*time.Millisecond, true, 5) + + dash := Render(c) + if dash.PolicyCount != 2 { + t.Fatalf("policy count = %d, want 2", dash.PolicyCount) + } + if len(dash.Policies) != 2 { + t.Fatalf("policies len = %d, want 2", len(dash.Policies)) + } + if dash.Policies[0].PolicyID != "alpha" { + t.Fatalf("expected sorted first policy alpha, got %q", dash.Policies[0].PolicyID) + } + // Overall rollup aggregates both policies. + if dash.Overall.Total != 15 { + t.Fatalf("overall total = %d, want 15", dash.Overall.Total) + } +} + +func TestRenderEmptyCollector(t *testing.T) { + c := NewCollector() + dash := Render(c) + if dash.PolicyCount != 0 || len(dash.Policies) != 0 { + t.Fatalf("empty dashboard should have no policies, got %+v", dash) + } + if dash.Overall.Total != 0 { + t.Fatalf("empty overall total = %d, want 0", dash.Overall.Total) + } +} + +func TestLatencyStatsBucketsAreCumulative(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "p", Tier: TierCEL, Latency: 1 * time.Millisecond}) + c.Record(Observation{PolicyID: "p", Tier: TierCEL, Latency: 1 * time.Millisecond}) + dash := Render(c) + ps := dash.Policies[0] + // Cumulative counts must be non-decreasing and end at the total. + var prev uint64 + for _, b := range ps.Latency.Buckets { + if b.CumulativeCount < prev { + t.Fatalf("cumulative bucket decreased: %d < %d", b.CumulativeCount, prev) + } + prev = b.CumulativeCount + } + if prev != ps.Total { + t.Fatalf("final cumulative bucket = %d, want total %d", prev, ps.Total) + } +} + +func TestSummarizeOmitsEmptyErrorsByCode(t *testing.T) { + c := NewCollector() + recordN(c, "p", TierCEL, time.Millisecond, true, 3) + dash := Render(c) + if dash.Policies[0].ErrorsByCode != nil { + t.Fatalf("expected ErrorsByCode nil for all-success policy, got %v", dash.Policies[0].ErrorsByCode) + } +} + +// --- Prometheus text --- + +func TestPrometheusTextContainsExpectedMetrics(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "rate-limit", Tier: TierCEL, Latency: 1 * time.Millisecond, Success: true}) + c.Record(Observation{PolicyID: "rate-limit", Tier: TierCEL, Latency: 2 * time.Millisecond, Success: false, ErrorCode: "timeout"}) + + out := c.PrometheusText() + wantSubstrings := []string{ + "# TYPE symkernel_policy_verification_duration_seconds histogram", + `symkernel_policy_verification_duration_seconds_bucket{policy_id="rate-limit",tier="cel",le="+Inf"} 2`, + `symkernel_policy_verification_duration_seconds_count{policy_id="rate-limit",tier="cel"} 2`, + `symkernel_policy_verification_total{policy_id="rate-limit",tier="cel",result="success"} 1`, + `symkernel_policy_verification_total{policy_id="rate-limit",tier="cel",result="failure"} 1`, + `symkernel_policy_verification_errors_total{policy_id="rate-limit",tier="cel",code="timeout"} 1`, + `symkernel_policy_verification_memory_bytes_max{policy_id="rate-limit",tier="cel"} 0`, + } + for _, s := range wantSubstrings { + if !strings.Contains(out, s) { + t.Errorf("PrometheusText missing %q\n--- output ---\n%s", s, out) + } + } +} + +func TestPrometheusTextEmptyCollectorHasHelpLines(t *testing.T) { + c := NewCollector() + out := c.PrometheusText() + if !strings.Contains(out, "# HELP symkernel_policy_verification_total") { + t.Errorf("empty exposition should still emit HELP/TYPE lines\n%s", out) + } + if strings.Contains(out, "policy_id=") { + t.Errorf("empty exposition should not emit any series\n%s", out) + } +} + +// --- Alerting --- + +func TestEvaluateFiresLatencyAlert(t *testing.T) { + c := NewCollector() + // p95 well above the 500ms warning threshold. + recordN(c, "slow", TierZ3, 800*time.Millisecond, true, 25) + engine := DefaultAlertEngine() + alerts := engine.Evaluate(c) + found := false + for _, a := range alerts { + if a.RuleID == "PolicyLatencyP95High" && a.PolicyID == "slow" { + found = true + if a.Severity != SeverityWarning { + t.Errorf("latency high alert severity = %q, want warning", a.Severity) + } + } + } + if !found { + t.Fatalf("expected PolicyLatencyP95High alert, got %+v", alerts) + } +} + +func TestEvaluateFiresErrorRateAndCriticalAlerts(t *testing.T) { + c := NewCollector() + // 50% failure rate -> both error-rate rules fire; p95 50ms is under + // the latency threshold so no latency alert. + for i := 0; i < 100; i++ { + c.Record(Observation{PolicyID: "flaky", Tier: TierWasm, Latency: 50 * time.Millisecond, Success: i%2 == 0}) + } + engine := DefaultAlertEngine() + alerts := engine.Evaluate(c) + + rules := map[string]bool{} + for _, a := range alerts { + rules[a.RuleID] = true + } + if !rules["PolicyErrorRateHigh"] || !rules["PolicyErrorRateCritical"] { + t.Errorf("expected both error-rate alerts, got %v", rules) + } + if rules["PolicyLatencyP95High"] { + t.Errorf("latency alert should not fire at p95=50ms") + } +} + +func TestEvaluateFiresMemoryAlert(t *testing.T) { + c := NewCollector() + c.Record(Observation{PolicyID: "hog", Tier: TierWasm, Latency: time.Millisecond, Success: true, + Resources: ResourceUsage{MemoryBytes: 600 * 1024 * 1024}}) + engine := DefaultAlertEngine() + alerts := engine.Evaluate(c) + var found bool + for _, a := range alerts { + if a.RuleID == "PolicyPeakMemoryCritical" { + found = true + if a.Observed < 600*1024*1024 { + t.Errorf("memory alert observed = %g, want >= 600MiB", a.Observed) + } + } + } + if !found { + t.Fatalf("expected memory alert, got %+v", alerts) + } +} + +func TestEvaluateRespectsMinSamples(t *testing.T) { + c := NewCollector() + // A single slow observation: PolicyLatencyP95High requires MinSamples=20. + c.Record(Observation{PolicyID: "oneshot", Tier: TierZ3, Latency: 2 * time.Second, Success: true}) + engine := DefaultAlertEngine() + alerts := engine.Evaluate(c) + for _, a := range alerts { + if a.RuleID == "PolicyLatencyP95High" { + t.Fatalf("PolicyLatencyP95High must respect MinSamples=20, fired: %+v", a) + } + } +} + +func TestEvaluateNoAlertsForHealthyPolicy(t *testing.T) { + c := NewCollector() + recordN(c, "healthy", TierCEL, time.Millisecond, true, 100) + engine := DefaultAlertEngine() + if alerts := engine.Evaluate(c); len(alerts) != 0 { + t.Fatalf("healthy policy should produce no alerts, got %+v", alerts) + } +} + +func TestEvaluateSortsCriticalFirst(t *testing.T) { + c := NewCollector() + // p95 over 1s fires warning AND critical latency alerts. + recordN(c, "slow", TierZ3, 1200*time.Millisecond, true, 25) + engine := DefaultAlertEngine() + alerts := engine.Evaluate(c) + if len(alerts) < 2 { + t.Fatalf("expected >=2 alerts, got %d", len(alerts)) + } + if alerts[0].Severity != SeverityCritical { + t.Errorf("first alert should be critical, got %q", alerts[0].Severity) + } +} + +func TestAlertRulesSortedCriticalFirstUnit(t *testing.T) { + c := NewCollector() + recordN(c, "slow", TierZ3, 1200*time.Millisecond, true, 25) + for _, a := range DefaultAlertEngine().Evaluate(c) { + if a.Unit == "" { + t.Errorf("alert %q has empty unit", a.RuleID) + } + if !strings.Contains(a.Message, "slow") { + t.Errorf("alert message should substitute policy_id, got %q", a.Message) + } + } +} + +func TestPrometheusRulesYAMLIsParsable(t *testing.T) { + engine := DefaultAlertEngine() + out, err := engine.PrometheusRulesYAML() + if err != nil { + t.Fatalf("PrometheusRulesYAML: %v", err) + } + wantSubstrings := []string{ + "name: symkernel_policy_degradation", + "alert: PolicyLatencyP95High", + "expr:", + "severity: warning", + "histogram_quantile(0.95", + } + for _, s := range wantSubstrings { + if !strings.Contains(out, s) { + t.Errorf("rules YAML missing %q\n---\n%s", s, out) + } + } +} + +// --- HTTP Service --- + +func TestServiceDashboardHandler(t *testing.T) { + svc := NewService(nil) + recordN(svc.Collector(), "p1", TierCEL, time.Millisecond, true, 5) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/observability/dashboard", nil) + svc.dashboardHandler().ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); !strings.HasPrefix(ct, "application/json") { + t.Errorf("content-type = %q, want application/json", ct) + } + if !strings.Contains(rec.Body.String(), `"policy_id": "p1"`) { + t.Errorf("dashboard body missing p1: %s", rec.Body.String()) + } +} + +func TestServiceMetricsHandlerContentType(t *testing.T) { + svc := NewService(nil) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/observability/metrics", nil) + svc.metricsHandler().ServeHTTP(rec, req) + if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "text/plain") { + t.Errorf("metrics content-type = %q, want text/plain", ct) + } + if !strings.Contains(rec.Body.String(), "# TYPE symkernel") { + t.Errorf("metrics body missing TYPE lines: %s", rec.Body.String()) + } +} + +func TestServiceAlertsHandlerReturnsObject(t *testing.T) { + svc := NewService(nil) + recordN(svc.Collector(), "slow", TierZ3, 800*time.Millisecond, true, 25) + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/observability/alerts", nil) + svc.alertsHandler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + if !strings.Contains(rec.Body.String(), `"alerts":`) { + t.Errorf("alerts body should be an object with alerts key: %s", rec.Body.String()) + } + if !strings.Contains(rec.Body.String(), "PolicyLatencyP95High") { + t.Errorf("alerts body should contain firing rule: %s", rec.Body.String()) + } +} + +func TestServiceGrafanaHandlerReturnsDashboard(t *testing.T) { + svc := NewService(nil) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/v1/observability/grafana", nil) + svc.grafanaHandler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rec.Code) + } + body := rec.Body.String() + if !strings.Contains(body, `"uid": "symkernel-policy-perf"`) { + t.Errorf("grafana body missing uid: %s", body) + } + if !strings.Contains(body, "histogram_quantile") { + t.Errorf("grafana body missing prometheus query: %s", body) + } +} + +func TestRegisterRoutesMountsAllFour(t *testing.T) { + svc := NewService(nil) + mux := http.NewServeMux() + svc.RegisterRoutes(mux) + + for _, path := range []string{ + "/v1/observability/dashboard", + "/v1/observability/metrics", + "/v1/observability/alerts", + "/v1/observability/grafana", + } { + _, pattern := mux.Handler(httptest.NewRequest(http.MethodGet, path, nil)) + // ServeMux returns the fully-qualified pattern, method-prefixed. + if want := "GET " + path; pattern != want { + t.Errorf("route %q not registered, got pattern %q", path, pattern) + } + } +} + +// --- concurrency --- + +func TestRecordConcurrentSafe(t *testing.T) { + c := NewCollector() + done := make(chan struct{}) + for g := 0; g < 8; g++ { + go func(id int) { + policy := "p" + if id%2 == 0 { + policy = "q" + } + for i := 0; i < 200; i++ { + c.Record(Observation{PolicyID: policy, Tier: TierCEL, Latency: time.Millisecond, Success: true}) + } + done <- struct{}{} + }(g) + } + for g := 0; g < 8; g++ { + <-done + } + // 8 goroutines × 200 = 1600 observations split across p and q. + total := 0 + for _, ps := range c.allSeries() { + total += int(ps.total) + } + if total != 1600 { + t.Fatalf("concurrent record lost samples: total = %d, want 1600", total) + } +}