Improve reliability, scoring transparency, and query classification

dashboard_tooling/api_clients.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import json
+import time
 from dataclasses import dataclass
 from typing import Callable
+from urllib.error import HTTPError
 from urllib.parse import urlencode
 from urllib.request import Request, urlopen
 
@@ -13,27 +15,69 @@ class ApiClientError(RuntimeError):
     pass
 
 
+class RateLimitError(ApiClientError):
+    pass
+
+
 Transport = Callable[[Request], object]
 
+# Number of retries on transient failures (5xx, network errors, rate limits)
+_MAX_RETRIES = 3
+# Seconds to wait between retries; doubled each attempt
+_RETRY_BASE_DELAY = 2.0
+
 
 def _default_transport(request: Request):
-    return urlopen(request, timeout=60)
+    return urlopen(request, timeout=60)  # nosec B310 — URLs are env-configured API endpoints, not user input
+
+
+def _retry_delay(exc: Exception, attempt: int) -> float:
+    if isinstance(exc, HTTPError) and exc.code == 429:
+        retry_after = exc.headers.get("Retry-After")
+        if retry_after:
+            try:
+                return float(retry_after)
+            except ValueError:
+                pass
+    return _RETRY_BASE_DELAY * (2 ** attempt)
 
 
 @dataclass
 class JsonHttpClient:
     transport: Transport = _default_transport
 
+    def _execute(self, request: Request, url: str) -> str:
+        last_exc: Exception | None = None
+        for attempt in range(_MAX_RETRIES):
+            try:
+                with self.transport(request) as response:
+                    return response.read().decode("utf-8")
+            except HTTPError as exc:
+                if exc.code == 429:
+                    if attempt < _MAX_RETRIES - 1:
+                        time.sleep(_retry_delay(exc, attempt))
+                        last_exc = exc
+                        continue
+                    raise RateLimitError(f"rate limited after {_MAX_RETRIES} attempts for {url}") from exc
+                if exc.code >= 500 and attempt < _MAX_RETRIES - 1:
+                    time.sleep(_retry_delay(exc, attempt))
+                    last_exc = exc
+                    continue
+                raise ApiClientError(f"HTTP {exc.code} for {url}: {exc.reason}") from exc
+            except Exception as exc:  # pragma: no cover
+                if attempt < _MAX_RETRIES - 1:
+                    time.sleep(_retry_delay(exc, attempt))
+                    last_exc = exc
+                    continue
+                raise ApiClientError(f"request failed for {url}: {exc}") from exc
+        raise ApiClientError(f"request failed after {_MAX_RETRIES} attempts for {url}: {last_exc}") from last_exc
+
     def get_json(self, url: str, headers: dict[str, str], *, query: dict[str, str] | None = None) -> object:
         final_url = url
         if query:
             final_url = f"{url}?{urlencode(query, doseq=True)}"
         request = Request(final_url, headers=headers, method="GET")
-        try:
-            with self.transport(request) as response:
-                payload = response.read().decode("utf-8")
-        except Exception as exc:  # pragma: no cover
-            raise ApiClientError(f"request failed for {final_url}: {exc}") from exc
+        payload = self._execute(request, final_url)
         try:
             return json.loads(payload)
         except json.JSONDecodeError as exc:
