# 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()