titan-iac/services/quality/sonarqube-exporter-configmap.yaml

# services/quality/sonarqube-exporter-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: sonarqube-exporter-script
  namespace: quality
data:
  exporter.py: |
    #!/usr/bin/env python3
    import base64
    import json
    import os
    import threading
    import time
    import urllib.error
    import urllib.parse
    import urllib.request
    from http.server import BaseHTTPRequestHandler, HTTPServer

    SONARQUBE_URL = os.getenv("SONARQUBE_URL", "http://sonarqube.quality.svc.cluster.local:9000").strip().rstrip("/")
    SONARQUBE_TOKEN = os.getenv("SONARQUBE_TOKEN", "").strip()
    SONARQUBE_TIMEOUT_SECONDS = float(os.getenv("SONARQUBE_TIMEOUT_SECONDS", "10"))
    SONARQUBE_EXPORTER_PORT = int(os.getenv("SONARQUBE_EXPORTER_PORT", "9798"))
    SONARQUBE_EXPORTER_CACHE_TTL_SECONDS = int(os.getenv("SONARQUBE_EXPORTER_CACHE_TTL_SECONDS", "45"))
    SONARQUBE_PROJECT_LIMIT = int(os.getenv("SONARQUBE_PROJECT_LIMIT", "200"))

    CACHE_LOCK = threading.Lock()
    CACHE_EXPIRES_AT = 0.0
    CACHE_BODY = ""


    def _escape(value: str) -> str:
      return value.replace("\\", "\\\\").replace("\"", "\\\"").replace("\n", "\\n")


    def _fetch_json(path: str):
      url = f"{SONARQUBE_URL}{path}"
      req = urllib.request.Request(url, method="GET")
      if SONARQUBE_TOKEN:
        encoded = base64.b64encode(f"{SONARQUBE_TOKEN}:".encode("utf-8")).decode("utf-8")
        req.add_header("Authorization", f"Basic {encoded}")
      try:
        with urllib.request.urlopen(req, timeout=SONARQUBE_TIMEOUT_SECONDS) as resp:
          payload = json.loads(resp.read().decode("utf-8"))
          return payload, ""
      except urllib.error.HTTPError as exc:
        return None, f"http_{exc.code}"
      except Exception as exc:  # noqa: BLE001
        return None, exc.__class__.__name__


    def _metrics_body() -> str:
      lines = []
      now = time.time()
      scrape_success = 1

      lines.append("# HELP sonarqube_exporter_last_scrape_timestamp_seconds Unix timestamp when exporter last refreshed data.")
      lines.append("# TYPE sonarqube_exporter_last_scrape_timestamp_seconds gauge")
      lines.append(f"sonarqube_exporter_last_scrape_timestamp_seconds {now:.3f}")

      system_payload, system_error = _fetch_json("/api/system/status")
      system_status = "unknown"
      sonarqube_up = 0
      if isinstance(system_payload, dict):
        system_status = str(system_payload.get("status") or "unknown")
      elif system_error:
        system_status = system_error
        scrape_success = 0

      if system_status.upper() in {
        "UP",
        "STARTING",
        "DB_MIGRATION_NEEDED",
        "DB_MIGRATION_RUNNING",
      }:
        sonarqube_up = 1

      lines.append("# HELP sonarqube_up SonarQube API reachability and health (1=reachable/healthy-ish, 0=down).")
      lines.append("# TYPE sonarqube_up gauge")
      lines.append(f"sonarqube_up {sonarqube_up}")

      lines.append("# HELP sonarqube_system_status Current SonarQube system status label.")
      lines.append("# TYPE sonarqube_system_status gauge")
      lines.append(f'sonarqube_system_status{{status="{_escape(system_status)}"}} 1')

      projects_payload, projects_error = _fetch_json("/api/projects/search?ps=500&p=1")
      project_items = []
      projects_total = 0
      if isinstance(projects_payload, dict):
        paging = projects_payload.get("paging") or {}
        projects_total = int(paging.get("total") or 0)
        project_items = list(projects_payload.get("components") or [])
      else:
        scrape_success = 0

      lines.append("# HELP sonarqube_projects_total Total discovered SonarQube projects.")
      lines.append("# TYPE sonarqube_projects_total gauge")
      lines.append(f"sonarqube_projects_total {projects_total}")

      gate_counts = {}
      gate_fetch_errors = 0
      inspected = 0
      project_samples = []
      for project in project_items:
        if inspected >= SONARQUBE_PROJECT_LIMIT:
          break
        key = str(project.get("key") or "").strip()
        if not key:
          continue
        inspected += 1

        gate_payload, gate_error = _fetch_json(
          "/api/qualitygates/project_status?projectKey=" + urllib.parse.quote_plus(key)
        )
        if not isinstance(gate_payload, dict):
          gate_fetch_errors += 1
          continue
        project_status = gate_payload.get("projectStatus") or {}
        gate_status = str(project_status.get("status") or "UNKNOWN").upper()
        gate_counts[gate_status] = gate_counts.get(gate_status, 0) + 1
        is_ok = 1 if gate_status == "OK" else 0
        project_samples.append(
          f'sonarqube_project_quality_gate_pass{{project_key="{_escape(key)}",status="{_escape(gate_status)}"}} {is_ok}'
        )

      lines.append("# HELP sonarqube_project_quality_gate_pass Project quality gate pass state (1=OK, 0=not OK).")
      lines.append("# TYPE sonarqube_project_quality_gate_pass gauge")
      lines.extend(project_samples)
      lines.append("# HELP sonarqube_quality_gate_projects_total Number of projects by quality gate status.")
      lines.append("# TYPE sonarqube_quality_gate_projects_total gauge")
      for status, count in sorted(gate_counts.items()):
        lines.append(f'sonarqube_quality_gate_projects_total{{status="{_escape(status)}"}} {count}')

      lines.append("# HELP sonarqube_quality_gate_fetch_errors_total Number of project gate API fetch failures in the last scrape.")
      lines.append("# TYPE sonarqube_quality_gate_fetch_errors_total gauge")
      lines.append(f"sonarqube_quality_gate_fetch_errors_total {gate_fetch_errors}")

      lines.append("# HELP sonarqube_exporter_scrape_success Exporter scrape success (1=success, 0=partial/error).")
      lines.append("# TYPE sonarqube_exporter_scrape_success gauge")
      lines.append(f"sonarqube_exporter_scrape_success {scrape_success}")

      if projects_error:
        lines.append("# HELP sonarqube_exporter_projects_error Indicates projects API failure on the most recent scrape.")
        lines.append("# TYPE sonarqube_exporter_projects_error gauge")
        lines.append(f'sonarqube_exporter_projects_error{{error="{_escape(projects_error)}"}} 1')

      return "\n".join(lines) + "\n"


    def _get_metrics() -> str:
      global CACHE_BODY, CACHE_EXPIRES_AT
      now = time.time()
      with CACHE_LOCK:
        if CACHE_BODY and now < CACHE_EXPIRES_AT:
          return CACHE_BODY
        CACHE_BODY = _metrics_body()
        CACHE_EXPIRES_AT = now + max(5, SONARQUBE_EXPORTER_CACHE_TTL_SECONDS)
        return CACHE_BODY


    class Handler(BaseHTTPRequestHandler):
      def do_GET(self):  # noqa: N802
        if self.path in ("/-/healthy", "/healthz"):
          body = b"ok\n"
          self.send_response(200)
          self.send_header("Content-Type", "text/plain; charset=utf-8")
          self.send_header("Content-Length", str(len(body)))
          self.end_headers()
          self.wfile.write(body)
          return
        if self.path == "/metrics":
          body = _get_metrics().encode("utf-8")
          self.send_response(200)
          self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
          self.send_header("Content-Length", str(len(body)))
          self.end_headers()
          self.wfile.write(body)
          return
        self.send_response(404)
        self.end_headers()

      def log_message(self, fmt, *args):  # noqa: A003
        return


    def main():
      server = HTTPServer(("0.0.0.0", SONARQUBE_EXPORTER_PORT), Handler)
      server.serve_forever()


    if __name__ == "__main__":
      main()