@@ -46,11 +90,7 @@ def request_json(self, method: str, url: str, headers: dict[str, str], *, body:
             payload_bytes = json.dumps(body).encode("utf-8")
             request_headers.setdefault("Content-Type", "application/json")
         request = Request(url, headers=request_headers, data=payload_bytes, method=method)
-        try:
-            with self.transport(request) as response:
-                payload = response.read().decode("utf-8")
-        except Exception as exc:  # pragma: no cover
-            raise ApiClientError(f"request failed for {url}: {exc}") from exc
+        payload = self._execute(request, url)
         try:
             return json.loads(payload)
         except json.JSONDecodeError as exc:

dashboard_tooling/compare.py

@@ -2,9 +2,20 @@
 
 from difflib import SequenceMatcher
 
+from dashboard_tooling.heuristics import COMPLEXITY_BLOCKER_THRESHOLD
 from dashboard_tooling.models import DashboardRecord, ParityRecord
 from dashboard_tooling.normalize import normalize_title
 
+# SequenceMatcher ratio thresholds for dashboard title matching.
+# STRONG (0.92): titles that differ only by minor word reordering or punctuation
+#   e.g. "Service Latency Overview" vs "Service Latency - Overview"
+# WEAK (0.72): titles that share most tokens but have one meaningful difference
+#   e.g. "Host CPU Usage" vs "Host Memory Usage" — useful for flagging candidates,
+#   not for automatic matching.
+# Values below WEAK are treated as no match (missing_in_target).
+DEFAULT_STRONG_MATCH_THRESHOLD: float = 0.92
+DEFAULT_WEAK_MATCH_THRESHOLD: float = 0.72
+
 
 def _title_similarity(left: str, right: str) -> float:
     return SequenceMatcher(None, normalize_title(left), normalize_title(right)).ratio()
@@ -14,8 +25,8 @@ def compare_dashboards(
     source_dashboards: list[DashboardRecord],
     target_dashboards: list[DashboardRecord],
     *,
-    strong_match_threshold: float = 0.92,
-    weak_match_threshold: float = 0.72,
+    strong_match_threshold: float = DEFAULT_STRONG_MATCH_THRESHOLD,
+    weak_match_threshold: float = DEFAULT_WEAK_MATCH_THRESHOLD,
 ) -> list[ParityRecord]:
     parity_records: list[ParityRecord] = []
     for source in source_dashboards:
@@ -56,7 +67,7 @@ def compare_dashboards(
 
         if source.query_count == 0:
             record.manual_review_reasons.append("source_needs_manual_query_capture")
-        if source.complexity_score >= 12:
+        if source.complexity_score >= COMPLEXITY_BLOCKER_THRESHOLD:
             record.manual_review_reasons.append("high_complexity")
         if source.variables and not best_target:
             record.heuristic_blockers.append("dynamic_filter_mapping_required")

dashboard_tooling/deployment.py

@@ -2,7 +2,7 @@
 
 import json
 import shutil
-import subprocess
+import subprocess  # nosec B404
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -96,4 +96,4 @@ def run_terraform(work_dir: Path, command: str, *, auto_approve: bool = False) -
     args = [terraform, command]
     if command == "apply" and auto_approve:
         args.append("-auto-approve")
-    return subprocess.run(args, cwd=work_dir, check=True, capture_output=True, text=True)
+    return subprocess.run(args, cwd=work_dir, check=True, capture_output=True, text=True)  # nosec B603

dashboard_tooling/heuristics.py

@@ -4,6 +4,11 @@
 
 from dashboard_tooling.models import DashboardRecord, QueryRecord
 
+# Complexity score at or above this value triggers the high_complexity_dashboard blocker.
+# Score is: widget_count + query_count*2 + variable_count + (table_widget_count * 2).
+# A value of 12 corresponds roughly to a 4-widget, 3-query dashboard with one table widget.
+COMPLEXITY_BLOCKER_THRESHOLD = 12
+
 
 def infer_query_signals(query: QueryRecord) -> list[str]:
     text = query.query_text.lower()
@@ -18,6 +23,7 @@ def infer_query_signals(query: QueryRecord) -> list[str]:
         signals.append("result_shaping_logic")
     if any(token in text for token in ("timezone", "timestamp", "date_trunc", "bin(", "timeshift")):
         signals.append("time_semantics_risk")
+    # ddsql is Datadog-native and does not need translation; sql_like is Dynatrace USQL
     if query.query_family in {"sql_like", "dynatrace_dql"}:
         signals.append("translation_review_required")
     if "select " in text and " from " in text:
@@ -43,7 +49,7 @@ def infer_dashboard_blockers(dashboard: DashboardRecord) -> list[str]:
         blockers["table_or_list_result_shaping"] += 1
     if any(token in item.lower() for item in dashboard.widget_types for token in ("markdown", "note")):
         blockers["textual_context_dependency"] += 1
-    if dashboard.complexity_score >= 12:
+    if dashboard.complexity_score >= COMPLEXITY_BLOCKER_THRESHOLD:
         blockers["high_complexity_dashboard"] += 1
     return sorted(blockers)
 

dashboard_tooling/io.py

@@ -5,29 +5,49 @@
 from pathlib import Path
 
 
+class IoError(RuntimeError):
+    pass
+
+
 def load_json(path: Path) -> object:
-    with path.open("r", encoding="utf-8") as fh:
-        return json.load(fh)
+    if not path.exists():
+        raise IoError(f"file not found: {path}")
+    try:
+        with path.open("r", encoding="utf-8") as fh:
+            return json.load(fh)
+    except json.JSONDecodeError as exc:
+        raise IoError(f"invalid JSON in {path}: {exc}") from exc
+    except OSError as exc:
+        raise IoError(f"could not read {path}: {exc}") from exc
 
 
 def ensure_dir(path: Path) -> None:
     path.mkdir(parents=True, exist_ok=True)
 
 
 def write_json(path: Path, payload: object) -> None:
-    with path.open("w", encoding="utf-8") as fh:
-        json.dump(payload, fh, indent=2)
+    try:
+        with path.open("w", encoding="utf-8") as fh:
+            json.dump(payload, fh, indent=2)
+    except OSError as exc:
+        raise IoError(f"could not write {path}: {exc}") from exc
 
 
 def write_csv(path: Path, rows: list[dict[str, object]], fieldnames: list[str]) -> None:
-    with path.open("w", encoding="utf-8", newline="") as fh:
-        writer = csv.DictWriter(fh, fieldnames=fieldnames)
-        writer.writeheader()
-        for row in rows:
-            writer.writerow(row)
+    try:
+        with path.open("w", encoding="utf-8", newline="") as fh:
+            writer = csv.DictWriter(fh, fieldnames=fieldnames)
+            writer.writeheader()
+            for row in rows:
+                writer.writerow(row)
+    except OSError as exc:
+        raise IoError(f"could not write {path}: {exc}") from exc
 
 
 def write_text(path: Path, content: str) -> None:
-    with path.open("w", encoding="utf-8") as fh:
-        fh.write(content)
+    try:
+        with path.open("w", encoding="utf-8") as fh:
+            fh.write(content)
+    except OSError as exc:
+        raise IoError(f"could not write {path}: {exc}") from exc
 

dashboard_tooling/normalize.py

@@ -16,16 +16,30 @@ def normalize_title(value: str) -> str:
     return re.sub(r"[^a-z0-9]+", " ", value.lower()).strip()
 
 
-def query_family(query_text: str) -> str:
-    lowered = query_text.lower()
-    if "select " in lowered:
-        return "sql_like"
-    if "timeseries " in lowered or "fetch " in lowered:
+def query_family(query_text: str, source_system: str = "") -> str:
+    # Strip single-line and block comments before classifying to avoid matching
+    # keywords that appear only in comment text (e.g. "// timeseries" or "-- select").
+    stripped = re.sub(r"(--[^\n]*|//[^\n]*|/\*.*?\*/)", " ", query_text, flags=re.DOTALL).lower().strip()
+
+    # SQL-shaped query (SELECT ... FROM): the family depends on origin.
+    # - Dynatrace USQL (user sessions query language) looks like SQL but targets DT data.
+    # - Datadog DDSQL is Datadog-native and does not require translation.
+    # Both are flagged separately so downstream logic can treat them correctly.
+    if re.search(r"\bselect\b.+\bfrom\b", stripped, re.DOTALL):
+        return "ddsql" if source_system == "datadog" else "sql_like"
+
+    # Dynatrace DQL: timeseries or fetch must appear at the start of a statement
+    if re.search(r"(?:^|;|\|)\s*(?:timeseries|fetch)\b", stripped):
         return "dynatrace_dql"
-    if "avg:" in lowered or "sum:" in lowered or "anomalies(" in lowered:
+
+    # Datadog metric query syntax: metric selectors use colon notation
+    if re.search(r"\b(?:avg|sum|min|max|count):[a-z]", stripped) or "anomalies(" in stripped:
         return "datadog_metric"
-    if "logs(" in lowered or "service:" in lowered or "@http." in lowered:
+
+    # Datadog log/trace query syntax
+    if "logs(" in stripped or re.search(r"service:[a-z\-_*]", stripped) or "@http." in stripped:
         return "datadog_logs"
+
     return "unknown"
 
 
@@ -56,6 +70,7 @@ def _append_query(
     widget_title: str,
     widget_type: str,
     query_text: str,
+    source_system: str = "",
 ) -> None:
     cleaned = str(query_text).strip()
     if not cleaned:
@@ -68,7 +83,7 @@ def _append_query(
             widget_title=widget_title,
             widget_type=widget_type,
             query_text=cleaned,
-            query_family=query_family(cleaned),
+            query_family=query_family(cleaned, source_system),
             heuristic_signals=[],
         )
     )
@@ -135,6 +150,7 @@ def normalize_dynatrace_dashboards(payload: object) -> list[DashboardRecord]:
                             widget_title=widget_title,
                             widget_type=widget_type,
                             query_text=str(item),
+                            source_system="dynatrace",
                         )
                 elif path:
                     _append_query(
@@ -143,6 +159,7 @@ def normalize_dynatrace_dashboards(payload: object) -> list[DashboardRecord]:
                         widget_title=widget_title,
                         widget_type=widget_type,
                         query_text=str(path),
+                        source_system="dynatrace",
                     )
 
             if isinstance(widget.get("markdown"), str):
