diff --git a/server/chat/backend/agent/access/mode_access_controller.py b/server/chat/backend/agent/access/mode_access_controller.py index 27bd0a8ae..3838f462d 100644 --- a/server/chat/backend/agent/access/mode_access_controller.py +++ b/server/chat/backend/agent/access/mode_access_controller.py @@ -85,7 +85,7 @@ def filter_tools(cls, mode: Optional[str], tools: Sequence[StructuredTool]) -> L LOGGER.info("ModeAccessController dropped tool %s due to read-only mode prefix match", name) continue - if name in {"iac_tool", "github_commit"}: + if name in {"iac_tool", "github_commit", "send_hpa_vpa_recommendation"}: LOGGER.info("ModeAccessController dropped tool %s for read-only mode", name) continue @@ -134,7 +134,7 @@ def is_tool_allowed(cls, mode: Optional[str], tool_name: str) -> bool: if any(name.startswith(prefix) for prefix in cls._POLICY.blocked_tool_prefixes): return False - return name not in {"iac_tool", "github_commit"} + return name not in {"iac_tool", "github_commit", "send_hpa_vpa_recommendation"} __all__ = ["ModeAccessController"] diff --git a/server/chat/backend/agent/skills/integrations/datadog/SKILL.md b/server/chat/backend/agent/skills/integrations/datadog/SKILL.md index 45840d032..818ee6877 100644 --- a/server/chat/backend/agent/skills/integrations/datadog/SKILL.md +++ b/server/chat/backend/agent/skills/integrations/datadog/SKILL.md @@ -25,16 +25,56 @@ Datadog integration for querying observability data during Root Cause Analysis. ## Instructions ### Tool Usage -`query_datadog(resource_type=TYPE, query=QUERY, time_from=START, time_to=END, limit=N)` +`query_datadog(resource_type=TYPE, query=QUERY, time_from=START, time_to=END, limit=N, interval=MS)` ### Resource Types 1. `'logs'` -- Search log entries. query=Datadog log query syntax e.g. `"service:web status:error"` -2. `'metrics'` -- Query metric timeseries. query=metric query e.g. `"avg:system.cpu.user{*}"` -3. `'monitors'` -- List monitors with status. query=name filter (optional) -4. `'events'` -- Platform events. query=source filter (optional) -5. `'traces'` -- APM spans/traces. query=span query e.g. `"service:web @http.status_code:500"` -6. `'hosts'` -- Infrastructure hosts. query=host filter (optional) -7. `'incidents'` -- Datadog incidents. Lists active/recent incidents (requires Incident Management; may 403 if not enabled). +2. `'metrics'` -- Query metric timeseries (raw points). query=metric query e.g. `"avg:system.cpu.user{*}"` +3. `'metric_stats'` -- Percentile summary per series (p50/p95/p99/max/mean). Same metric query + syntax as `'metrics'`, but returns one compact row per series instead of raw points. + Use this for capacity and right-sizing questions over long windows. +4. `'monitors'` -- List monitors with status. query=name filter (optional) +5. `'events'` -- Platform events. query=source filter (optional) +6. `'traces'` -- APM spans/traces. query=span query e.g. `"service:web @http.status_code:500"` +7. `'hosts'` -- Infrastructure hosts. query=host filter (optional) +8. `'incidents'` -- Datadog incidents. Lists active/recent incidents (requires Incident Management; may 403 if not enabled). + +### The `interval` Parameter +`interval` is the rollup granularity in **milliseconds**, and applies to `'metrics'` and +`'metric_stats'`. Omit it and a granularity is auto-picked that keeps each series under +~1000 points -- a 30-day window auto-picks `3600000` (1 hour, 720 points). Values are +clamped to `60000`..`14400000`. Datadog caps a series at 1500 points, so a long window +with a fine interval returns less than you asked for; prefer the auto-pick. + +### Percentiles +**Datadog cannot compute a time-percentile.** Do not attempt any of these -- every one +is rejected or silently empty: +- `.rollup(percentile, 95, 3600)` and `.rollup(p95, 3600)` -- `400 Unrecognized rollup method`. + `.rollup()` accepts only `avg`, `sum`, `min`, `max`, `count`. +- `p95:my.metric{...}` -- returns `200` with **zero series** for gauges. The `pXX:` prefix + needs *distribution* metrics; `kubernetes.cpu.usage.total` and `kubernetes.memory.usage` + are gauges, so it can never apply to them. +- `formula: "p95(a)"` -- `400 function "p95()" does not exist`. +- `formula: "percentile(a, 95, 3600)"` -- `percentile()` exists but is a **space** aggregator: + arguments 2 and 3 are *group tags*, not a percentile value and window. +- scalar `aggregator: "percentile"` or `"p95"` -- `400`. + +Use `resource_type='metric_stats'` instead. It fetches the rolled-up points and computes +percentiles server-side, using **nearest-rank** order statistics -- so every reported p95 +is a value that actually occurred and a reviewer can find it in the Datadog UI. + +Each row carries `points` and `nulls`. **Always check them.** Datadog emits nulls for gaps, +and a row with `p95: null` plus a `note` means *no data*, which is not the same as *low +usage* -- never treat an empty series as an idle workload. + +### Reconciliation vs Sizing +These are two different questions and must not be conflated: +- **Reconciliation** -- "does our view match the team's?" Read the org's own monitor + definitions with `resource_type='monitors'` to get their real `query` strings and + `options.thresholds`, then reproduce that formula exactly. This tells you which + workloads run hot. It is *not* the basis of any recommended number. +- **Sizing** -- "what should this value be?" Use `resource_type='metric_stats'` for the + usage distribution over the window. Never derive a sizing number from a monitor threshold. ### Datadog Query Syntax - Filter by service: `service:X` @@ -46,6 +86,8 @@ Datadog integration for querying observability data during Root Cause Analysis. ### Examples - Logs: `query_datadog(resource_type='logs', query='service:web status:error', time_from='-1h')` - Metrics: `query_datadog(resource_type='metrics', query='avg:system.cpu.user{*}', time_from='-2h')` +- Metric stats (30-day p95 per deployment, one query for all of them): + `query_datadog(resource_type='metric_stats', query='sum:kubernetes.memory.usage{env:production} by {kube_deployment}', time_from='-30d')` - Traces: `query_datadog(resource_type='traces', query='service:web @http.status_code:500', time_from='-1h')` - Monitors: `query_datadog(resource_type='monitors', query='web')` @@ -71,7 +113,12 @@ Datadog integration for querying observability data during Root Cause Analysis. ## Important Rules - Datadog is a REMOTE service. Use ONLY the `query_datadog` API tool. -- The `resource_type` parameter is required and must be one of: logs, metrics, monitors, events, traces, hosts, incidents. +- The `resource_type` parameter is required and must be one of: logs, metrics, metric_stats, monitors, events, traces, hosts, incidents. - Time parameters accept relative strings (`'-1h'`, `'-24h'`, `'-7d'`) or ISO 8601 timestamps. - The `incidents` resource type requires Datadog Incident Management to be enabled; may return 403 if not. - Results are truncated at the output size limit. Use more specific queries to narrow results. +- Never reason from a truncated result set. If a response carries `truncated`, `truncated_all`, + `series_truncated` or `series_dropped`, narrow the query and re-run before drawing a conclusion. +- Group with `by {tag}` rather than issuing one query per workload -- a single grouped query + returns every deployment at once. A 30-day multi-group query is close to the request timeout, + so query one environment at a time. diff --git a/server/chat/backend/agent/tools/cloud_tools.py b/server/chat/backend/agent/tools/cloud_tools.py index 6727fc22d..e44efbf27 100644 --- a/server/chat/backend/agent/tools/cloud_tools.py +++ b/server/chat/backend/agent/tools/cloud_tools.py @@ -2213,10 +2213,15 @@ def _pinned_trigger(action_id: str = "", _pid=pinned_id, _fn=final_func, **kw): func=final_dd_func, name="query_datadog", description=( - "Query Datadog for logs, metrics, monitors, events, traces, hosts, or incidents. " - "Set resource_type to 'logs', 'metrics', 'monitors', 'events', 'traces', 'hosts', or 'incidents'. " + "Query Datadog for logs, metrics, metric_stats, monitors, events, traces, hosts, or incidents. " + "Set resource_type to 'logs', 'metrics', 'metric_stats', 'monitors', 'events', 'traces', " + "'hosts', or 'incidents'. Use 'metric_stats' for p50/p95/p99/max per series over long " + "windows (capacity and right-sizing questions) -- Datadog cannot compute a time-percentile " + "in a query, so 'metrics' plus a percentile rollup does not work. " "Examples: query_datadog(resource_type='logs', query='service:web status:error', time_from='-1h') " - "or query_datadog(resource_type='metrics', query='avg:system.cpu.user{*}', time_from='-2h')" + "or query_datadog(resource_type='metrics', query='avg:system.cpu.user{*}', time_from='-2h') " + "or query_datadog(resource_type='metric_stats', " + "query='sum:kubernetes.memory.usage{env:production} by {kube_deployment}', time_from='-30d')" ), args_schema=QueryDatadogArgs, )) @@ -2471,6 +2476,34 @@ def _pinned_trigger(action_id: str = "", _pid=pinned_id, _fn=final_func, **kw): except Exception as e: logging.warning(f"Failed to add Notion tools: {e}") + # Add HPA/VPA right-sizing card tools -- needs Slack (to post the card) and + # GitHub (the PR the card links to). Gated on connectors only: State + # .trigger_action_id is never set for background action runs, so there is no + # way today to scope these to the right-sizing action specifically. + # Excluded from PR review, which is read-only, like every other write tool here. + try: + # Imported here rather than relying on the Slack block ~1000 lines above: + # that name is only bound if this function reached that try block without + # raising, so depending on it would fail with UnboundLocalError. + from .slack_tool import is_slack_connected as _is_slack_connected + from .hpa_vpa_card_tool import HPA_VPA_TOOL_SPECS + if (not is_pr_review + and _safe_connected(_is_slack_connected, "Slack") + and _safe_connected(is_github_connected, "GitHub")): + for _func, _name, _schema, _desc in HPA_VPA_TOOL_SPECS: + _ctx = with_user_context(_func) + _notif = with_completion_notification(_ctx) + _final = wrap_func_with_capture(_notif, _name) if tool_capture else _notif + tools.append(StructuredTool.from_function( + func=_final, + name=_name, + description=_desc, + args_schema=_schema, + )) + logging.info(f"Added {len(HPA_VPA_TOOL_SPECS)} HPA/VPA right-sizing tools for user {user_id}") + except Exception as e: + logging.warning(f"Failed to add HPA/VPA right-sizing tools: {e}") + # Add Jira tools if enabled try: from utils.flags.feature_flags import is_jira_enabled diff --git a/server/chat/backend/agent/tools/datadog_tool.py b/server/chat/backend/agent/tools/datadog_tool.py index ed32f97b1..9860b9fc4 100644 --- a/server/chat/backend/agent/tools/datadog_tool.py +++ b/server/chat/backend/agent/tools/datadog_tool.py @@ -2,6 +2,7 @@ import json import logging +import math import re from datetime import datetime, timedelta, timezone from typing import Any, Optional @@ -20,12 +21,15 @@ class QueryDatadogArgs(BaseModel): resource_type: str = Field( - description="Type of data to query: 'logs', 'metrics', 'monitors', 'events', 'traces', 'hosts', or 'incidents'" + description="Type of data to query: 'logs', 'metrics', 'metric_stats', 'monitors', " + "'events', 'traces', 'hosts', or 'incidents'" ) query: str = Field( default="", description="Search query. Syntax varies by resource type. " "Logs: 'service:web status:error'. Metrics: 'avg:system.cpu.user{*}'. " + "Metric_stats: same metric syntax as metrics, grouping encouraged " + "e.g. 'sum:kubernetes.memory.usage{env:production} by {kube_deployment}'. " "Monitors: name filter. Events: source filter. " "Traces: 'service:web @http.status_code:500'. Hosts: host filter. " "Incidents: not used.", @@ -39,6 +43,14 @@ class QueryDatadogArgs(BaseModel): description="End time: 'now' or ISO 8601", ) limit: int = Field(default=100, description="Maximum results to return") + interval: Optional[int] = Field( + default=None, + description="Rollup granularity in MILLISECONDS for 'metrics' and 'metric_stats'. " + "Clamped to 60000..14400000. For 'metric_stats', omit it to auto-pick a " + "granularity that keeps each series under ~1000 points (a 30-day window " + "auto-picks 1h). For 'metrics', omitting it leaves Datadog's own default " + "resolution in place.", + ) # --------------------------------------------------------------------------- @@ -123,7 +135,7 @@ def _truncate_results(results: list, serialized: list[str]) -> tuple[list, bool] # --------------------------------------------------------------------------- -def _query_logs(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_logs(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: start = _to_iso8601(time_from) end = _to_iso8601(time_to) response = client.search_logs(query=query or "*", start=start, end=end, limit=limit) @@ -131,12 +143,18 @@ def _query_logs(client, query: str, time_from: str, time_to: str, limit: int) -> return {"resource_type": "logs", "count": len(logs), "results": logs} -def _query_metrics(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_metrics(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: if not query: raise ValueError("query is required for resource_type='metrics' (e.g., 'avg:system.cpu.user{*}')") start_ms = _to_unix_ms(time_from) end_ms = _to_unix_ms(time_to) - response = client.query_metrics(query=query, start_ms=start_ms, end_ms=end_ms) + # Honour an explicit interval, but do NOT auto-pick one here. Datadog's own + # default resolution is part of this handler's existing output contract, which + # the RCA prompts and SKILL.md depend on; changing it is out of scope for + # right-sizing. metric_stats auto-picks because it is new and owns its own + # contract. + interval = _clamp_interval(kwargs.get("interval")) + response = client.query_metrics(query=query, start_ms=start_ms, end_ms=end_ms, interval=interval) data = response.get("data", {}) attrs = data.get("attributes", {}) if isinstance(data, dict) else {} result_data = { @@ -147,7 +165,306 @@ def _query_metrics(client, query: str, time_from: str, time_to: str, limit: int) return {"resource_type": "metrics", "count": len(result_data["series"]), "results": [result_data]} -def _query_monitors(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +# --------------------------------------------------------------------------- +# metric_stats: server-side percentile summaries +# +# Datadog has NO time-percentile capability. Every documented route is closed: +# .rollup(percentile, 95, 3600) -> 400 "Unrecognized rollup method: percentile" +# .rollup(p95, 3600) -> 400 "Unrecognized rollup method: p95" +# p95:{...} -> 200 but ZERO series (needs distribution metrics) +# formula "p95(a)" -> 400 'function "p95()" does not exist' +# formula "percentile(a,95,3600)"-> 400 (percentile() is a SPACE aggregator; +# args 2 and 3 are group tags, not q/window) +# scalar aggregator "percentile" -> 400 "cannot use the sum aggregator with +# the percentile reducer" +# .rollup() accepts only avg/sum/min/max/count, and both right-sizing metrics +# (kubernetes.cpu.usage.total, kubernetes.memory.usage) are GAUGES, so the +# distribution-only p95: prefix can never apply. Percentiles are therefore +# computed here, in Python, from the raw rolled-up points. +# --------------------------------------------------------------------------- + +# Datadog caps a timeseries at 1500 points. Whole-minute granularities, smallest +# first; we pick the first that keeps a series under _TARGET_POINTS. +_INTERVAL_LADDER_MS = (60_000, 300_000, 600_000, 900_000, 1_800_000, 3_600_000, 7_200_000, 14_400_000) +_MIN_INTERVAL_MS = _INTERVAL_LADDER_MS[0] +_MAX_INTERVAL_MS = _INTERVAL_LADDER_MS[-1] +_TARGET_POINTS = 1000 +# Datadog's hard per-series limit. A window long enough that even the 4h ceiling +# cannot fit under it (beyond ~250 days) gets a partial series back; say so +# rather than letting a partial window read as a complete one. +_DATADOG_POINT_CAP = 1500 + +# Budget against cap_tool_output's PASS_THROUGH_CHARS (40_000), NOT +# MAX_OUTPUT_SIZE (120_000). Anything above 40K is routed through an LLM +# summarizer, which would paraphrase percentiles into plausible fiction -- +# strictly worse than truncation, because the numbers come back looking +# authoritative. ~280 B/row x 100 rows ~= 28 KB. +_MAX_STATS_SERIES = 100 +_STATS_BYTE_BUDGET = 30_000 + +_SIG_DIGITS = 6 + + +def _clamp_interval(interval: Optional[int]) -> Optional[int]: + """Clamp a caller-supplied interval (ms) into the legal range, or None.""" + if interval is None: + return None + try: + value = int(interval) + except (TypeError, ValueError): + return None + return min(max(value, _MIN_INTERVAL_MS), _MAX_INTERVAL_MS) + + +def _pick_interval(span_ms: int, caller_interval: Optional[int] = None) -> int: + """Choose a rollup granularity in ms that keeps a series under ~1000 points. + + A caller value wins (clamped). Otherwise walk the whole-minute ladder and + take the finest granularity that stays under the point target -- a 30-day + window lands on 3_600_000 (1h) for 720 points. + """ + clamped = _clamp_interval(caller_interval) + if clamped is not None: + return clamped + span_ms = max(int(span_ms), 0) + for candidate in _INTERVAL_LADDER_MS: + if span_ms // candidate <= _TARGET_POINTS: + return candidate + return _MAX_INTERVAL_MS + + +def _percentile(sorted_vals: list[float], q: float) -> Optional[float]: + """Nearest-rank percentile: returns a real observed sample. + + Deliberately not statistics.quantiles -- these numbers land in a PR body as + evidence a reviewer will look up in Datadog, so they must be a value that + actually occurred. quantiles() interpolates, raises on n < 2, and its + inclusive/exclusive methods disagree. + """ + if not sorted_vals: + # The max(0, ...) guard cannot rescue an empty list: min(-1, 0) is -1, + # so the index expression would raise IndexError. Callers filter empty + # series out first; this is defence in depth. + return None + idx = min(len(sorted_vals) - 1, max(0, math.ceil(q * len(sorted_vals)) - 1)) + return sorted_vals[idx] + + +def _round_sig(value: Optional[float], digits: int = _SIG_DIGITS) -> Optional[float]: + """Round to N significant digits: keeps rows compact and keeps 10+ digit + integers out of anything the agent might paste into a PR.""" + if value is None: + return None + if not math.isfinite(value) or value == 0: + return 0.0 if value == 0 else None + return round(value, -int(math.floor(math.log10(abs(value)))) + (digits - 1)) + + +def _series_unit(unit: Any) -> Optional[str]: + """Extract a unit name from Datadog's unit field (a 2-list of nullable dicts).""" + if isinstance(unit, str): + return unit + if isinstance(unit, dict): + return unit.get("name") + if isinstance(unit, list): + for entry in unit: + if isinstance(entry, dict) and entry.get("name"): + return entry["name"] + return None + + +def _scope_and_group(group_tags: Any) -> tuple[str, dict]: + """Build a stable scope string plus a k/v dict from a series' group_tags. + + scope is a sorted 'k:v' join so the agent can use it as a dict key across + calls. Bare values (no colon) still appear in scope but cannot be keyed. + """ + tags = [str(t) for t in group_tags if t] if isinstance(group_tags, list) else [] + group: dict[str, str] = {} + for tag in tags: + key, sep, value = tag.partition(":") + if sep: + group[key] = value + return ",".join(sorted(tags)), group + + +def _summarize_series(group_tags: Any, unit: Any, row: Any) -> dict: + """Reduce one series' points to a compact stats row (never per-point arrays).""" + points = row if isinstance(row, list) else [] + # isfinite is not paranoia: a single NaN point breaks list.sort() (NaN + # compares False against everything, so the "sorted" list stays unsorted and + # vals[-1] is no longer the max), and json.dumps emits bare NaN/Infinity, + # which is invalid JSON that the agent's parser rejects. bool is an int + # subclass, so it is excluded explicitly. + vals = [ + float(v) for v in points + if isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v) + ] + scope, group = _scope_and_group(group_tags) + + stats = { + "scope": scope, + "group": group, + "unit": _series_unit(unit), + "points": len(points), + "nulls": len(points) - len(vals), + } + if not vals: + # All-null or empty: report explicitly so the agent can tell "no data" + # from "low usage" instead of recommending a cut on an idle-looking gap. + stats.update({"p50": None, "p95": None, "p99": None, "max": None, "mean": None, + "note": "no non-null points in window"}) + return stats + + vals.sort() + stats.update({ + "p50": _round_sig(_percentile(vals, 0.50)), + "p95": _round_sig(_percentile(vals, 0.95)), + "p99": _round_sig(_percentile(vals, 0.99)), + "max": _round_sig(vals[-1]), + "mean": _round_sig(sum(vals) / len(vals)), + }) + return stats + + +def _point_cap_note(span_ms: int, chosen_interval: int, caller_interval: Optional[int]) -> Optional[str]: + """Warn when a window cannot fit in one series, or None when it fits. + + Datadog clips the series rather than erroring, and a percentile over a + silently-clipped window is not the percentile that was asked for -- so this + has to be said out loud, with advice that matches why it happened. + """ + expected_points = max(span_ms, 0) // chosen_interval + if expected_points <= _DATADOG_POINT_CAP: + return None + + partial = ( + f"This window needs ~{expected_points} points at a {chosen_interval} ms interval, " + f"above Datadog's {_DATADOG_POINT_CAP}-point per-series limit, so the series is " + "partial and these percentiles cover only part of the window. " + ) + if _clamp_interval(caller_interval) is not None and chosen_interval < _MAX_INTERVAL_MS: + # Caller pinned a sub-ceiling interval, so the fix is theirs to make: a + # coarser one (or omitting it to auto-pick) fits the same window. + return partial + "Re-run with a coarser interval, or omit interval to auto-pick one." + # Already at the ceiling, whether pinned there or auto-picked: no coarser + # interval exists, so only a shorter window can help. + return partial + ( + "This is already the coarsest supported interval, so use a window of 250 days or less." + ) + + +def _p95_datadog(client, query: str, time_from: str, time_to: str, limit: int, + interval: Optional[int] = None) -> dict: + """Datadog backend: fetch raw rolled-up points and reduce them in Python.""" + start_ms = _to_unix_ms(time_from) + end_ms = _to_unix_ms(time_to) + chosen_interval = _pick_interval(end_ms - start_ms, interval) + + response = client.query_metrics( + query=query, start_ms=start_ms, end_ms=end_ms, interval=chosen_interval + ) + data = response.get("data", {}) + attrs = data.get("attributes", {}) if isinstance(data, dict) else {} + series = attrs.get("series") or [] + values = attrs.get("values") or [] + + max_series = min(max(limit, 1), _MAX_STATS_SERIES) + rows, used_bytes, truncated = [], 0, False + for idx, entry in enumerate(series[:max_series]): + meta = entry if isinstance(entry, dict) else {} + row = _summarize_series( + meta.get("group_tags"), + meta.get("unit"), + values[idx] if idx < len(values) else [], + ) + row_bytes = len(json.dumps(row)) + if used_bytes + row_bytes > _STATS_BYTE_BUDGET: + truncated = True + break + rows.append(row) + used_bytes += row_bytes + + result = { + "resource_type": "metric_stats", + "metrics_source": "datadog", + "interval_ms": chosen_interval, + "series_returned": len(series), + "series_included": len(rows), + "count": len(rows), + "results": rows, + } + + window_note = _point_cap_note(end_ms - start_ms, chosen_interval, interval) + if window_note: + result["window_exceeds_point_cap"] = True + result["window_note"] = window_note + + if truncated: + result["series_truncated"] = True + result["note"] = ( + f"Only {len(rows)} of {len(series)} series fit the output budget. " + "Narrow the query filter or group by fewer tags and re-run." + ) + elif len(series) > max_series: + result["series_dropped"] = len(series) - max_series + result["note"] = ( + f"Returned {max_series} of {len(series)} series (per-call series cap). " + "Narrow the query filter to see the rest." + ) + return result + + +# Only one backend is implemented today. Adding New Relic, Dynatrace or Coroot +# is a table entry plus a branch that pushes the percentile into the query +# (percentile(x,95) / :percentile(95) / quantile_over_time) and reads back a +# scalar -- the row shape must stay identical so neither the prompt nor the +# Slack card ever learns which provider answered. +_METRIC_STATS_SOURCES = ("datadog",) +_P95_BY_SOURCE = {"datadog": _p95_datadog} + +# Fail fast at import if a source is advertised without a dispatch entry -- +# mirrors the _ALERTS_PATH_BY_SOURCE guard in aurora_mcp/tools_gated.py, but +# raises instead of asserting: `python -O` strips asserts and would silently +# drop the guard. +_uncovered_sources = set(_METRIC_STATS_SOURCES) - set(_P95_BY_SOURCE) +if _uncovered_sources: + raise RuntimeError( + f"metric_stats sources not fully covered by _P95_BY_SOURCE: missing {_uncovered_sources}" + ) + + +def _query_metric_stats(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: + """Percentile summary per series -- one compact row per series, no raw points. + + One row PER SERIES is load-bearing, not cosmetic. _truncate_results keeps a + *prefix* of `results`, so the single fat dict that resource_type='metrics' + returns gets dropped whole: a 563 KB payload becomes count=0, which the + agent cannot distinguish from "this workload has no data" -- and it would + then recommend cutting an idle-looking workload. With N compact rows no + single item can exceed the limit, so that failure mode cannot occur. + """ + if not query: + raise ValueError( + "query is required for resource_type='metric_stats' " + "(e.g., 'sum:kubernetes.memory.usage{env:production} by {kube_deployment}')" + ) + # `source` is deliberately NOT part of QueryDatadogArgs: this tool is the + # Datadog surface, so the agent never selects a backend. It is an internal + # seam for a future caller that multiplexes providers (mirroring + # _resolve_source in aurora_mcp/tools_gated.py), reachable because + # query_datadog forwards **kwargs. + source = (kwargs.get("source") or "datadog").lower().strip() + backend = _P95_BY_SOURCE.get(source) + if backend is None: + raise ValueError( + f"Unsupported metrics source '{source}' for metric_stats. " + f"Supported: {', '.join(sorted(_P95_BY_SOURCE))}" + ) + return backend(client, query, time_from, time_to, limit, interval=kwargs.get("interval")) + + +def _query_monitors(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: params: dict[str, Any] = {"page_size": limit} if query: params["name"] = query @@ -157,7 +474,7 @@ def _query_monitors(client, query: str, time_from: str, time_to: str, limit: int return {"resource_type": "monitors", "count": len(monitors[:limit]), "results": monitors[:limit]} -def _query_events(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_events(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: start_ts = _to_unix_seconds(time_from) end_ts = _to_unix_seconds(time_to) params: dict[str, Any] = {} @@ -168,7 +485,7 @@ def _query_events(client, query: str, time_from: str, time_to: str, limit: int) return {"resource_type": "events", "count": len(events), "results": events} -def _query_traces(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_traces(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: start = _to_iso8601(time_from) end = _to_iso8601(time_to) response = client.search_traces(query=query or "*", start=start, end=end, limit=limit) @@ -176,14 +493,14 @@ def _query_traces(client, query: str, time_from: str, time_to: str, limit: int) return {"resource_type": "traces", "count": len(spans), "results": spans} -def _query_hosts(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_hosts(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: from_ts = _to_unix_seconds(time_from) response = client.list_hosts(query=query, count=limit, from_ts=from_ts) host_list = response.get("host_list", [])[:limit] return {"resource_type": "hosts", "count": len(host_list), "results": host_list} -def _query_incidents(client, query: str, time_from: str, time_to: str, limit: int) -> dict: +def _query_incidents(client, query: str, time_from: str, time_to: str, limit: int, **kwargs) -> dict: response = client.list_incidents(page_size=limit) incidents = response.get("data", [])[:limit] return {"resource_type": "incidents", "count": len(incidents), "results": incidents} @@ -192,6 +509,7 @@ def _query_incidents(client, query: str, time_from: str, time_to: str, limit: in _HANDLERS = { "logs": _query_logs, "metrics": _query_metrics, + "metric_stats": _query_metric_stats, "monitors": _query_monitors, "events": _query_events, "traces": _query_traces, @@ -211,10 +529,11 @@ def query_datadog( time_from: str = "-1h", time_to: str = "now", limit: int = 100, + interval: Optional[int] = None, user_id: Optional[str] = None, **kwargs, ) -> str: - """Query Datadog for logs, metrics, monitors, events, traces, hosts, or incidents.""" + """Query Datadog for logs, metrics, metric_stats, monitors, events, traces, hosts, or incidents.""" if not user_id: return json.dumps({"error": "User context not available"}) @@ -235,7 +554,11 @@ def query_datadog( logger.info("[DATADOG-TOOL] user=%s resource=%s query=%s", user_id, resource_type, query[:100] if query else "") try: - result = handler(client, query, time_from, time_to, limit) + # Forward **kwargs so internal-only handler options (e.g. metric_stats' + # `source` seam) actually reach the handler. `interval` is passed + # explicitly because it is a declared agent-facing argument. + _handler_kwargs = {k: v for k, v in kwargs.items() if k != "interval"} + result = handler(client, query, time_from, time_to, limit, interval=interval, **_handler_kwargs) result["success"] = True result["time_range"] = f"{time_from} to {time_to}" @@ -245,9 +568,28 @@ def query_datadog( if was_truncated: result["results"] = truncated_results result["truncated"] = True - result["note"] = f"Results truncated from {len(results_list)} to {len(truncated_results)} due to size limit." + # Append rather than assign: metric_stats already puts its own + # series_dropped / series_truncated explanation in `note`, and + # overwriting it would hide the fact that series were dropped too. + _prior_note = result.get("note") + _note = f"Results truncated from {len(results_list)} to {len(truncated_results)} due to size limit." + result["note"] = f"{_prior_note} {_note}" if _prior_note else _note result["count"] = len(truncated_results) + if was_truncated and not truncated_results and results_list: + # A single oversized item leaves results empty, which reads as + # count=0 -- indistinguishable from "no data" and a silent wrong + # answer. Say so loudly instead. + result["truncated_all"] = True + # Append, like the branch above: any series_dropped / series_truncated + # context already in `note` is exactly what the agent needs to tell + # "too big" apart from "no data". + _all_note = ("A single result exceeded the output size limit; nothing could be " + "returned. Narrow the window or grouping, or use " + "resource_type='metric_stats' for percentile summaries.") + _existing = result.get("note") + result["note"] = f"{_existing} {_all_note}" if _existing else _all_note + return json.dumps(result) except DatadogAPIError as exc: diff --git a/server/chat/backend/agent/tools/github_rca_tool.py b/server/chat/backend/agent/tools/github_rca_tool.py index 4b55944e0..d6d139861 100644 --- a/server/chat/backend/agent/tools/github_rca_tool.py +++ b/server/chat/backend/agent/tools/github_rca_tool.py @@ -163,11 +163,13 @@ def _call_github_api( ): """Call GitHub REST API for a user/repo using the auth router. - Bypasses the MCP github server (which spawns a Docker container and - therefore cannot run in Aurora's chatbot service, which deliberately - has no docker socket). The auth router transparently picks App or - OAuth credentials per the hybrid mode rules and returns a token that - works with the standard ``Authorization: token `` header. + A direct REST call rather than going through the MCP github server: for + simple reads it avoids the cost of spawning a subprocess per call. (The MCP + server itself is a native Go binary baked into the image and spawned over + stdio -- it does *not* need a Docker socket, and it runs fine in the chatbot + and Celery containers.) The auth router transparently picks App or OAuth + credentials per the hybrid mode rules and returns a token that works with + the standard ``Authorization: token `` header. Returns the parsed JSON body on success (list or dict, matching GitHub's native shape so existing downstream parsers keep working), diff --git a/server/chat/backend/agent/tools/hpa_vpa_card_tool.py b/server/chat/backend/agent/tools/hpa_vpa_card_tool.py new file mode 100644 index 000000000..08ae5d665 --- /dev/null +++ b/server/chat/backend/agent/tools/hpa_vpa_card_tool.py @@ -0,0 +1,582 @@ +"""Slack card for HPA/VPA right-sizing recommendations. + +Deliberately not in ``slack_tool.py``: that module is the read-only Slack +surface. This is the **first agent-callable Slack write tool** in the repo, so +there is no prior art to copy -- the shape here (claim row -> post -> attach ts, +with a compensating delete) is new. + +Card construction lives in one builder taking a neutral recommendation dict. +Slack and Google Chat button schemas are irreconcilable (``action_id`` string +prefixes vs ``function`` + typed ``parameters``), so keeping the intermediate +representation separate from the renderer makes a Google Chat card an addition +rather than a rewrite of this tool. + +There is intentionally no ``tool_registry.py`` entry and no ``gate_action`` +call. See the module docstring notes on each in ``send_hpa_vpa_recommendation``. +""" + +import json +import logging +import math +import re +from typing import Optional + +from pydantic import BaseModel, Field + +from routes.slack.slack_events_helpers import SLACK_MAX_SECTION_TEXT, validate_slack_blocks +from services.actions import hpa_vpa_recommendations as recs +from utils.auth.stateless_auth import set_rls_context +from utils.db.connection_pool import db_pool +from utils.log_sanitizer import sanitize + +logger = logging.getLogger(__name__) + +# The card must not become the leak path for the prompt's numeric guardrail: +# a raw nanocore/byte integer pasted into Slack is unreadable and tends to end +# up copied into a PR body. +_LONG_INT_RE = re.compile(r"\d{10,}") +_MAX_DISPLAY_CHARS = 200 +_REPO_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$") + +# Column widths from the hpa_vpa_recommendations DDL. Enforced here because an +# over-long value is not a display problem: the INSERT raises "value too long", +# the whole tool call fails with a generic error, and no card is ever posted. +_COLUMN_LIMITS = { + "workload": 255, + "service": 255, + "environment": 128, + "autoscaler": 64, + "metrics_source": 32, + "vcs_provider": 32, +} + +_DIMENSIONS = ( + ("memory", "Memory request"), + ("cpu", "CPU request"), + ("max_replicas", "HPA maxReplicas"), +) + + +class SendHpaVpaRecommendationArgs(BaseModel): + workload: str = Field(description="Workload / deployment name, e.g. 'apiv2worker'") + repo: str = Field(description="Repository holding the IaC change, as 'owner/repo'") + pr_number: int = Field(description="Number of the already-opened right-sizing PR") + pr_url: str = Field(description="Web URL of the PR, used for the View PR button") + service: Optional[str] = Field(default=None, description="Service name if different from the workload") + environment: Optional[str] = Field( + default=None, + description="Environment the workload runs in, e.g. 'production'. Part of the " + "dedup key: the same workload in two environments is tracked separately.", + ) + autoscaler: Optional[str] = Field( + default=None, + description="Autoscaler in play, e.g. 'HPA (CPU)', 'KEDA (queue depth)', 'none'. " + "Shown on the card so a reviewer understands a partial recommendation.", + ) + memory_current: Optional[str] = Field(default=None, description="Current memory request as display text, e.g. '2 Gi'") + memory_recommended: Optional[str] = Field(default=None, description="Recommended memory request, e.g. '768 Mi'") + memory_evidence: Optional[str] = Field( + default=None, description="Short evidence, e.g. '30-day p95 usage 512 Mi -- 25% of request'" + ) + cpu_current: Optional[str] = Field(default=None, description="Current CPU request as display text, e.g. '2000 m'") + cpu_recommended: Optional[str] = Field(default=None, description="Recommended CPU request, e.g. '750 m'") + cpu_evidence: Optional[str] = Field(default=None, description="Short evidence, e.g. 'p95 480 m, no throttling'") + max_replicas_current: Optional[str] = Field(default=None, description="Current HPA maxReplicas, e.g. '10'") + max_replicas_recommended: Optional[str] = Field(default=None, description="Recommended maxReplicas, e.g. '6'") + max_replicas_evidence: Optional[str] = Field(default=None, description="Short evidence, e.g. 'never exceeded 4 in 30 d'") + reviewer: Optional[str] = Field( + default=None, + description="Slack user ID of the reviewer (must be a real ID like 'U0966GURFUK'). " + "Omit if unknown -- a fabricated ID renders as a blank grey box in Slack.", + ) + severity_score: Optional[float] = Field( + default=None, + description="Max relative mis-size across dimensions: abs(recommended - current) / current. " + "Used to decide whether a re-proposal during a cooldown is materially worse.", + ) + metrics_source: Optional[str] = Field( + default=None, description="Provider the usage percentiles came from, e.g. 'datadog'" + ) + vcs_provider: Optional[str] = Field( + default=None, + description="VCS hosting the PR. Only 'github' is supported today (the default); " + "the column exists so GitLab and Bitbucket become a fill-in rather than a migration.", + ) + + +class ListHpaVpaRecommendationsArgs(BaseModel): + """No arguments -- returns every live and cooling recommendation for the org.""" + + +# --------------------------------------------------------------------------- +# Validation +# --------------------------------------------------------------------------- + + +def _check_display(value: Optional[str], label: str) -> Optional[str]: + """Return an error string if a display value is unusable, else None.""" + if value is None: + return None + if len(value) > _MAX_DISPLAY_CHARS: + return f"{label} is too long ({len(value)} chars, max {_MAX_DISPLAY_CHARS})" + if _LONG_INT_RE.search(value): + return (f"{label} contains a raw 10+ digit number. Convert to human units " + f"(e.g. '768 Mi', '750 m') before sending the card.") + return None + + +def _validate(args: dict) -> Optional[str]: + """Validate card arguments. Returns an error message, or None when valid.""" + workload = (args.get("workload") or "").strip() + if not workload: + return "workload is required" + + repo = (args.get("repo") or "").strip() + if not _REPO_RE.match(repo): + return f"repo must be in 'owner/repo' form, got '{repo}'" + + try: + pr_number = int(args.get("pr_number") or 0) + except (TypeError, ValueError): + return "pr_number must be an integer" + if pr_number <= 0: + return "pr_number must be a positive integer" + + pr_url = (args.get("pr_url") or "").strip() + if not pr_url: + return "pr_url is required so the View PR button has a destination" + # Slack rejects a button whose url is not http(s), which fails the whole + # chat.postMessage call -- and a javascript:/data: url would be a live XSS + # vector in any other renderer this intermediate dict later feeds. HTTPS + # only: every supported VCS serves PRs over TLS, so plain http here is + # either a typo or a downgrade. + if not pr_url.lower().startswith("https://"): + return f"pr_url must be an https:// URL, got '{pr_url[:60]}'" + + # Reject a provider we cannot close a PR on, before a row is claimed and a + # card is posted. Otherwise Dismiss would fail on every click, leaving the + # PR open with no way to act on the card. + provider = (args.get("vcs_provider") or "github").lower().strip() + supported = recs.supported_vcs_providers() + if provider not in supported: + return (f"vcs_provider '{provider}' is not supported yet. " + f"Supported: {', '.join(sorted(supported))}.") + + present = [key for key, _ in _DIMENSIONS + if args.get(f"{key}_current") and args.get(f"{key}_recommended")] + if not present: + return ("No recommendation to report: at least one of memory, cpu or max_replicas " + "needs both a current and a recommended value.") + + for key, label in _DIMENSIONS: + for suffix in ("current", "recommended", "evidence"): + err = _check_display(args.get(f"{key}_{suffix}"), f"{label} {suffix}") + if err: + return err + for field in ("workload", "service", "environment", "autoscaler"): + err = _check_display(args.get(field), field) + if err: + return err + + for field, width in _COLUMN_LIMITS.items(): + value = args.get(field) + if isinstance(value, str) and len(value.strip()) > width: + return f"{field} is too long ({len(value.strip())} chars, max {width})" + return None + + +# --------------------------------------------------------------------------- +# Card construction (neutral dict in, Slack blocks out) +# --------------------------------------------------------------------------- + + +def _mention(slack_user_id: Optional[str]) -> Optional[str]: + """Render a real Slack mention, or None. + + Only ``<@U...>`` renders as a mention; a fabricated id renders as a blank + grey box, so an unresolvable reviewer falls back to plain text instead. + """ + value = (slack_user_id or "").strip() + if re.fullmatch(r"[UW][A-Z0-9]{6,}", value): + return f"<@{value}>" + return None + + +def _build_body(rec: dict, status_line: str) -> str: + """Body section text: one line per recommended dimension, present data only.""" + lines = [] + if rec.get("service"): + lines.append(f"*Service:* {rec['service']}") + if rec.get("environment"): + lines.append(f"*Environment:* {rec['environment']}") + + for key, label in _DIMENSIONS: + current = rec.get(f"{key}_current") + recommended = rec.get(f"{key}_recommended") + if not (current and recommended): + continue + line = f"*{label}:* {current} -> *{recommended}*" + evidence = rec.get(f"{key}_evidence") + if evidence: + line += f" ({evidence})" + lines.append(line) + + if rec.get("autoscaler"): + lines.append(f"*Autoscaler:* {rec['autoscaler']}") + if status_line: + lines.append(status_line) + + body = "\n".join(lines) + if len(body) > SLACK_MAX_SECTION_TEXT: + body = body[: SLACK_MAX_SECTION_TEXT - 3] + "..." + return body + + +def build_recommendation_blocks(rec: dict, rec_id: str, *, with_actions: bool = True, + status_line: Optional[str] = None) -> list: + """Build the card from a neutral recommendation dict. + + ``with_actions=False`` drops the buttons entirely, which is how the + post-dismiss rewrite guarantees a card cannot be re-clicked. + """ + pr_number = rec.get("pr_number") + if status_line is None: + reviewer = _mention(rec.get("reviewer")) + status_line = f"*Status:* PR #{pr_number} open -- awaiting review" + if reviewer: + status_line += f" by {reviewer}" + + blocks = [ + # Slack groups consecutive messages from the same app, which visually + # fuses cards into whatever preceded them. A leading divider makes each + # card read as its own unit (build_suggestions_blocks does the same). + {"type": "divider"}, + {"type": "header", "text": {"type": "plain_text", "text": "Right-Sizing Recommendation"}}, + {"type": "section", "text": {"type": "mrkdwn", + "text": f"_Recommendation for_ `{rec.get('workload', 'unknown')}`"}}, + ] + + if with_actions: + # An actions block, not a section accessory: Slack's section `accessory` + # takes a single element object, not an array, so the mockup's two + # flush-right buttons are not expressible. This keeps the pair together. + blocks.append({ + "type": "actions", + "elements": [ + {"type": "button", "text": {"type": "plain_text", "text": f"View PR #{pr_number}"}, + "url": rec.get("pr_url"), "style": "primary", + "action_id": f"hpa_vpa_view_pr_{rec_id}"}, + {"type": "button", "text": {"type": "plain_text", "text": "Dismiss"}, + "value": rec_id, "action_id": f"hpa_vpa_dismiss_{rec_id}"}, + ], + }) + + blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": _build_body(rec, status_line)}}) + blocks.append({"type": "context", "elements": [ + {"type": "mrkdwn", "text": "Nothing is applied until the PR is approved and merged by your team."} + ]}) + return blocks + + +def _fallback_text(rec: dict) -> str: + return f"Right-Sizing Recommendation: {rec.get('workload', 'unknown')} (PR #{rec.get('pr_number')})" + + +def _recommendation_payload(args: dict) -> dict: + """Per-dimension {current, recommended, evidence} for the JSONB column.""" + payload = {} + for key, _label in _DIMENSIONS: + current = args.get(f"{key}_current") + recommended = args.get(f"{key}_recommended") + if current and recommended: + payload[key] = { + "current": current, + "recommended": recommended, + "evidence": args.get(f"{key}_evidence"), + } + return payload + + +# --------------------------------------------------------------------------- +# Tools +# --------------------------------------------------------------------------- + + +def send_hpa_vpa_recommendation(user_id: Optional[str] = None, **kwargs) -> str: + """Post (or update) the Slack card for one right-sized workload. + + Ordering matters. The cooldown is checked and the row claimed *before* the + card goes out, and the caller is expected to have checked + ``list_hpa_vpa_recommendations`` before opening the PR -- otherwise a + suppressed workload still gets a PR nobody asked for. + + Returns a JSON string. Never raises: a failure here must not abort the + agent loop mid-run. + """ + if not user_id: + return json.dumps({"error": "User context not available"}) + + error = _validate(kwargs) + if error: + return json.dumps({"error": error}) + + workload = kwargs["workload"].strip() + environment = (kwargs.get("environment") or "").strip() or None + workload_key = recs.build_workload_key(workload, environment) + repo_full_name = kwargs["repo"].strip() + pr_number = int(kwargs["pr_number"]) + vcs_provider = (kwargs.get("vcs_provider") or "github").lower().strip() + payload = _recommendation_payload(kwargs) + severity = kwargs.get("severity_score") + if severity is not None: + try: + severity = float(severity) + except (TypeError, ValueError): + severity = None + # A NaN/inf severity would clear is_materially_worse's isinstance check and + # then compare as >= anything (inf) or as never-worse (NaN), i.e. the LLM + # could break any cooldown by passing Infinity. Treat it as unknown. + if severity is not None and not math.isfinite(severity): + logger.warning("[HpaVpaCard] Discarding non-finite severity_score for %s", sanitize(workload_key)) + severity = None + + rec = { + "workload": workload, + "service": (kwargs.get("service") or "").strip() or None, + "environment": environment, + "autoscaler": (kwargs.get("autoscaler") or "").strip() or None, + "pr_number": pr_number, + "pr_url": kwargs["pr_url"].strip(), + "reviewer": kwargs.get("reviewer"), + **{f"{k}_{s}": kwargs.get(f"{k}_{s}") + for k, _ in _DIMENSIONS for s in ("current", "recommended", "evidence")}, + } + + # Resolve Slack up-front, OUTSIDE the DB block. Both helpers take their own + # pool connection internally, so calling them while this function holds one + # would nest two checkouts per invocation and can deadlock the pool under + # concurrency. Doing it first also fails fast before a row is claimed. + slack = _resolve_slack_target(user_id) + if slack.get("error"): + return json.dumps({"error": slack["error"]}) + + try: + with db_pool.get_admin_connection() as conn: + with conn.cursor() as cur: + org_id = set_rls_context(cur, conn, user_id, log_prefix="[HpaVpaCard]") + if not org_id: + return json.dumps({"error": "Could not resolve organization context"}) + + recs.lock_workload(cur, org_id, workload_key) + + cooldown = recs.get_active_cooldown(cur, org_id, workload_key) + if cooldown: + if not recs.is_materially_worse(severity, cooldown.get("severity_score")): + conn.commit() + logger.info( + "[HpaVpaCard] Suppressed %s for org=%s (cooldown until %s)", + sanitize(workload_key), sanitize(org_id), cooldown.get("cooldown_until"), + ) + return json.dumps({ + "status": "suppressed", + "reason": "cooldown", + "workload": workload, + "cooldown_until": cooldown.get("cooldown_until"), + "dismissed_severity_score": cooldown.get("severity_score"), + "message": ( + "A human dismissed this workload and the cooldown is still " + "running, and the mis-size has not materially worsened. No card " + "was posted. This is the anti-nag rule working -- record it in " + "the living document and move on. Do not work around it." + ), + }) + # Materially worse: retire the dismissal and fall through. + recs.mark_superseded(cur, cooldown["id"]) + logger.info( + "[HpaVpaCard] Cooldown superseded for %s (severity %s -> %s)", + sanitize(workload_key), cooldown.get("severity_score"), severity, + ) + + live = recs.get_live_recommendation(cur, org_id, workload_key) + if live: + return _update_existing(conn, cur, slack, live, rec, payload, severity, + repo_full_name, pr_number, kwargs) + + rec_id = recs.claim_recommendation( + cur, org_id, user_id, + workload_key=workload_key, workload=workload, environment=environment, + service=rec["service"], autoscaler=rec["autoscaler"], + metrics_source=(kwargs.get("metrics_source") or None), + vcs_provider=vcs_provider, repo_full_name=repo_full_name, + pr_number=pr_number, pr_url=rec["pr_url"], + recommendation=payload, severity_score=severity, + ) + conn.commit() + + return _post_new(conn, cur, slack, rec_id, rec) + except Exception: + logger.exception("[HpaVpaCard] Failed to send recommendation for user=%s", sanitize(user_id)) + return json.dumps({"error": "Could not post the right-sizing recommendation card"}) + + +def _resolve_slack_target(user_id: str) -> dict: + """Resolve the Slack client and destination channel for a user. + + Called before any DB connection is checked out: both helpers below take + their own pool connection, so resolving them inside an open transaction + would hold two checkouts at once. + """ + from connectors.slack_connector.client import get_slack_client_for_user + from utils.notifications.slack_notification_service import _get_incidents_channel_id + + try: + client = get_slack_client_for_user(user_id) + if not client: + return {"error": "Slack is not connected for this user"} + channel_id = _get_incidents_channel_id(user_id, client) + if not channel_id: + return {"error": "No Slack incidents channel is configured"} + return {"client": client, "channel_id": channel_id} + except Exception as exc: + logger.exception("[HpaVpaCard] Could not resolve Slack target for user=%s", sanitize(user_id)) + return {"error": f"Could not resolve Slack destination: {type(exc).__name__}"} + + +def _post_new(conn, cur, slack: dict, rec_id: str, rec: dict) -> str: + """Post a fresh card and attach its ts, releasing the claim on failure.""" + client, channel_id = slack["client"], slack["channel_id"] + + try: + blocks = build_recommendation_blocks(rec, rec_id) + text = _fallback_text(rec) + if not validate_slack_blocks(blocks): + logger.error("[HpaVpaCard] Block validation failed; posting plain text for %s", sanitize(rec["workload"])) + blocks = None + + response = client.send_message(channel=channel_id, text=text, blocks=blocks) + message_ts = (response or {}).get("ts") + if not message_ts: + raise ValueError("Slack did not return a message timestamp") + except Exception as exc: + # Compensate: a transient Slack outage must not permanently block this + # workload behind the partial unique index. + try: + recs.delete_recommendation(cur, rec_id) + conn.commit() + except Exception: + logger.exception("[HpaVpaCard] Failed to release claimed row %s", rec_id) + logger.exception("[HpaVpaCard] Slack post failed for %s", sanitize(rec["workload"])) + return json.dumps({"error": f"Could not post the Slack card: {type(exc).__name__}: {str(exc)[:150]}"}) + + recs.attach_slack_message(cur, rec_id, channel_id, message_ts) + conn.commit() + logger.info("[HpaVpaCard] Posted card for %s (rec=%s)", sanitize(rec["workload"]), rec_id) + return json.dumps({ + "status": "posted", + "recommendation_id": rec_id, + "workload": rec["workload"], + "channel": channel_id, + "pr_number": rec["pr_number"], + }) + + +def _update_existing(conn, cur, slack: dict, live: dict, rec: dict, payload: dict, + severity: Optional[float], repo_full_name: str, pr_number: int, + kwargs: dict) -> str: + """Refresh an open recommendation in place -- one card per workload, ever.""" + rec_id = live["id"] + recs.refresh_recommendation( + cur, rec_id, recommendation=payload, severity_score=severity, + repo_full_name=repo_full_name, pr_number=pr_number, pr_url=rec["pr_url"], + autoscaler=rec["autoscaler"], metrics_source=(kwargs.get("metrics_source") or None), + ) + conn.commit() + + blocks = build_recommendation_blocks(rec, rec_id) + text = _fallback_text(rec) + if not validate_slack_blocks(blocks): + blocks = None + + channel_id, message_ts = live.get("slack_channel_id"), live.get("slack_message_ts") + client = slack["client"] + try: + if channel_id and message_ts: + client.update_message(channel=channel_id, ts=message_ts, text=text, blocks=blocks) + logger.info("[HpaVpaCard] Updated card in place for %s (rec=%s)", sanitize(rec["workload"]), rec_id) + return json.dumps({ + "status": "updated", + "recommendation_id": rec_id, + "workload": rec["workload"], + "message": "Existing card updated in place with fresh numbers -- no duplicate posted.", + }) + except Exception: + # Message deleted, or a stale ts. Fall through and post a replacement. + logger.warning("[HpaVpaCard] update_message failed for rec=%s; reposting", rec_id) + + return _post_new_for_existing(conn, cur, slack, rec_id, rec) + + +def _post_new_for_existing(conn, cur, slack: dict, rec_id: str, rec: dict) -> str: + """Repost a card for a row that already exists, overwriting its ts. + + Distinct from :func:`_post_new` only in that a failure must NOT delete the + row -- the recommendation predates this post attempt. + """ + client, channel_id = slack["client"], slack["channel_id"] + + try: + blocks = build_recommendation_blocks(rec, rec_id) + if not validate_slack_blocks(blocks): + blocks = None + response = client.send_message(channel=channel_id, text=_fallback_text(rec), blocks=blocks) + message_ts = (response or {}).get("ts") + if not message_ts: + raise ValueError("Slack did not return a message timestamp") + except Exception as exc: + logger.exception("[HpaVpaCard] Repost failed for rec=%s", rec_id) + return json.dumps({"error": f"Could not post the Slack card: {type(exc).__name__}: {str(exc)[:150]}"}) + + recs.attach_slack_message(cur, rec_id, channel_id, message_ts) + conn.commit() + return json.dumps({ + "status": "reposted", + "recommendation_id": rec_id, + "workload": rec["workload"], + "message": "The previous card was gone, so a replacement was posted.", + }) + + +def list_hpa_vpa_recommendations(user_id: Optional[str] = None, **kwargs) -> str: + """List open right-sizing recommendations and workloads still in cooldown. + + Call this BEFORE opening any PR. Checking after the fact produces the worst + sequence: open PR -> post card -> suppressed, leaving a PR nobody wanted. + """ + if not user_id: + return json.dumps({"error": "User context not available"}) + return json.dumps(recs.list_recommendations(user_id)) + + +HPA_VPA_TOOL_SPECS = [ + ( + send_hpa_vpa_recommendation, + "send_hpa_vpa_recommendation", + SendHpaVpaRecommendationArgs, + "Post a Slack card to the incidents channel for ONE workload you have already opened a " + "right-sizing PR for. The card carries View PR and Dismiss buttons. Pass current and " + "recommended values as display strings in human units ('2 Gi' -> '768 Mi', '2000 m' -> " + "'750 m') -- never raw byte or nanocore integers. Omit any dimension you are not " + "recommending a change to; the card degrades cleanly. Returns status 'posted', 'updated', " + "or 'suppressed' (a human dismissed this workload and the cooldown is still running -- " + "respect it, do not work around it).", + ), + ( + list_hpa_vpa_recommendations, + "list_hpa_vpa_recommendations", + ListHpaVpaRecommendationsArgs, + "List right-sizing recommendations already open, plus workloads a human dismissed that " + "are still inside their cooldown window. Call this BEFORE opening any right-sizing PR so " + "you update an existing proposal instead of opening a duplicate, and skip workloads that " + "are suppressed.", + ), +] diff --git a/server/routes/slack/slack_events.py b/server/routes/slack/slack_events.py index 6ba2bdb47..4d5820267 100644 --- a/server/routes/slack/slack_events.py +++ b/server/routes/slack/slack_events.py @@ -31,6 +31,24 @@ # Get frontend URL from environment FRONTEND_URL = os.getenv("FRONTEND_URL") +# action_id prefix for the Dismiss button on right-sizing recommendation cards. +# The bare recommendation UUID also travels in the button's `value`; the +# action_id is what the dispatcher matches on. +_HPA_VPA_DISMISS_PREFIX = "hpa_vpa_dismiss_" + +# Slack expects an interaction response within ~3s. The dismissal transition is +# already committed before this call, so a timeout only costs the automatic PR +# close (surfaced on the card as "could not be closed automatically"), never the +# cooldown. Kept short for that reason; moving the close and card rewrite onto a +# Celery task would remove the constraint entirely and is the better long-term fix. +_GITHUB_CLOSE_TIMEOUT = 5 + +# The "View PR" button carries a `url`, but Slack still delivers a block_actions +# interaction for url buttons and expects an acknowledgement. It must be matched +# and acked with an EMPTY body: any non-empty `text` in a block_actions response +# replaces the original message, which would wipe the card the user just opened. +_HPA_VPA_VIEW_PR_PREFIX = "hpa_vpa_view_pr_" + @slack_events_bp.route("/events", methods=["POST"]) def slack_events(): @@ -276,7 +294,22 @@ def slack_interactions(): team_id=team_id, channel_id=channel_id ) - + + # Handle "Dismiss" on a right-sizing recommendation card + if action_id.startswith(_HPA_VPA_DISMISS_PREFIX): + return _handle_hpa_vpa_dismiss( + payload=payload, + action=action, + slack_user_id=slack_user_id, + team_id=team_id, + channel_id=channel_id + ) + + # "View PR" is a url button: ack with an empty body so Slack does not + # replace the card with the default "Interaction received" text. + if action_id.startswith(_HPA_VPA_VIEW_PR_PREFIX): + return jsonify({"text": ""}), 200 + # Default response for unhandled interactions return jsonify({"text": "Interaction received"}), 200 @@ -556,4 +589,207 @@ def _handle_suggestion_details(payload: dict, action: dict, slack_user_id: str, except Exception as e: logger.error(f"Error handling suggestion_details action: {e}", exc_info=True) - return jsonify({"text": ""}), 200 \ No newline at end of file + return jsonify({"text": ""}), 200 + + +def _send_ephemeral(client, channel_id: str, slack_user_id: str, text: str) -> None: + """Best-effort ephemeral reply. Never raises into an interaction response.""" + try: + client._make_request( + "POST", + "chat.postEphemeral", + {"channel": channel_id, "user": slack_user_id, "text": text}, + ) + except Exception: + logger.exception( + "Failed to send ephemeral message to Slack user %s", sanitize(slack_user_id) + ) + + +def _handle_hpa_vpa_dismiss(payload: dict, action: dict, slack_user_id: str, team_id: str, channel_id: str) -> tuple: + """Handle "Dismiss" on a right-sizing recommendation card. + + Dismissing means: close the PR and do not raise this workload again for + HPA_VPA_COOLDOWN_DAYS, unless the mis-size materially worsens. + + Ordering is deliberate -- DB transition first, then the PR close. A GitHub + failure therefore leaves the rec dismissed with the PR still open, which is + the safer direction: we never nag about a workload a human rejected, and a + stale open PR is visible and closable by hand. GitHub-first would risk + closing the PR and then losing the cooldown, i.e. exactly the nagging this + feature exists to prevent. The failure is surfaced on the card so it is + never silent. + + Always returns 200: Slack retries non-200 responses, which would turn one + failure into a retry storm. + """ + from services.actions.hpa_vpa_recommendations import ( + HPA_VPA_COOLDOWN_DAYS, + close_pull_request, + dismiss_recommendation, + ) + from utils.validation import is_valid_uuid + + try: + # 1. AUTHENTICATE: who clicked? + clicker_user_id = get_user_id_from_slack_user(slack_user_id, team_id) + if not clicker_user_id: + logger.warning( + f"Unauthenticated Slack user {sanitize(slack_user_id)} (team {sanitize(team_id)}) tried to dismiss a recommendation" + ) + workspace_user_id = get_user_id_from_slack_team(team_id) + if workspace_user_id: + try: + workspace_token_data = get_user_token_data(workspace_user_id, "slack") + if workspace_token_data: + _send_ephemeral( + SlackClient(workspace_token_data.get('access_token')), + channel_id, slack_user_id, + f"WARNING: You're not authenticated in Aurora.\n\nTo dismiss recommendations, connect your Aurora account:\n{FRONTEND_URL}/settings/integrations\n\nClick 'Connect' for Slack and authorize this workspace.", + ) + except Exception: + logger.exception("Failed to warn unauthenticated Slack user %s", sanitize(slack_user_id)) + return jsonify({"text": ""}), 200 + + # 2. PARSE + VALIDATE the recommendation id before any SQL. + action_id = action.get('action_id', '') or '' + rec_id = (action.get('value') or action_id[len(_HPA_VPA_DISMISS_PREFIX):]).strip() + if not is_valid_uuid(rec_id): + # Empty body: a non-empty `text` in a block_actions response REPLACES + # the original message, so returning an error string here would wipe + # the card instead of reporting a problem with it. + logger.error(f"Invalid recommendation id on dismiss action: {sanitize(rec_id)}") + return jsonify({"text": ""}), 200 + + client = get_slack_client_for_user(clicker_user_id) + + # 3. TRANSITION: atomic and idempotent. The status='proposed' predicate + # *is* the double-click defence -- a second click matches no rows. + with db_pool.get_admin_connection() as conn: + with conn.cursor() as cursor: + clicker_org_id = set_rls_context( + cursor, conn, clicker_user_id, log_prefix="[SlackEvents:hpa_vpa_dismiss]" + ) + if not clicker_org_id: + logger.error(f"Could not resolve org for Slack clicker {sanitize(slack_user_id)}") + return jsonify({"text": ""}), 200 + + # 4. AUTHORIZE: RLS already scopes reads to the clicker's org; + # this exists to give the right message instead of a + # confusing "not found", and as defence in depth. + cursor.execute( + "SELECT org_id, status FROM hpa_vpa_recommendations WHERE id = %s::uuid", + (rec_id,), + ) + row = cursor.fetchone() + if not row or row[0] != clicker_org_id: + logger.warning( + f"Slack user {sanitize(slack_user_id)} tried to dismiss recommendation outside their org" + ) + if client: + _send_ephemeral(client, channel_id, slack_user_id, + "Unauthorized: that recommendation does not belong to your organization.") + return jsonify({"text": ""}), 200 + + dismissed = dismiss_recommendation(cursor, rec_id, clicker_org_id, slack_user_id) + # Commit before the GitHub call so a slow API cannot hold the row lock. + conn.commit() + + if not dismissed: + logger.info(f"Recommendation {rec_id} was already dismissed; no-op") + if client: + _send_ephemeral(client, channel_id, slack_user_id, + "That recommendation was already dismissed.") + return jsonify({"text": ""}), 200 + + # 5. CLOSE the PR on whichever VCS hosts it. Close as the account that + # opened it -- the clicker is a teammate in the same org who may have + # no GitHub credential of their own, which would fail every dismissal + # except the opener's own. + pr_number = dismissed.get("pr_number") + close_as_user_id = dismissed.get("user_id") or clicker_user_id + close_result = close_pull_request( + close_as_user_id, + dismissed.get("vcs_provider") or "github", + dismissed.get("repo_full_name") or "", + pr_number, + timeout=_GITHUB_CLOSE_TIMEOUT, + ) if pr_number else {"error": "No PR recorded for this recommendation"} + + if close_result.get("already_merged"): + # A merge is acceptance, not rejection: it must not start an + # anti-nag window, or the next genuine drift goes unreported. + try: + with db_pool.get_admin_connection() as conn: + with conn.cursor() as cursor: + set_rls_context(cursor, conn, clicker_user_id, + log_prefix="[SlackEvents:hpa_vpa_dismiss]") + from services.actions.hpa_vpa_recommendations import mark_merged + mark_merged(cursor, rec_id, clicker_org_id) + conn.commit() + except Exception: + logger.exception("Failed to mark recommendation %s merged", rec_id) + status_line = (f"*Status:* PR #{pr_number} was already merged -- the change was accepted. " + "No cooldown applied.") + elif close_result.get("success"): + status_line = (f"*Status:* Dismissed by <@{slack_user_id}> -- PR #{pr_number} closed. " + f"Aurora will not raise this workload again for {HPA_VPA_COOLDOWN_DAYS} days.") + else: + logger.error( + f"Failed to close PR #{pr_number} for recommendation {rec_id}: {close_result.get('error')}" + ) + status_line = (f"*Status:* Dismissed by <@{slack_user_id}> -- PR #{pr_number} could not be " + "closed automatically; please close it on GitHub.") + + # 6. REWRITE the card with no buttons at all. Nothing to re-click is the + # strongest idempotency guarantee, layered on the conditional UPDATE. + _rewrite_dismissed_card(client, payload, dismissed, rec_id, channel_id, status_line) + + return jsonify({"text": ""}), 200 + + except Exception: + logger.exception("Error handling hpa_vpa_dismiss action") + return jsonify({"text": ""}), 200 + + +def _rewrite_dismissed_card(client, payload: dict, dismissed: dict, rec_id: str, + channel_id: str, status_line: str) -> None: + """Replace the card with a button-free dismissed version.""" + if not client: + return + try: + from chat.backend.agent.tools.hpa_vpa_card_tool import build_recommendation_blocks + + stored = dismissed.get("recommendation") or {} + rec = { + "workload": dismissed.get("workload"), + "service": dismissed.get("service"), + "environment": dismissed.get("environment"), + "autoscaler": dismissed.get("autoscaler"), + "pr_number": dismissed.get("pr_number"), + "pr_url": dismissed.get("pr_url"), + } + for dimension, spec in stored.items(): + if isinstance(spec, dict): + rec[f"{dimension}_current"] = spec.get("current") + rec[f"{dimension}_recommended"] = spec.get("recommended") + rec[f"{dimension}_evidence"] = spec.get("evidence") + + blocks = build_recommendation_blocks( + rec, rec_id, with_actions=False, status_line=status_line + ) + # Prefer the row's own channel/ts; fall back to the click payload. + target_channel = dismissed.get("slack_channel_id") or channel_id + target_ts = dismissed.get("slack_message_ts") or (payload.get("message") or {}).get("ts") + if not (target_channel and target_ts): + logger.warning(f"No channel/ts to rewrite dismissed card for recommendation {rec_id}") + return + + client.update_message( + channel=target_channel, + ts=target_ts, + text=f"Right-Sizing Recommendation dismissed: {rec.get('workload')}", + blocks=blocks, + ) + except Exception: + logger.exception("Failed to rewrite dismissed card for recommendation %s", rec_id) \ No newline at end of file diff --git a/server/services/actions/hpa_vpa_action.py b/server/services/actions/hpa_vpa_action.py new file mode 100644 index 000000000..3e7601e57 --- /dev/null +++ b/server/services/actions/hpa_vpa_action.py @@ -0,0 +1,201 @@ +""" +Built-in Right-Sizing Audit Action + +Default instructions for the system action that compares real CPU and memory +usage against the requests, limits, and autoscaler bounds declared in IaC, and +opens one PR per materially mis-sized workload. +""" + +DEFAULT_HPA_VPA_INSTRUCTIONS = """**Step 1: Gather context** + +Understand what the team actually declared before judging any of it. + +- Call get_infrastructure_context to learn the org's services, environments, and clusters. +- Call get_connected_repos (or equivalent) to find connected code repositories. +- Search those repos for the config that sets resource requests, limits, and autoscaler + bounds. This could be in any format: Terraform (.tf), Helm charts and values files, + Kustomize overlays, raw Kubernetes manifests, Pulumi, Jsonnet, or an operator CRD. +- For each workload you find, record: current CPU request and limit, current memory request + and limit, the autoscaler kind (HPA, KEDA, VPA, none), minReplicas and maxReplicas, the + environment, and the exact file path and line the value lives on. +- Note which metrics provider and which VCS provider are connected, and use the matching + tools for each. Match the repo's own conventions rather than imposing a layout. + +If you cannot find any repo declaring resource requests, update the living document +explaining what you looked for and stop. Do not guess or hallucinate a repo structure. + +**Step 2: Scope the audit** + +A workload is in scope only if all three hold: +- The IaC *actually declares* the value you would change. Never propose a number for a + field the team has not set -- adding a request where there was none is a behaviour change, + not a right-sizing. +- The workload is identifiable in metrics by a stable grouping tag (a deployment or service + label), so usage can be attributed to it unambiguously. +- It ran for the whole measurement window. A workload created last week has a 30-day *gap*, + not a 30-day signal. Say so and exclude it. + +**Step 3: Measure -- two separate computations, never conflated** + +*Reconciliation -- does our view match the team's?* + +Read the org's **own** monitor and alert definitions from whichever monitoring provider is +connected (for Datadog, `resource_type='monitors'`). Take their real query strings and +thresholds from those definitions -- **do not assume a formula**. Then reproduce it exactly: +same aggregation, same grouping, same filters, same evaluation window, same delay. Require +the firing counts to line up with what the team saw. + +Read every part of their definition off the monitor rather than assuming a default, and state +each one you used in the living document: + +- *Shape.* A resource-saturation monitor is usually a **ratio of two summed series** -- + `sum(usage) / sum(limits)`, grouped by the workload tag -- evaluated against a fractional + threshold like `0.90`. Compare like with like: a raw usage figure tested against a + fractional threshold will reconcile with nothing. Whatever shape their monitor actually + uses, mirror it. +- *Grouping.* Group by the same tag they group by (commonly the deployment tag), so a + workload's pods aggregate the way the monitor aggregates them. +- *Scaling.* Apply the same unit scaling the monitor applies -- a CPU ratio built from + nanocore usage needs the `/1e9` the monitor has; a memory ratio in bytes/bytes does not. + A ratio whose numerator and denominator are in different units is silently meaningless. +- *Filters.* Reuse their namespace and tag exclusions verbatim (for example excluding + system namespaces, or ephemeral environments). Dropping an exclusion changes the counts. +- *Window and delay.* Use their evaluation window and apply the same evaluation delay, or + samples will not line up even when the formula is right. +- *Per-environment variation.* Windows, thresholds, and filters commonly differ per + environment. Read each environment's own values; never apply production's to staging. + +If they do not line up, our view of this system disagrees with the team's, and every number +downstream is suspect. Stop, record the mismatch in the living document, and do not open +PRs off numbers you could not corroborate. + +This step tells you *which workloads run hot*. It is **not** the basis of any recommended +number -- never derive a sizing figure from an alert threshold. + +Note the asymmetry and keep it straight: a saturation monitor usually measures usage against +**limits**, while right-sizing changes **requests**. They are different fields with different +consequences -- requests drive scheduling and cost, limits drive throttling and OOM-kills. Do +not read a limits-based utilization figure as a statement about requests, and say which field +each number refers to whenever you report one. + +*Sizing -- what should the value be?* + +Ask for the 30-day p95 of actual usage, summed across pods and grouped by workload (for +Datadog, `resource_type='metric_stats'`, which computes percentiles server-side). Do CPU +and memory as separate queries. Also measure replica count over the same window. + +Aggregate **across all pods of the workload** (`sum`, grouped by the workload tag) -- the +total the deployment consumes, not a hot-pod signal. A max-across-pods reading would size +every replica for the worst one and inflate every recommendation. + +- **Check the unit before doing any arithmetic, and convert to the unit the IaC uses.** + Container CPU usage is typically reported in *nanocores* (Datadog's + `kubernetes.cpu.usage.total` is), so divide by 1e9 to get cores, then express as millicores + to compare against a `500m`-style request. Memory usage is reported in *bytes* and needs no + such scaling -- convert to Mi/Gi only for display. Every row carries a `unit` field: read + it. Comparing a nanocore p95 against a millicore request is a ~1,000,000x error, and it + looks like a plausible number all the way into the PR body. +- Group with `by {tag}` rather than one query per workload; a single grouped query returns + every workload at once. Query one environment at a time to stay inside the request timeout. +- Check `points` and `nulls` on **every** row. A row with a null p95, or one whose points + count falls well short of the window, means *no data* -- which is not the same as *low + usage*. Never treat an empty series as an idle workload. +- Never reason from a truncated result set. If the response carries `truncated`, + `truncated_all`, `series_truncated` or `series_dropped`, narrow the query and re-run. + +**Step 4: Size the recommendation** + +- `new_request = p95 x 1.3` -- a 30% headroom band over sustained real usage. +- Round *up*, never down: CPU to the next 50m, memory to the next 64Mi. +- Preserve the existing request:limit ratio. If that ratio is itself unreasonable, say so in + the PR body rather than silently normalizing it -- the ratio is someone's decision. +- **Movement under 20% is not a recommendation.** A 5% shave is measurement noise, and + proposing it teaches reviewers to ignore us. Drop it. +- Convert to human units before writing anything anyone reads. Never write a raw 10+ digit + integer (nanocores, bytes) in a PR body, a card, or a comment. + +**Step 5: Decide, asymmetrically** + +This is the most important step. Treating the two directions as symmetric is the classic way +to cause an outage while saving money. + +*CPU* -- symmetric. Move it when the mis-size is at least 20% and has been sustained for at +least 7 continuous days. The failure mode of too little CPU is throttling, which is +recoverable. Before proposing any decrease, confirm there is no meaningful throttling +already occurring; if there is, the correct direction is *up*, not down. + +*Memory* -- asymmetric. Increases follow CPU's terms. A **decrease** is allowed only when +`max` never exceeded the proposed request across the entire window. Check `max`, not `p95`: +the failure mode of too little memory is an OOM-kill, which is an outage, and a p95 that +looks comfortable tells you nothing about the peak that will kill the pod. + +*maxReplicas* -- trim only when peak replicas stayed well below the current maximum for the +whole window, and only to a value that leaves clear room above that observed peak. The +maximum is a ceiling for a traffic event that has not happened yet, not a target. + +**Step 6: Account for the autoscaler you actually found** + +- *CPU or memory-scaled (HPA on resource metrics)*: all three dimensions are in play. +- *Externally-scaled (KEDA on queue depth, event lag, a custom metric)*: the resource + request is not the scaling input. Keep the memory-p95 finding, drop any + CPU-as-scaling-signal framing, and name the real scaling signal. +- *Scale-to-zero*: percentiles computed over long idle periods understate active demand. + Measure only active periods, or state plainly that you cannot size it. +- *Node-level autoscaling only (Karpenter, cluster-autoscaler)*: this scales nodes, not + replicas, so there is no replica bound to trim. Requests still matter here -- they are the + input the node autoscaler bin-packs against, so right-sizing them is what actually reduces + node count. Say that in the PR body. +- *Pinned replicas (min == max)*: be conservative -- a single pod absorbs every spike alone. + Flag the pinning itself as a finding. + +Always state the constraint on the card and in the PR body. A reviewer who is not told about +it will assume we simply missed it. + +**Step 7: Check prior work before writing anything** + +- Call list_hpa_vpa_recommendations **first**, before opening any PR. +- An open recommendation for a workload means *update* it -- do not open a second PR. +- A workload in cooldown that is still quiet gets skipped. A human said no; asking again in a + week is how this action gets muted. +- Also search the VCS for prior right-sizing work, covering **open and recently merged** + (roughly the last 4 weeks) -- a change merged days ago may already be in effect, and + re-proposing it makes us look like we are not reading their repo. Search on all of: + the workload name, the IaC file path you intend to touch, the branch-name pattern this + action uses, and titles suggesting an earlier attempt (for example "consolidate", "tune", + "right-size", or "supersedes"). +- If a matching PR is open and unmerged, comment on it instead of opening a duplicate, and + reference the earlier PR number in anything you write. + +**Step 8: Open the PR** + +- One workload per PR. A reviewer should be able to accept or reject a single decision. +- Change only what your evidence supports. Do not tidy adjacent config. +- Match the repo's branch naming, title style, and label conventions. +- At most one short inline comment, and only where a number would otherwise look arbitrary. +- PR body, a few lines per dimension: current value, recommended value, the p95, the max, the + measurement window, the autoscaler constraint, and a link to this run. No essay. +- Before opening, grep your own diff for 10+ digit integers and human-readable units. + +**Step 9: Notify** + +- Call send_hpa_vpa_recommendation once per workload you opened a PR for, passing the + per-dimension current, recommended, and evidence values, and the autoscaler so the card can + explain a partial recommendation. Omit any dimension you are not recommending a change to. +- If it returns `suppressed`, that is the anti-nag rule working correctly. Record it in the + living document and move on. **Do not work around it** -- do not re-post, rename the + workload, or open the PR anyway. +- If you found no material mis-sizing anywhere: no PR, no card. Update the living document + with which workloads you checked, over what window, and what you found. That is a good + outcome, not a failed run. + +Hard anti-slop rules: +- No recommendation without a sustained measurement behind it -- never from a single spike, + a short window, or an alert threshold +- No memory decrease when the observed `max` exceeded the proposed request +- No CPU decrease while the workload is already being throttled +- No recommendation under 20% movement +- No raw nanocore or byte integers in any human-facing text +- No second PR for a workload that already has an open recommendation +- No re-proposing a workload inside its cooldown unless the mis-size materially worsened +- No numbers you could not reconcile against the team's own monitor definitions +- Nothing is ever applied: the PR is the change, and a human merges it""" diff --git a/server/services/actions/hpa_vpa_recommendations.py b/server/services/actions/hpa_vpa_recommendations.py new file mode 100644 index 000000000..1185d83ba --- /dev/null +++ b/server/services/actions/hpa_vpa_recommendations.py @@ -0,0 +1,488 @@ +"""Lifecycle helpers for HPA/VPA right-sizing recommendations. + +Owns every read and write of ``hpa_vpa_recommendations`` plus the PR-close +dispatch, so the two callers that live in different containers share one set of +invariants: + +- the card tool (``celery_worker`` / ``chatbot``) claims a workload and posts +- the Slack Dismiss handler (``aurora-server``) transitions it and closes the PR + +No raw SQL against this table belongs anywhere else. Every function fails soft +and returns ``{"error": ...}`` rather than raising -- ``initialize_tables()`` +runs at boot in ``aurora-server`` only, so a worker that starts first can +legitimately see a missing table and must not crash the agent loop. +""" + +import hashlib +import json +import logging +import math +import re +from datetime import datetime, timedelta, timezone +from typing import Any, Optional + +import requests + +from utils.auth.stateless_auth import set_rls_context +from utils.db.connection_pool import db_pool +from utils.log_sanitizer import sanitize + +logger = logging.getLogger(__name__) + +# Days a dismissed workload stays suppressed. Module constant so tests can +# monkeypatch it and so tuning it later does not mean re-deriving cooldowns +# from dismissed_at -- cooldown_until is stamped absolute at dismiss time. +HPA_VPA_COOLDOWN_DAYS = 30 + +# A re-proposal during cooldown must clear this multiple of the dismissed +# severity to count as "materially worse". This is the only thing that makes +# the design doc's "unless the mis-size materially worsens" computable rather +# than an LLM judgement call. +MATERIALLY_WORSE_FACTOR = 1.25 + +# Terminal/live status values -- see the DDL comment in db_utils.py. +STATUS_PROPOSED = "proposed" +STATUS_DISMISSED = "dismissed" +STATUS_MERGED = "merged" +STATUS_CLOSED = "closed" +STATUS_SUPERSEDED = "superseded" + +# Exactly owner/repo, the only shape any supported provider accepts. Guards the +# API URL these values are interpolated into. +_REPO_FULL_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$") + +_ROW_FIELDS = ( + "id", "workload_key", "workload", "environment", "service", "autoscaler", + "metrics_source", "vcs_provider", "repo_full_name", "pr_number", "pr_url", + "status", "recommendation", "severity_score", "slack_channel_id", + "slack_message_ts", "dismissed_by", "dismissed_at", "cooldown_until", + "created_at", "updated_at", +) + + +# --------------------------------------------------------------------------- +# Keys and scoring +# --------------------------------------------------------------------------- + + +def build_workload_key(workload: str, environment: Optional[str] = None) -> str: + """Normalized dedup/cooldown key. Computed here, never accepted from the LLM. + + The same workload in two environments is two independent rows, and + case/whitespace drift in the model's output cannot defeat a cooldown. + """ + env = (environment or "").strip() or "-" + return f"{env}::{(workload or '').strip()}".lower() + + +def _advisory_lock_key(org_id: str, workload_key: str) -> int: + """Stable bigint lock key for pg_advisory_xact_lock (executor.py pattern).""" + digest = hashlib.sha256(f"{org_id}:{workload_key}".encode()).digest()[:7] + return int.from_bytes(digest, byteorder="big", signed=False) & 0x7FFFFFFFFFFFFFFF + + +def _is_real_number(value: object) -> bool: + """Whether a value is a finite int/float usable in arithmetic. + + Rejects bool (an int subclass, so True would score as 1) and NaN/inf, both of + which reach us from LLM-supplied JSON and would otherwise corrupt a + comparison rather than fail it. + + ``math.isfinite`` raises OverflowError on an arbitrary-precision int too + large for a float (``json.loads`` imposes no bound, so a model can emit one), + which would propagate out of the cooldown gate as an unhandled error rather + than a decision. Treat unrepresentable as not usable. + """ + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + +def _dimension_score(spec: object) -> Optional[float]: + """Relative mis-size for one dimension, or None if it is not scoreable.""" + if not isinstance(spec, dict): + return None + current, recommended = spec.get("current"), spec.get("recommended") + if not (_is_real_number(current) and _is_real_number(recommended)): + return None + if current <= 0: + return None + return abs(float(recommended) - float(current)) / float(current) + + +def compute_severity_score(dimensions: dict) -> Optional[float]: + """Max relative mis-size across dimensions: max(abs(current - rec) / current). + + ``dimensions`` maps a dimension name to a dict with numeric ``current`` and + ``recommended``. Dimensions missing either value, or with a non-positive + current, are skipped. Returns None when nothing is scoreable. + """ + scores = [s for s in map(_dimension_score, (dimensions or {}).values()) if s is not None] + return max(scores) if scores else None + + +def is_materially_worse(new_score: Optional[float], prior_score: Optional[float]) -> bool: + """Whether a new recommendation justifies breaking an active cooldown. + + Unknown scores are treated as NOT worse: a missing number must never be a + reason to nag someone who already said no. That includes NaN and infinity -- + the new score originates from the LLM, and `inf` would otherwise compare as + worse than everything and break any cooldown on demand. Callers sanitize too; + this is the gate, so it enforces the invariant itself. + """ + if not (_is_real_number(new_score) and _is_real_number(prior_score)): + return False + if prior_score <= 0: + return False + return float(new_score) >= float(prior_score) * MATERIALLY_WORSE_FACTOR + + +def _row_to_dict(row: tuple) -> dict: + """Map a _ROW_FIELDS-ordered tuple to a JSON-safe dict.""" + out: dict[str, Any] = {} + for key, value in zip(_ROW_FIELDS, row): + if isinstance(value, datetime): + out[key] = value.isoformat() + elif key == "severity_score" and value is not None: + out[key] = float(value) + elif key == "id": + out[key] = str(value) + else: + out[key] = value + return out + + +_SELECT_ROW = f"SELECT {', '.join(_ROW_FIELDS)} FROM hpa_vpa_recommendations" + + +# --------------------------------------------------------------------------- +# Reads +# --------------------------------------------------------------------------- + + +def get_active_cooldown(cursor, org_id: str, workload_key: str) -> Optional[dict]: + """Most recent dismissal whose cooldown is still running, or None.""" + cursor.execute( + f"""{_SELECT_ROW} + WHERE org_id = %s AND workload_key = %s + AND status = %s AND cooldown_until > NOW() + ORDER BY dismissed_at DESC LIMIT 1""", + (org_id, workload_key, STATUS_DISMISSED), + ) + row = cursor.fetchone() + return _row_to_dict(row) if row else None + + +def get_live_recommendation(cursor, org_id: str, workload_key: str) -> Optional[dict]: + """The single open ('proposed') recommendation for a workload, or None.""" + cursor.execute( + f"{_SELECT_ROW} WHERE org_id = %s AND workload_key = %s AND status = %s LIMIT 1", + (org_id, workload_key, STATUS_PROPOSED), + ) + row = cursor.fetchone() + return _row_to_dict(row) if row else None + + +def list_recommendations(user_id: str) -> dict: + """Live proposals plus workloads still inside a cooldown window. + + This is what lets the prompt check prior work *before* opening a PR. Doing + it after is the worst ordering: a suppressed workload still gets a PR + nobody asked for. + """ + try: + with db_pool.get_admin_connection() as conn: + with conn.cursor() as cur: + org_id = set_rls_context(cur, conn, user_id, log_prefix="[HpaVpaRecs]") + if not org_id: + return {"error": "Could not resolve organization context"} + cur.execute( + f"""{_SELECT_ROW} + WHERE org_id = %s + AND (status = %s + OR (status = %s AND cooldown_until > NOW())) + ORDER BY created_at DESC LIMIT 200""", + (org_id, STATUS_PROPOSED, STATUS_DISMISSED), + ) + rows = [_row_to_dict(r) for r in cur.fetchall()] + except Exception: + logger.exception("[HpaVpaRecs] Failed to list recommendations for user=%s", sanitize(user_id)) + return {"error": "Could not read existing right-sizing recommendations"} + + open_recs = [r for r in rows if r["status"] == STATUS_PROPOSED] + cooling = [r for r in rows if r["status"] == STATUS_DISMISSED] + return { + "open_recommendations": open_recs, + "in_cooldown": cooling, + "counts": {"open": len(open_recs), "in_cooldown": len(cooling)}, + "cooldown_days": HPA_VPA_COOLDOWN_DAYS, + "guidance": ( + "Update an open recommendation instead of opening a second PR for the same " + "workload. Skip a workload in cooldown unless the mis-size has materially " + f"worsened (>= {MATERIALLY_WORSE_FACTOR}x the dismissed severity)." + ), + } + + +# --------------------------------------------------------------------------- +# Writes +# --------------------------------------------------------------------------- + + +def lock_workload(cursor, org_id: str, workload_key: str) -> None: + """Serialize concurrent dedup for one (org, workload) inside this txn. + + SELECT-then-INSERT under an advisory lock, not ON CONFLICT against the + partial unique index -- Postgres only infers a partial index when the + statement repeats its predicate, which is fiddly and easy to get subtly + wrong. The unique index stays as the backstop. + """ + cursor.execute("SELECT pg_advisory_xact_lock(%s)", (_advisory_lock_key(org_id, workload_key),)) + + +def claim_recommendation( + cursor, org_id: str, user_id: str, *, workload_key: str, workload: str, + environment: Optional[str], service: Optional[str], autoscaler: Optional[str], + metrics_source: Optional[str], vcs_provider: str, repo_full_name: str, + pr_number: int, pr_url: str, recommendation: dict, + severity_score: Optional[float], action_run_id: Optional[str] = None, +) -> str: + """INSERT a 'proposed' row with a NULL message ts and return its UUID. + + Called under :func:`lock_workload` and *before* the Slack post, so the + workload is claimed and the UUID the Dismiss button needs exists before any + external call can fail halfway. + """ + cursor.execute( + """INSERT INTO hpa_vpa_recommendations + (org_id, user_id, workload_key, workload, environment, service, autoscaler, + metrics_source, vcs_provider, repo_full_name, pr_number, pr_url, + status, recommendation, severity_score, action_run_id) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s, %s) + RETURNING id""", + (org_id, user_id, workload_key, workload, environment, service, autoscaler, + metrics_source, vcs_provider, repo_full_name, pr_number, pr_url, + STATUS_PROPOSED, json.dumps(recommendation or {}), severity_score, action_run_id), + ) + return str(cursor.fetchone()[0]) + + +def attach_slack_message(cursor, rec_id: str, channel_id: str, message_ts: str) -> None: + """Record where the card landed, so it can be updated and rewritten later.""" + cursor.execute( + """UPDATE hpa_vpa_recommendations + SET slack_channel_id = %s, slack_message_ts = %s, updated_at = NOW() + WHERE id = %s::uuid""", + (channel_id, message_ts, rec_id), + ) + + +def delete_recommendation(cursor, rec_id: str) -> None: + """Release a claimed row after a failed post, so a transient Slack outage + does not permanently block the workload.""" + cursor.execute("DELETE FROM hpa_vpa_recommendations WHERE id = %s::uuid", (rec_id,)) + + +def refresh_recommendation( + cursor, rec_id: str, *, recommendation: dict, severity_score: Optional[float], + repo_full_name: str, pr_number: int, pr_url: str, + autoscaler: Optional[str] = None, metrics_source: Optional[str] = None, +) -> None: + """Update an existing open recommendation with fresh numbers (card dedup path).""" + cursor.execute( + """UPDATE hpa_vpa_recommendations + SET recommendation = %s::jsonb, severity_score = %s, repo_full_name = %s, + pr_number = %s, pr_url = %s, + autoscaler = COALESCE(%s, autoscaler), + metrics_source = COALESCE(%s, metrics_source), + updated_at = NOW() + WHERE id = %s::uuid""", + (json.dumps(recommendation or {}), severity_score, repo_full_name, + pr_number, pr_url, autoscaler, metrics_source, rec_id), + ) + + +def mark_superseded(cursor, rec_id: str) -> None: + """Retire a row whose mis-size worsened, or that a newer rec took over.""" + cursor.execute( + """UPDATE hpa_vpa_recommendations + SET status = %s, cooldown_until = NULL, updated_at = NOW() + WHERE id = %s::uuid""", + (STATUS_SUPERSEDED, rec_id), + ) + + +def dismiss_recommendation(cursor, rec_id: str, org_id: str, slack_user_id: str) -> Optional[dict]: + """Atomically transition 'proposed' -> 'dismissed' and start the cooldown. + + The ``status = 'proposed'`` predicate *is* the double-click defence: a + second click matches zero rows. Returns the row's details on the winning + transition, or None when it had already been dismissed. + """ + now = datetime.now(timezone.utc) + cursor.execute( + """UPDATE hpa_vpa_recommendations + SET status = %s, dismissed_by = %s, dismissed_at = %s, + cooldown_until = %s, updated_at = %s + WHERE id = %s::uuid AND org_id = %s AND status = %s + RETURNING repo_full_name, pr_number, pr_url, workload, environment, service, + autoscaler, vcs_provider, recommendation, severity_score, + slack_channel_id, slack_message_ts, user_id""", + (STATUS_DISMISSED, slack_user_id, now, + now + timedelta(days=HPA_VPA_COOLDOWN_DAYS), now, + rec_id, org_id, STATUS_PROPOSED), + ) + row = cursor.fetchone() + if not row: + return None + # user_id is the account whose GitHub credential opened the PR. The clicker + # is a different person and may have no GitHub connection at all, so the + # close must be able to fall back to the opener. + keys = ("repo_full_name", "pr_number", "pr_url", "workload", "environment", "service", + "autoscaler", "vcs_provider", "recommendation", "severity_score", + "slack_channel_id", "slack_message_ts", "user_id") + out = dict(zip(keys, row)) + out["cooldown_until"] = (now + timedelta(days=HPA_VPA_COOLDOWN_DAYS)).isoformat() + if out.get("severity_score") is not None: + out["severity_score"] = float(out["severity_score"]) + return out + + +def mark_merged(cursor, rec_id: str, org_id: str) -> None: + """A merged PR was *accepted*, not rejected. + + Clearing cooldown_until matters: a merge must never start an anti-nag + window, or the next genuine drift on this workload goes unreported. + """ + cursor.execute( + """UPDATE hpa_vpa_recommendations + SET status = %s, cooldown_until = NULL, updated_at = NOW() + WHERE id = %s::uuid AND org_id = %s""", + (STATUS_MERGED, rec_id, org_id), + ) + + +# --------------------------------------------------------------------------- +# PR close -- provider dispatch +# --------------------------------------------------------------------------- + + +def _close_github_pr(user_id: str, repo_full_name: str, pr_number: int, timeout: int) -> dict: + """PATCH /repos/{repo}/pulls/{n} {"state": "closed"}. + + First ``requests.patch`` in ``server/`` -- every other GitHub call here is + read-only (``_call_github_api`` hardcodes ``requests.get``). Not routed + through MCP on purpose: ``mcp_github_update_pull_request`` is allowlisted in + ``mcp_tools.py`` but absent from ``tool_registry.py``, so ``gate_action`` + denies it in background context. Direct REST is the only reliable path. + + Uses ``get_auth_for_user_repo`` rather than ``get_installation_token``: we + know ``(user_id, repo_full_name)`` and not the installation id, and the + router handles App-vs-OAuth and sets its own RLS context (Celery-safe). + """ + from utils.auth.github_auth_router import ( + NoGitHubAuthError, + get_auth_for_user_repo, + make_auth_header, + ) + + try: + auth = get_auth_for_user_repo(user_id, repo_full_name) + except NoGitHubAuthError as exc: + return {"error": f"No GitHub credential for {repo_full_name}: {exc}"} + except Exception as exc: + logger.exception("[HpaVpaRecs] GitHub auth resolution failed for user=%s", sanitize(user_id)) + return {"error": f"GitHub auth resolution failed: {type(exc).__name__}"} + + headers = { + **make_auth_header(auth), + "Accept": "application/vnd.github+json", + "X-GitHub-Api-Version": "2022-11-28", + } + url = f"https://api.github.com/repos/{repo_full_name}/pulls/{int(pr_number)}" + + try: + resp = requests.patch(url, headers=headers, json={"state": "closed"}, timeout=timeout) + except requests.RequestException as exc: + return {"error": f"GitHub API request failed: {type(exc).__name__}"} + + if resp.status_code == 200: + try: + body = resp.json() + except ValueError: + body = {} + # A merged PR means the human already accepted. GitHub reports the + # merge rather than erroring, and treating it as a rejection would + # start an anti-nag window on an *approved* change. + if body.get("merged"): + return {"success": True, "already_merged": True, "state": body.get("state")} + return {"success": True, "state": body.get("state", "closed")} + if resp.status_code == 404: + # Already gone, or no access. Idempotent either way -- there is no open + # PR left to close, which is the state the caller wanted. + return {"success": True, "already_gone": True} + if resp.status_code == 403: + return {"error": f"GitHub returned 403 closing PR #{pr_number} (rate limit or missing permission)"} + if resp.status_code == 422: + return {"error": f"GitHub returned 422 closing PR #{pr_number} (invalid state): {resp.text[:200]}"} + return {"error": f"GitHub API status={resp.status_code} closing PR #{pr_number}: {resp.text[:200]}"} + + +# Only GitHub is implemented today. The other two are a fill-in, not a schema +# migration, which is what the vcs_provider column buys us: +# "gitlab" -> PUT /projects/{id}/merge_requests/{iid} {"state_event": "close"} +# via gitlab_api_request (verb-generic: requests.request(method, ...)) +# "bitbucket" -> client.decline_pull_request(...) (already exists) +_CLOSERS = {"github": _close_github_pr} + + +def supported_vcs_providers() -> frozenset: + """Providers a PR can actually be closed on. + + Callers validate against this *before* posting a card, so a workload is + never proposed on a provider whose Dismiss could not close the PR. + """ + return frozenset(_CLOSERS) + + +def close_pull_request( + user_id: str, vcs_provider: str, repo_full_name: str, pr_number: int, timeout: int = 30 +) -> dict: + """Close an open PR/MR on the provider that hosts it. + + Returns ``{"success": True, ...}`` or ``{"error": ...}``. An unknown + provider is a loud error rather than a silent no-op -- same discipline as + the trailing ``raise AssertionError`` in ``_do_query_logs``. + + A blank provider is an error too, not a silent default to GitHub: callers + already supply the fallback, so a blank here means the row is malformed and + guessing would fire a real GitHub PATCH at whatever repo string came with it. + """ + provider = (vcs_provider or "").lower().strip() + closer = _CLOSERS.get(provider) + if closer is None: + return {"error": (f"Cannot close PR: unsupported vcs_provider '{provider}'. " + f"Supported: {', '.join(sorted(_CLOSERS))}")} + if not repo_full_name or not pr_number: + return {"error": "Cannot close PR: missing repository or PR number"} + # The repo slug is interpolated into an API URL, so constrain it to exactly + # owner/repo. Rejects traversal ("../.."), extra path segments and query or + # fragment characters that would otherwise retarget the request. + if not _REPO_FULL_NAME_RE.match(repo_full_name): + return {"error": ("Cannot close PR: malformed repository name " + f"'{repo_full_name[:60]}' (expected 'owner/repo')")} + try: + pr_number = int(pr_number) + except (TypeError, ValueError): + return {"error": f"Cannot close PR: PR number '{pr_number}' is not an integer"} + if pr_number <= 0: + return {"error": f"Cannot close PR: PR number {pr_number} is not positive"} + + logger.info( + "[HpaVpaRecs] Closing %s PR #%s in %s for user=%s", + provider, pr_number, sanitize(repo_full_name), sanitize(user_id), + ) + return closer(user_id, repo_full_name, pr_number, timeout) diff --git a/server/services/actions/system_actions.py b/server/services/actions/system_actions.py index 5df59f6e4..ae127dc53 100644 --- a/server/services/actions/system_actions.py +++ b/server/services/actions/system_actions.py @@ -11,6 +11,7 @@ from services.actions.postmortem_action import DEFAULT_POSTMORTEM_INSTRUCTIONS from services.actions.alert_gap_action import DEFAULT_ALERT_GAP_INSTRUCTIONS +from services.actions.hpa_vpa_action import DEFAULT_HPA_VPA_INSTRUCTIONS from utils.db.connection_pool import db_pool @@ -36,16 +37,55 @@ "enabled": False, "instructions": None, }, + { + "system_key": "hpa_vpa_rightsizing", + # This name becomes the living-document artifact title, the chat session + # title, and the text in Slack action notifications -- so it reads as a + # document title and pairs with its sibling "Alert Gap Audit". "HPA/VPA" + # is avoided deliberately: the slash reads badly as a title, and VPA is + # not actually deployed in the target environment. + "name": "Right-Sizing Audit", + "description": "Periodically compares real CPU and memory usage against configured " + "requests, limits, and autoscaler bounds, and opens a PR per materially " + "mis-sized workload with a Slack card to review or dismiss.", + "trigger_type": "on_schedule", + "trigger_config": {"interval_seconds": 604800}, + # mode 'agent' is required, not stylistic: ModeAccessController strips + # mcp_* tools in ask mode, which would kill PR creation outright. + "mode": "agent", + "enabled": False, + "instructions": None, + }, ] +_DEFAULT_INSTRUCTIONS = { + "generate_postmortem": DEFAULT_POSTMORTEM_INSTRUCTIONS, + "alert_gap_audit": DEFAULT_ALERT_GAP_INSTRUCTIONS, + "hpa_vpa_rightsizing": DEFAULT_HPA_VPA_INSTRUCTIONS, +} + +# Fail fast at import if a SYSTEM_ACTIONS entry has no instructions. Deferring +# this to seeding time is far worse: _get_default_instructions runs inside the +# loop, so one unmapped key raises on every pass, gets swallowed by the except +# in seed_system_actions, and silently stops ALL system actions from seeding for +# every org that lacks them. +# Raise rather than assert: `python -O` strips assert statements, which would +# silently disable exactly the check that keeps a missing key from breaking +# seeding for every org. +_missing_instructions = {a["system_key"] for a in SYSTEM_ACTIONS} - set(_DEFAULT_INSTRUCTIONS) +if _missing_instructions: + raise RuntimeError( + f"SYSTEM_ACTIONS entries missing default instructions: {_missing_instructions}" + ) + + def _get_default_instructions(system_key: str) -> str: """Resolve the default instructions for a given system action.""" - if system_key == "generate_postmortem": - return DEFAULT_POSTMORTEM_INSTRUCTIONS - if system_key == "alert_gap_audit": - return DEFAULT_ALERT_GAP_INSTRUCTIONS - raise ValueError(f"Unknown system action: {system_key}") + try: + return _DEFAULT_INSTRUCTIONS[system_key] + except KeyError: + raise ValueError(f"Unknown system action: {system_key}") from None def seed_system_actions(org_id: str, user_id: Optional[str] = None) -> int: @@ -61,7 +101,14 @@ def seed_system_actions(org_id: str, user_id: Optional[str] = None) -> int: with conn.cursor() as cur: for action_def in SYSTEM_ACTIONS: key = action_def["system_key"] - instructions = _get_default_instructions(key) + try: + instructions = _get_default_instructions(key) + except ValueError: + # Skip only the broken entry. Letting this propagate would + # abort the whole loop and leave the remaining system + # actions unseeded for this org. + logger.exception("[SystemActions] No default instructions for '%s'; skipping", key) + continue cur.execute( "SELECT id FROM actions WHERE org_id = %s AND system_key = %s", diff --git a/server/tests/chat/test_datadog_metric_stats.py b/server/tests/chat/test_datadog_metric_stats.py new file mode 100644 index 000000000..da670438d --- /dev/null +++ b/server/tests/chat/test_datadog_metric_stats.py @@ -0,0 +1,408 @@ +"""Tests the server-side percentile summaries behind +``query_datadog(resource_type='metric_stats')``. + +Datadog has no time-percentile capability at all -- every documented route +(``.rollup(percentile,...)``, the ``pXX:`` prefix, ``formula: "p95(a)"``, the +scalar ``percentile`` aggregator) is rejected or silently returns zero series, +and both right-sizing metrics are gauges. So percentiles are computed here, in +Python, and these tests are the only thing pinning that arithmetic. + +Two properties are load-bearing and get explicit coverage: + +1. A reported p95 must be a **real observed sample**, because it lands in a PR + body as evidence a reviewer will look up in Datadog. +2. One row PER SERIES, small enough that ``_truncate_results`` can never drop a + whole payload. The single fat dict that ``resource_type='metrics'`` returns + truncates to ``count: 0``, which an agent cannot distinguish from "this + workload has no data" -- and it would then recommend cutting an idle-looking + workload. That failure mode is reproduced below to pin the contrast. + +Pure functions only: no DB, no network, no Datadog credentials. +""" + +import json +import os +import sys + +import pytest + +_server_dir = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir) +if os.path.abspath(_server_dir) not in sys.path: + sys.path.insert(0, os.path.abspath(_server_dir)) + +from chat.backend.agent.tools.datadog_tool import ( # noqa: E402 + _DATADOG_POINT_CAP, + _MAX_STATS_SERIES, + _MAX_INTERVAL_MS, + _MIN_INTERVAL_MS, + _P95_BY_SOURCE, + _METRIC_STATS_SOURCES, + _clamp_interval, + _p95_datadog, + _percentile, + _point_cap_note, + _pick_interval, + _query_metric_stats, + _query_metrics, + _round_sig, + _scope_and_group, + _summarize_series, + _truncate_results, +) +from chat.backend.agent.utils.tool_output_cap import PASS_THROUGH_CHARS # noqa: E402 + +_THIRTY_DAYS_MS = 30 * 24 * 3600 * 1000 + + +class _FakeClient: + """Stands in for DatadogClient, recording the interval it was handed.""" + + def __init__(self, series, values): + self._series = series + self._values = values + self.interval = None + + def query_metrics(self, query, start_ms, end_ms, interval=None): + self.interval = interval + return {"data": {"attributes": {"series": self._series, "values": self._values}}} + + +def _make_payload(n_series, n_points, value=5.0e8): + series = [ + {"group_tags": ["env:production", f"kube_deployment:svc-{i:03d}"], + "unit": [{"name": "byte"}, None]} + for i in range(n_series) + ] + values = [[value for _ in range(n_points)] for _ in range(n_series)] + return series, values + + +# --------------------------------------------------------------------------- +# _percentile -- nearest rank +# --------------------------------------------------------------------------- + + +def test_percentile_single_point(): + """n=1 must not raise -- statistics.quantiles does.""" + assert _percentile([42.0], 0.95) == 42.0 + assert _percentile([42.0], 0.5) == 42.0 + + +def test_percentile_two_points(): + """n=2 must not raise -- statistics.quantiles requires n >= 2 and interpolates.""" + assert _percentile([1.0, 2.0], 0.95) == 2.0 + assert _percentile([1.0, 2.0], 0.5) == 1.0 + + +def test_percentile_exact_quantile_boundary(): + values = [float(i) for i in range(1, 101)] + assert _percentile(values, 0.50) == 50.0 + assert _percentile(values, 0.95) == 95.0 + assert _percentile(values, 0.99) == 99.0 + assert _percentile(values, 1.0) == 100.0 + + +@pytest.mark.parametrize("q", [0.5, 0.95, 0.99]) +def test_percentile_returns_an_observed_sample(q): + """The whole reason for nearest-rank: a reviewer must find this exact value.""" + values = [1.0, 7.5, 19.25, 100.0, 512.0] + assert _percentile(values, q) in values + + +def test_percentile_never_indexes_out_of_range(): + for n in range(1, 40): + values = [float(i) for i in range(n)] + for q in (0.0, 0.5, 0.95, 0.99, 1.0): + assert _percentile(values, q) in values + + +# --------------------------------------------------------------------------- +# Interval selection +# --------------------------------------------------------------------------- + + +def test_pick_interval_thirty_days_is_one_hour(): + """30 days -> 1h -> 720 points: legal against Datadog's 1500-point cap, and + finer than Datadog's own 4h default for a 30-day window.""" + chosen = _pick_interval(_THIRTY_DAYS_MS) + assert chosen == 3_600_000 + assert _THIRTY_DAYS_MS // chosen == 720 + + +def test_pick_interval_short_window_uses_finest(): + assert _pick_interval(3600 * 1000) == _MIN_INTERVAL_MS + + +@pytest.mark.parametrize("days", [1, 7, 14, 30, 60, 90, 180, 250]) +def test_pick_interval_stays_under_datadog_point_cap(days): + """Every window we realistically ask for must fit in one series.""" + span = days * 24 * 3600 * 1000 + assert span // _pick_interval(span) <= _DATADOG_POINT_CAP + + +def test_window_beyond_the_point_cap_is_flagged_not_silently_clipped(): + """Past ~250 days even the 4h ceiling overruns Datadog's 1500-point limit, so + the series comes back partial. A percentile over a silently-clipped window is + not the percentile that was asked for.""" + span_days = 365 + series, values = _make_payload(2, 1500) + client = _FakeClient(series, values) + result = _p95_datadog(client, "q", f"-{span_days}d", "now", 100) + assert result["interval_ms"] == _MAX_INTERVAL_MS + assert result["window_exceeds_point_cap"] is True + assert str(_DATADOG_POINT_CAP) in result["window_note"] + + +def test_point_cap_advice_matches_why_the_window_overran(): + """The advice has to be actionable: telling a caller who already pinned the + coarsest interval to "use a coarser interval" sends them nowhere.""" + long_span = 365 * 24 * 3600 * 1000 + + auto = _point_cap_note(long_span, _MAX_INTERVAL_MS, None) + assert "coarsest supported" in auto + + pinned_below_ceiling = _point_cap_note(long_span, 3_600_000, 3_600_000) + assert "coarser interval" in pinned_below_ceiling + + pinned_at_ceiling = _point_cap_note(long_span, _MAX_INTERVAL_MS, _MAX_INTERVAL_MS) + assert "coarser interval" not in pinned_at_ceiling + assert "coarsest supported" in pinned_at_ceiling + + assert _point_cap_note(30 * 24 * 3600 * 1000, 3_600_000, None) is None + + +def test_normal_window_carries_no_point_cap_warning(): + series, values = _make_payload(2, 720) + result = _p95_datadog(_FakeClient(series, values), "q", "-30d", "now", 100) + assert "window_exceeds_point_cap" not in result + + +def test_pick_interval_caller_value_wins(): + assert _pick_interval(_THIRTY_DAYS_MS, 300_000) == 300_000 + + +def test_pick_interval_clamps_caller_value(): + assert _pick_interval(_THIRTY_DAYS_MS, 1) == _MIN_INTERVAL_MS + assert _pick_interval(_THIRTY_DAYS_MS, 99_999_999) == _MAX_INTERVAL_MS + + +@pytest.mark.parametrize("bad", [None, "bogus", "", [], {}]) +def test_clamp_interval_rejects_non_numeric(bad): + assert _clamp_interval(bad) is None + + +def test_plain_metrics_does_not_auto_pick_an_interval(): + """resource_type='metrics' must keep Datadog's own default resolution when no + interval is given. Its output contract predates this feature and the RCA + prompts and SKILL.md depend on it -- auto-picking here would silently change + every existing RCA metric query. Only metric_stats, which is new and owns its + own contract, auto-picks.""" + client = _FakeClient(*_make_payload(1, 10)) + _query_metrics(client, "avg:system.cpu.user{*}", "-30d", "now", 100) + assert client.interval is None + + # An explicit value is still honoured (and clamped). + _query_metrics(client, "avg:system.cpu.user{*}", "-30d", "now", 100, interval=300_000) + assert client.interval == 300_000 + _query_metrics(client, "avg:system.cpu.user{*}", "-30d", "now", 100, interval=1) + assert client.interval == _MIN_INTERVAL_MS + + +# --------------------------------------------------------------------------- +# Rounding +# --------------------------------------------------------------------------- + + +def test_round_sig_keeps_six_significant_digits(): + assert _round_sig(536870912.0) == 536871000.0 + assert _round_sig(1.23456789) == 1.23457 + + +def test_round_sig_handles_zero_and_none(): + assert _round_sig(0.0) == 0.0 + assert _round_sig(None) is None + + +# --------------------------------------------------------------------------- +# Per-series summarization +# --------------------------------------------------------------------------- + + +def test_scope_is_stable_and_sorted(): + """scope doubles as a dict key across calls, so tag order must not matter.""" + a, _ = _scope_and_group(["kube_deployment:api", "env:production"]) + b, _ = _scope_and_group(["env:production", "kube_deployment:api"]) + assert a == b == "env:production,kube_deployment:api" + + +def test_group_parses_key_values(): + _, group = _scope_and_group(["env:production", "kube_deployment:apiv2worker"]) + assert group == {"env": "production", "kube_deployment": "apiv2worker"} + + +def test_all_null_series_reports_no_data_not_zero_usage(): + """The distinction that stops us cutting a workload we simply cannot see.""" + row = _summarize_series(["kube_deployment:api"], [{"name": "byte"}, None], [None, None, None]) + assert row["p95"] is None + assert row["points"] == 3 + assert row["nulls"] == 3 + assert "no non-null points" in row["note"] + + +def test_empty_series_reports_no_data(): + row = _summarize_series(["kube_deployment:api"], None, []) + assert row["p95"] is None + assert row["points"] == 0 + + +def test_nan_does_not_corrupt_the_sort_and_understate_max(): + """The load-bearing one. NaN compares False against everything, so a single + NaN point leaves list.sort() unsorted and vals[-1] is no longer the maximum. + max is exactly what guards the memory-decrease rule ("allowed only when max + never exceeded the proposed request"), so an understated max approves a + decrease against a peak that would OOM-kill the pod.""" + row = _summarize_series(["kube_deployment:api"], [{"name": "byte"}, None], + [1.0, float("nan"), 3.0, 2.0]) + assert row["max"] == 3.0 + assert row["nulls"] == 1 + assert row["points"] == 4 + + +def test_non_finite_points_never_reach_json(): + """json.dumps emits bare NaN/Infinity, which is invalid JSON and would fail + the agent's parse of the whole tool result.""" + row = _summarize_series(["x"], None, [1.0, float("inf"), float("-inf"), 2.0]) + serialized = json.dumps(row) + assert "Infinity" not in serialized and "NaN" not in serialized + json.loads(serialized) + assert row["max"] == 2.0 + assert row["nulls"] == 2 + + +def test_booleans_are_not_counted_as_samples(): + """bool is an int subclass, so True would otherwise be summarized as 1.0.""" + row = _summarize_series(["x"], None, [True, 5.0, False]) + assert row["nulls"] == 2 + assert row["max"] == 5.0 + + +def test_percentile_of_empty_list_returns_none(): + """Must not raise: min(-1, max(0, -1)) is -1, which would index the last + element (or IndexError on empty) rather than signalling "no data".""" + assert _percentile([], 0.95) is None + + +def test_nulls_are_counted_and_excluded_from_percentiles(): + row = _summarize_series(["kube_deployment:api"], [{"name": "byte"}, None], + [100.0, None, 200.0, None, 300.0]) + assert row["points"] == 5 + assert row["nulls"] == 2 + assert row["max"] == 300.0 + assert row["p50"] == 200.0 + + +def test_summary_never_contains_raw_points(): + row = _summarize_series(["kube_deployment:api"], [{"name": "byte"}, None], + [float(i) for i in range(720)]) + assert "values" not in row + assert set(row) == {"scope", "group", "unit", "points", "nulls", + "p50", "p95", "p99", "max", "mean"} + + +def test_unit_extracted_from_datadog_nullable_pair(): + row = _summarize_series([], [{"name": "byte"}, None], [1.0]) + assert row["unit"] == "byte" + + +def test_cpu_unit_is_surfaced_so_the_agent_can_convert_nanocores(): + """The prompt requires dividing nanocore CPU usage by 1e9 before comparing it + against a millicore request -- comparing the two directly is a ~1e6x error + that still looks like a plausible number in a PR body. That conversion is the + agent's job, but it can only do it if the unit reaches it, so pin that.""" + row = _summarize_series( + ["kube_deployment:apiv2worker"], + [{"name": "nanocore"}, None], + [480_000_000.0, 500_000_000.0], + ) + assert row["unit"] == "nanocore" + # Raw nanocores are reported as-is; scaling is deliberately not done here. + assert row["max"] == 500_000_000.0 + assert row["p95"] == 500_000_000.0 + + +# --------------------------------------------------------------------------- +# Handler output contract +# --------------------------------------------------------------------------- + + +def test_thirty_day_query_reports_720_points_at_one_hour(): + series, values = _make_payload(3, 720) + client = _FakeClient(series, values) + result = _p95_datadog(client, "sum:kubernetes.memory.usage{*} by {kube_deployment}", + "-30d", "now", 100) + assert client.interval == 3_600_000 + assert result["interval_ms"] == 3_600_000 + assert all(row["points"] == 720 for row in result["results"]) + assert result["metrics_source"] == "datadog" + + +def test_seventy_series_stays_under_summarizer_threshold(): + """Above PASS_THROUGH_CHARS the output is routed through an LLM summarizer, + which would paraphrase percentiles into plausible fiction -- numbers that + survive but come back looking authoritative are worse than truncated ones.""" + series, values = _make_payload(70, 720) + result = _p95_datadog(_FakeClient(series, values), "q", "-30d", "now", 100) + assert result["series_included"] == 70 + assert not result.get("series_truncated") + assert len(json.dumps(result)) < PASS_THROUGH_CHARS + + +def test_series_cap_is_reported_not_silent(): + series, values = _make_payload(400, 720) + result = _p95_datadog(_FakeClient(series, values), "q", "-30d", "now", 1000) + assert result["series_included"] <= _MAX_STATS_SERIES + assert result["series_returned"] == 400 + assert result.get("series_dropped") or result.get("series_truncated") + assert "note" in result + assert len(json.dumps(result)) < PASS_THROUGH_CHARS + + +def test_zero_series_is_an_empty_answer_not_an_error(): + result = _p95_datadog(_FakeClient([], []), "q", "-30d", "now", 100) + assert result["count"] == 0 + assert result["series_returned"] == 0 + assert result["results"] == [] + + +def test_per_series_rows_survive_truncation_but_one_fat_dict_does_not(): + """Pins the reason for the per-series shape, both directions at once.""" + series, values = _make_payload(70, 720) + + fat = [{"series": series, "times": list(range(720)), "values": values}] + kept_fat, truncated_fat = _truncate_results(fat, [json.dumps(i) for i in fat]) + assert kept_fat == [] and truncated_fat is True # -> count: 0, silent wrong answer + + rows = _p95_datadog(_FakeClient(series, values), "q", "-30d", "now", 100)["results"] + kept_rows, truncated_rows = _truncate_results(rows, [json.dumps(i) for i in rows]) + assert len(kept_rows) == len(rows) and truncated_rows is False + + +# --------------------------------------------------------------------------- +# Provider seam +# --------------------------------------------------------------------------- + + +def test_every_advertised_source_has_a_dispatch_branch(): + assert set(_P95_BY_SOURCE) >= set(_METRIC_STATS_SOURCES) + + +def test_unsupported_source_raises_clearly(): + client = _FakeClient([], []) + with pytest.raises(ValueError, match="Unsupported metrics source"): + _query_metric_stats(client, "q", "-30d", "now", 100, source="newrelic") + + +def test_empty_query_is_rejected(): + client = _FakeClient([], []) + with pytest.raises(ValueError, match="query is required"): + _query_metric_stats(client, "", "-30d", "now", 100) diff --git a/server/tests/conftest.py b/server/tests/conftest.py index c261a2466..fddfc0091 100644 --- a/server/tests/conftest.py +++ b/server/tests/conftest.py @@ -8,11 +8,12 @@ from __future__ import annotations +import importlib.machinery as _importlib_machinery import importlib.util as _importlib_util import os import sys from collections.abc import Iterator -from types import SimpleNamespace +from types import ModuleType, SimpleNamespace from typing import Any from unittest.mock import MagicMock @@ -32,6 +33,12 @@ os.environ.setdefault("POSTGRES_HOST", "localhost") os.environ.setdefault("POSTGRES_PORT", "5432") +# routes/gcp/auth.py does `os.getenv("FRONTEND_URL") + "/chat"` at import time, +# which raises TypeError when the var is unset. Any test that transitively +# imports a routes package (routes.datadog -> celery_config -> routes.gcp) hits +# it. Inert placeholder; nothing here dials the frontend. +os.environ.setdefault("FRONTEND_URL", "http://localhost:3000") + # Stub heavy third-party packages so source modules import in a lightweight # test env. Only stub when the real package isn't installed — some tests # (e.g. test_input_rail.py) need real classes like BaseChatModel / AIMessage. @@ -46,19 +53,114 @@ "langchain_core", "langchain_core.tools", "langchain_core.language_models", "langchain_core.language_models.chat_models", "langchain_anthropic", "langchain_openai", "langchain_google_genai", + "langchain_aws", "langchain_ollama", "kubernetes", "kubernetes.client", "kubernetes.client.rest", - "kubernetes.config", "kubernetes.stream", + "kubernetes.client.exceptions", "kubernetes.config", "kubernetes.stream", + # Submodules imported directly by source modules. Each needs its own entry: + # `from langgraph.types import ...` cannot resolve against a stub of the + # parent alone, since a stub package has no real search path. + "langchain.agents", "langchain.agents.middleware", + "langchain.agents.middleware.types", + "langchain_core.callbacks", "langchain_core.messages", + "langchain_core.messages.utils", "langchain_core.runnables", + "langchain_core.runnables.config", + "langgraph.checkpoint", "langgraph.checkpoint.memory", + "langgraph.graph", "langgraph.graph.state", "langgraph.types", + "celery.exceptions", "celery.signals", ) -for _pkg in _OPTIONAL_PACKAGES: - if _pkg in sys.modules: - continue +# Namespace packages that must be stubbed *prefix-wise*, not module-by-module. +# Reached transitively by anything importing a routes package (routes.datadog -> +# routes.datadog.tasks -> chat.background -> google_chat_connector and +# agent.providers.google_provider). Enumerating submodules does not work here: +# langchain_google_genai is installed in CI and walks arbitrary google.genai.* +# submodules at import time, so a fixed list always misses one. +_STUBBED_NAMESPACES = ("google", "googleapiclient") + + +class _StubModule(ModuleType): + """A real module object that mocks any attribute on demand. + + A bare ``MagicMock`` will not do here: as a stand-in for a *package* it needs + ``__path__`` (or Python raises "'google' is not a package") and a real + ``__spec__`` for pytest's assertion-rewriting machinery -- but setting those + on a MagicMock stops it auto-creating the *other* attributes callers want, + e.g. ``service_account.Credentials``. Subclassing ModuleType and mocking via + ``__getattr__`` satisfies both at once. + """ + + def __init__(self, fullname: str) -> None: + super().__init__(fullname) + self.__path__: list[str] = [] + self.__spec__ = _importlib_machinery.ModuleSpec(fullname, None, is_package=True) + + def __getattr__(self, item: str) -> Any: + if item.startswith("__") and item.endswith("__"): + raise AttributeError(item) + value = MagicMock(name=f"{self.__name__}.{item}") + setattr(self, item, value) + return value + + +# Submodules of the google* namespace packages that source modules import. +# Registered explicitly, parents before children: every one must be present in +# sys.modules before the import machinery sees it, because an empty __path__ +# lets Python's namespace-package resolution return a real, attribute-less +# module instead of these stubs. +_STUBBED_SUBMODULES = ( + "google", "google.auth", "google.auth.transport", + "google.auth.transport.requests", "google.oauth2", + "google.oauth2.credentials", "google.oauth2.service_account", + "google.cloud", "google.cloud.bigquery", + "google.api_core", "google.api_core.exceptions", + "google.genai", "google.genai.types", "google.genai.client", + "googleapiclient", "googleapiclient.discovery", "googleapiclient.errors", +) + +# langchain_google_genai is installed in CI but imports google.genai.* at module +# scope, so it cannot load against stubs. When google is absent, stub the wrapper +# too rather than chasing whichever google.genai submodule it reaches next. +_STUB_WHEN_GOOGLE_ABSENT = ("langchain_google_genai",) + +def _is_installed(name: str) -> bool: try: - spec = _importlib_util.find_spec(_pkg) + return _importlib_util.find_spec(name) is not None except (ImportError, ValueError): - spec = None - if spec is None: - sys.modules[_pkg] = MagicMock() + return False + + +def _stub(name: str) -> None: + if name not in sys.modules: + sys.modules[name] = _StubModule(name) + + +# Stub each optional package (and any listed submodule) only when the real one +# is absent, so a real installation always wins. _StubModule rather than a bare +# MagicMock because several of these are imported as packages -- e.g. +# `from langgraph.types import ...` fails against a MagicMock with +# "'langgraph' is not a package". +for _pkg in _OPTIONAL_PACKAGES: + # Check the exact module, not just its root: CI installs langchain-core, + # which brings a real `langgraph` package that has no `langgraph.types`. + # Gating on the root would skip the submodule and leave the import failing. + if not _is_installed(_pkg): + _stub(_pkg) + +for _ns in _STUBBED_NAMESPACES: + # Per-submodule, not per-namespace: `google` is a namespace package that any + # installed google-* distribution provides, so gating the whole loop on the + # root would leave a missing google.oauth2 unstubbed whenever some unrelated + # google package happens to be present. _stub() and _is_installed() both + # no-op on real modules, so nothing is ever masked. + for _mod in _STUBBED_SUBMODULES: + if _mod.split(".")[0] == _ns and not _is_installed(_mod): + _stub(_mod) + if _ns == "google" and not _is_installed("google.oauth2"): + # These wrappers import google.* at module scope, so they cannot load + # against stubs even when they are themselves installed. + for _wrapper in _STUB_WHEN_GOOGLE_ABSENT: + sys.modules.pop(_wrapper, None) + sys.modules[_wrapper] = _StubModule(_wrapper) try: from cryptography.hazmat.primitives import serialization diff --git a/server/tests/services/test_hpa_vpa_recommendations.py b/server/tests/services/test_hpa_vpa_recommendations.py new file mode 100644 index 000000000..d6822fc5a --- /dev/null +++ b/server/tests/services/test_hpa_vpa_recommendations.py @@ -0,0 +1,230 @@ +"""Tests the pure lifecycle logic behind HPA/VPA right-sizing recommendations. + +Three things here are the difference between an action a customer keeps enabled +and one they mute: + +- ``build_workload_key`` -- computed in Python and never accepted from the LLM, + so case or whitespace drift in model output cannot defeat a cooldown, and the + same workload in two environments stays two independent rows. +- ``compute_severity_score`` -- makes "the mis-size materially worsened" + computable rather than an LLM judgement call. +- ``is_materially_worse`` -- the gate that decides whether we are allowed to + re-raise a workload a human already said no to. + +Pure functions only: no DB, no network. The DB transitions themselves are +exercised against a live Postgres, not mocked here. +""" + +import os +import sys + +import pytest + +_server_dir = os.path.join(os.path.dirname(__file__), os.pardir, os.pardir) +if os.path.abspath(_server_dir) not in sys.path: + sys.path.insert(0, os.path.abspath(_server_dir)) + +from services.actions.hpa_vpa_recommendations import ( # noqa: E402 + HPA_VPA_COOLDOWN_DAYS, + MATERIALLY_WORSE_FACTOR, + _CLOSERS, + build_workload_key, + close_pull_request, + compute_severity_score, + is_materially_worse, +) + + +# --------------------------------------------------------------------------- +# workload_key normalization +# --------------------------------------------------------------------------- + + +def test_workload_key_is_case_and_whitespace_insensitive(): + """Model output drift must not open a second PR for the same workload.""" + canonical = build_workload_key("apiv2worker", "production") + assert build_workload_key("APIv2Worker", "Production") == canonical + assert build_workload_key(" apiv2worker ", " production ") == canonical + + +def test_workload_key_separates_environments(): + """Same workload in two envs is two independent recommendations.""" + assert build_workload_key("api", "production") != build_workload_key("api", "staging") + + +def test_workload_key_handles_missing_environment(): + assert build_workload_key("api", None) == "-::api" + assert build_workload_key("api", "") == "-::api" + assert build_workload_key("api", " ") == "-::api" + + +def test_workload_key_is_stable_across_calls(): + """Pin the exact format, not self-equality: the key is a persisted dedup + handle, so a format change silently orphans every existing cooldown row.""" + assert build_workload_key("api", "prod") == "prod::api" + + +# --------------------------------------------------------------------------- +# severity_score +# --------------------------------------------------------------------------- + + +def test_severity_is_relative_mis_size(): + score = compute_severity_score({"memory": {"current": 2048, "recommended": 768}}) + assert score == pytest.approx((2048 - 768) / 2048) + + +def test_severity_takes_the_worst_dimension(): + score = compute_severity_score({ + "memory": {"current": 100, "recommended": 50}, # 0.5 + "cpu": {"current": 100, "recommended": 10}, # 0.9 + }) + assert score == pytest.approx(0.9) + + +def test_severity_is_direction_agnostic(): + """An under-provisioned workload is just as material as an over-provisioned one.""" + down = compute_severity_score({"cpu": {"current": 100, "recommended": 50}}) + up = compute_severity_score({"cpu": {"current": 100, "recommended": 150}}) + assert down == pytest.approx(0.5) + assert up == pytest.approx(0.5) + + +@pytest.mark.parametrize("dimensions", [ + {}, + None, + {"cpu": {"current": 0, "recommended": 5}}, # non-positive current + {"cpu": {"current": -10, "recommended": 5}}, + {"cpu": {"current": "2Gi", "recommended": "1Gi"}}, # display strings, not numbers + {"cpu": {"current": 100}}, # incomplete + {"cpu": {"recommended": 100}}, + {"cpu": "not-a-dict"}, + {"cpu": {"current": True, "recommended": 50}}, # bool is an int subclass + {"cpu": {"current": 100, "recommended": False}}, + {"cpu": {"current": float("inf"), "recommended": 50}}, + {"cpu": {"current": 100, "recommended": float("nan")}}, +]) +def test_severity_is_none_when_nothing_is_scoreable(dimensions): + assert compute_severity_score(dimensions) is None + + +# --------------------------------------------------------------------------- +# "materially worse" -- the anti-nag gate +# --------------------------------------------------------------------------- + + +def test_materially_worse_at_and_above_the_factor(): + prior = 0.60 + assert is_materially_worse(prior * MATERIALLY_WORSE_FACTOR, prior) is True + assert is_materially_worse(0.90, prior) is True + + +def test_not_materially_worse_below_the_factor(): + assert is_materially_worse(0.70, 0.60) is False # 1.17x + assert is_materially_worse(0.60, 0.60) is False # unchanged + assert is_materially_worse(0.30, 0.60) is False # improved + + +@pytest.mark.parametrize("new,prior", [ + (None, 0.60), # unknown new score + (0.90, None), # unknown prior score + (0.90, 0.0), # unusable prior + (0.90, -1.0), + ("0.9", 0.60), # non-numeric +]) +def test_unknown_scores_never_break_a_cooldown(new, prior): + """A missing number must never become a reason to nag someone who said no.""" + assert is_materially_worse(new, prior) is False + + +@pytest.mark.parametrize("new,prior", [ + (float("inf"), 0.60), + (float("-inf"), 0.60), + (float("nan"), 0.60), + (0.90, float("inf")), + (0.90, float("nan")), +]) +def test_non_finite_scores_never_break_a_cooldown(new, prior): + """severity_score arrives from the LLM. Without a finiteness check, passing + Infinity compares as worse than everything and unlocks any cooldown on + demand -- defeating the anti-nag rule this feature exists to enforce.""" + assert is_materially_worse(new, prior) is False + + +def test_booleans_are_not_treated_as_scores(): + """bool is an int subclass, so True would otherwise sneak through as 1.0.""" + assert is_materially_worse(True, 0.60) is False + assert is_materially_worse(0.90, True) is False + + +def test_integers_too_large_for_a_float_do_not_raise(): + """math.isfinite raises OverflowError on an arbitrary-precision int that + cannot convert to a float, and json.loads imposes no bound -- so a model can + emit one. Unhandled, it escapes the cooldown gate as an error instead of a + decision, and the caller reports a generic post failure.""" + huge = 10 ** 400 + + assert is_materially_worse(huge, 0.60) is False + assert is_materially_worse(0.90, huge) is False + assert compute_severity_score({"cpu": {"current": huge, "recommended": 50}}) is None + assert compute_severity_score({"cpu": {"current": 100, "recommended": huge}}) is None + + # Large but representable values must still be accepted. + assert compute_severity_score({"cpu": {"current": 10 ** 300, "recommended": 10 ** 299}}) is not None + + +# --------------------------------------------------------------------------- +# Cooldown + provider dispatch +# --------------------------------------------------------------------------- + + +def test_cooldown_window_is_thirty_days(): + assert HPA_VPA_COOLDOWN_DAYS == 30 + + +def test_only_github_close_is_implemented_today(): + assert set(_CLOSERS) == {"github"} + + +@pytest.mark.parametrize("provider", ["gitlab", "bitbucket", "svn", ""]) +def test_unsupported_provider_is_a_loud_error(provider): + """A gap must be loud, not a silent no-op that leaves the PR open.""" + result = close_pull_request("user-1", provider, "owner/repo", 5) + assert "error" in result + if provider: + assert provider in result["error"] + + +def test_missing_pr_reference_is_rejected_before_any_network_call(monkeypatch): + """Assert the closer is never reached, not merely that an error came back -- + otherwise this passes for the wrong reason (the fake user's GitHub auth + lookup failing) and would keep passing if the guard were removed.""" + calls = [] + monkeypatch.setitem( + _CLOSERS, "github", lambda *args, **_kwargs: calls.append(args) or {"success": True} + ) + + assert "error" in close_pull_request("user-1", "github", "", 5) + assert "error" in close_pull_request("user-1", "github", "owner/repo", 0) + assert calls == [] + + +@pytest.mark.parametrize("bad_repo", [ + "../../etc/passwd", + "owner/repo/extra", + "owner", + "owner/repo?x=1", + "owner/repo#frag", + "https://evil.example.com/owner/repo", +]) +def test_malformed_repo_name_is_rejected_before_any_network_call(monkeypatch, bad_repo): + """The repo slug is interpolated into an API URL, so anything but owner/repo + must be refused before a request is built.""" + calls = [] + monkeypatch.setitem( + _CLOSERS, "github", lambda *args, **_kwargs: calls.append(args) or {"success": True} + ) + + result = close_pull_request("user-1", "github", bad_repo, 5) + assert "error" in result + assert calls == [] diff --git a/server/utils/db/db_utils.py b/server/utils/db/db_utils.py index 7cd1022f9..0de34e1b2 100644 --- a/server/utils/db/db_utils.py +++ b/server/utils/db/db_utils.py @@ -1409,6 +1409,63 @@ def initialize_tables(): CREATE INDEX IF NOT EXISTS idx_action_runs_action ON action_runs(action_id); CREATE INDEX IF NOT EXISTS idx_action_runs_status ON action_runs(org_id, status); """, + # Right-sizing recommendations proposed by the hpa_vpa_rightsizing action. + # One live row per (org, workload) while a PR is open; dismissal starts a + # cooldown so the action does not re-propose a change a human rejected. + # + # status values (VARCHAR + documented set, not an ENUM -- there are zero + # ENUMs in this module and ALTER TYPE ADD VALUE is transaction-hostile): + # proposed -- card posted, PR open (at most one per workload) + # dismissed -- human clicked Dismiss; cooldown_until is running + # merged -- human merged the PR; accepted, NO cooldown + # closed -- closed on the VCS by hand, no cooldown + # superseded -- mis-size materially worsened, or a newer rec took over + "hpa_vpa_recommendations": """ + CREATE TABLE IF NOT EXISTS hpa_vpa_recommendations ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + org_id VARCHAR(255) NOT NULL, + user_id VARCHAR(255) NOT NULL, + workload_key VARCHAR(512) NOT NULL, + workload VARCHAR(255) NOT NULL, + environment VARCHAR(128), + service VARCHAR(255), + autoscaler VARCHAR(64), + metrics_source VARCHAR(32), + vcs_provider VARCHAR(32) NOT NULL DEFAULT 'github', + repo_full_name VARCHAR(512), + pr_number INTEGER, + pr_url TEXT, + status VARCHAR(32) NOT NULL DEFAULT 'proposed', + recommendation JSONB NOT NULL DEFAULT '{}'::jsonb, + severity_score NUMERIC, + slack_channel_id VARCHAR(64), + slack_message_ts VARCHAR(64), + action_run_id UUID, + dismissed_by VARCHAR(255), + dismissed_at TIMESTAMP, + cooldown_until TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_hpa_vpa_recs_org_workload + ON hpa_vpa_recommendations(org_id, workload_key, created_at DESC); + CREATE UNIQUE INDEX IF NOT EXISTS idx_hpa_vpa_recs_live + ON hpa_vpa_recommendations(org_id, workload_key) WHERE status = 'proposed'; + CREATE INDEX IF NOT EXISTS idx_hpa_vpa_recs_cooldown + ON hpa_vpa_recommendations(org_id, workload_key, cooldown_until) + WHERE cooldown_until IS NOT NULL; + -- Superseded by idx_hpa_vpa_recs_live_pr below. Its predicate + -- covered every status, so a re-proposal pointing at a PR that + -- an earlier dismissal failed to close (a tolerated degraded + -- path) hit a unique violation and lost the whole card post. + -- CREATE ... IF NOT EXISTS cannot redefine an existing index, + -- hence the explicit drop; it is a no-op after the first boot. + DROP INDEX IF EXISTS idx_hpa_vpa_recs_pr; + CREATE UNIQUE INDEX IF NOT EXISTS idx_hpa_vpa_recs_live_pr + ON hpa_vpa_recommendations(org_id, repo_full_name, pr_number) + WHERE pr_number IS NOT NULL AND status = 'proposed'; + """, } # List of tables that should have RLS enabled and a policy applied. @@ -1485,6 +1542,7 @@ def initialize_tables(): rls_tables.append("postmortem_versions") rls_tables.append("artifacts") rls_tables.append("artifact_versions") + rls_tables.append("hpa_vpa_recommendations") # Migration: Add rca_celery_task_id column to incidents table if it doesn't exist