@@ -152,6 +169,7 @@ def normalize_dynatrace_dashboards(payload: object) -> list[DashboardRecord]:
                     widget_title=widget_title,
                     widget_type=widget_type,
                     query_text=str(widget["markdown"]),
+                    source_system="dynatrace",
                 )
 
         dashboard.widget_types = sorted(type_counter)
@@ -220,6 +238,7 @@ def normalize_datadog_dashboards(payload: object) -> list[DashboardRecord]:
                             widget_title=widget_title,
                             widget_type=widget_type,
                             query_text=str(request["q"]),
+                            source_system="datadog",
                         )
                     formulas = request.get("formulas")
                     if isinstance(formulas, list):
@@ -231,6 +250,7 @@ def normalize_datadog_dashboards(payload: object) -> list[DashboardRecord]:
                                     widget_title=widget_title,
                                     widget_type=widget_type,
                                     query_text=str(formula["formula"]),
+                                    source_system="datadog",
                                 )
             if definition.get("query"):
                 _append_query(
@@ -239,6 +259,7 @@ def normalize_datadog_dashboards(payload: object) -> list[DashboardRecord]:
                     widget_title=widget_title,
                     widget_type=widget_type,
                     query_text=str(definition["query"]),
+                    source_system="datadog",
                 )
 
         dashboard.widget_types = sorted(type_counter)