from prometheus_client import Counter, Gauge, Histogram, generate_latest, CONTENT_TYPE_LATEST from typing import Any, Callable, Dict, Optional, Set, Tuple import threading from datetime import datetime import time import uuid from flask import Flask, jsonify, request class Metrics: _instance = None _lock = threading.Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super(Metrics, cls).__new__(cls) cls._instance._initialize_metrics() return cls._instance def _initialize_metrics(self): # Total logs written counter per cluster self.logs_written = Counter( 'slowlogconverter_logs_written_total', 'Total number of logs written to DeltaLake', ['cluster'] ) self.logs_total = Counter( 'slowlogconverter_logs_total', 'Total number of logs', ['cluster'] ) self.logs_parse_error = Counter( 'slowlogconverter_logs_parse_error_total', 'Total number of logs parse error', ['cluster'] ) self.logs_written_1m = Counter( 'slowlogconverter_logs_written_1m', 'Total number of logs written to DeltaLake in last 1 minute', ['cluster'] ) # Oldest processed hour per cluster (as timestamp) self.processed_hour = Gauge( 'slowlogconverter_oldest_processed_hour', 'Oldest processed UTC hour as Unix timestamp', ['cluster'] ) # Last update timestamp self.last_update = Gauge( 'slowlogconverter_metrics_last_update', 'Timestamp of the last metrics update', ['cluster'] ) self.latest_record_timestamp = Gauge( 'slowlogconverter_latest_record_timestamp', 'Timestamp of the latest record', ['cluster'] ) self.last_optimize_timestamp = Gauge( 'slowlogconverter_metrics_last_optimize_timestamp', 'Timestamp of the last optimize', ['cluster'] ) self.optimize_pending = Gauge( 'slowlogconverter_optimize_pending', 'Whether optimize task is pending in queue (0/1)', ['cluster'] ) self.optimize_running = Gauge( 'slowlogconverter_optimize_running', 'Whether optimize task is currently running (0/1)', ['cluster'] ) self.optimize_pending_global = Gauge( 'slowlogconverter_optimize_pending_global', 'Total number of pending optimize tasks across clusters' ) self.optimize_running_global = Gauge( 'slowlogconverter_optimize_running_global', 'Total number of running optimize tasks across clusters' ) self.optimize_scheduled = Counter( 'slowlogconverter_optimize_scheduled_total', 'Total number of optimize tasks scheduled', ['cluster'] ) self.optimize_skipped = Counter( 'slowlogconverter_optimize_skipped_total', 'Total number of optimize schedule attempts skipped', ['cluster', 'reason'] ) self.optimize_completed = Counter( 'slowlogconverter_optimize_completed_total', 'Total number of optimize tasks completed', ['cluster', 'status'] ) self.optimize_duration_seconds = Histogram( 'slowlogconverter_optimize_duration_seconds', 'Optimize task execution duration in seconds', ['cluster', 'status'], buckets=(1, 3, 5, 10, 30, 60, 120, 300, 600, 1200) ) # Track the oldest hour for each cluster self._cluster_hours: Dict[str, int] = {} self._lock = threading.Lock() self._backfill_handler: Optional[Callable[[Dict[str, Any]], None]] = None self._backfill_jobs: Dict[str, Dict[str, Any]] = {} self._backfill_lock = threading.Lock() # Initialize the diagnostic server self._start_diagnostic_server() def record_logs_written(self, count: int, cluster: str, utc_hour: str, latest_record_timestamp: int, optimize_timestamp: int, parse_error_count: int, last_update_timestamp = None): """Record logs written to DeltaLake and track the oldest hour.""" # Convert utc_hour string to timestamp (format: YYYYMMDDHH) try: hour_ts = int(datetime.strptime(utc_hour, '%Y%m%d%H').timestamp()) except ValueError: return with self._lock: # Update logs written counter if last_update_timestamp is None: last_update_timestamp = time.time() self.last_update.labels(cluster=cluster).set(last_update_timestamp) self.logs_written.labels(cluster=cluster).inc(count) self.logs_written_1m.labels(cluster=cluster).inc(count) self._cluster_hours[cluster] = hour_ts self.processed_hour.labels(cluster=cluster).set(hour_ts) self.last_optimize_timestamp.labels(cluster=cluster).set(optimize_timestamp) self.latest_record_timestamp.labels(cluster=cluster).set(latest_record_timestamp) self.logs_parse_error.labels(cluster=cluster).inc(parse_error_count) self.logs_total.labels(cluster=cluster).inc(count + parse_error_count) def update_optimize_state(self, cluster: str, pending: int, running: int, pending_global: int, running_global: int): with self._lock: self.optimize_pending.labels(cluster=cluster).set(pending) self.optimize_running.labels(cluster=cluster).set(running) self.optimize_pending_global.set(pending_global) self.optimize_running_global.set(running_global) def record_optimize_scheduled(self, cluster: str): with self._lock: self.optimize_scheduled.labels(cluster=cluster).inc() def record_optimize_skipped(self, cluster: str, reason: str): with self._lock: self.optimize_skipped.labels(cluster=cluster, reason=reason).inc() def record_optimize_completed(self, cluster: str, status: str, duration_seconds: float): with self._lock: self.optimize_completed.labels(cluster=cluster, status=status).inc() self.optimize_duration_seconds.labels(cluster=cluster, status=status).observe(duration_seconds) def register_backfill_handler(self, handler: Callable[[Dict[str, Any]], None]): with self._backfill_lock: self._backfill_handler = handler def _set_backfill_job(self, job_id: str, **fields): with self._backfill_lock: if job_id not in self._backfill_jobs: self._backfill_jobs[job_id] = {} self._backfill_jobs[job_id].update(fields) def _get_backfill_job(self, job_id: str) -> Optional[Dict[str, Any]]: with self._backfill_lock: job = self._backfill_jobs.get(job_id) if job is None: return None return dict(job) def _start_diagnostic_server(self): """Start a Flask server for metrics and diagnostic endpoints.""" app = Flask(__name__) @app.route('/metrics') def metrics(): return generate_latest(), 200, {'Content-Type': CONTENT_TYPE_LATEST} @app.route('/diagnostics/cluster_hours', methods=['GET']) def get_cluster_hours(): """Get all cluster hours.""" with self._lock: return jsonify({ 'status': 'success', 'data': self._cluster_hours }) @app.route('/diagnostics/cluster_hours/', methods=['GET']) def get_cluster_hour(cluster: str): """Get hour for a specific cluster.""" with self._lock: if cluster not in self._cluster_hours: return jsonify({ 'status': 'error', 'message': f'Cluster {cluster} not found' }), 404 return jsonify({ 'status': 'success', 'data': { 'cluster': cluster, 'hour_timestamp': self._cluster_hours[cluster], 'hour_human': datetime.utcfromtimestamp(self._cluster_hours[cluster]).strftime('%Y-%m-%d %H:00:00 UTC') } }) @app.route('/diagnostics/cluster_hours/clean', methods=['POST']) def clean_cluster_hours(): """Clean up cluster hours.""" data = request.get_json() or {} clusters = data.get('clusters', []) # List of cluster names to clean force = data.get('force', False) # If True, clean all if no clusters specified with self._lock: if not clusters: if not force: return jsonify({ 'status': 'error', 'message': 'No clusters specified. Use force=true to clean all clusters.' }), 400 # Clean all clusters cleaned = list(self._cluster_hours.keys()) self._cluster_hours.clear() else: # Clean only specified clusters cleaned = [] for cluster in clusters: if cluster in self._cluster_hours: del self._cluster_hours[cluster] cleaned.append(cluster) return jsonify({ 'status': 'success', 'cleaned_clusters': cleaned, 'remaining_clusters': list(self._cluster_hours.keys()) }) @app.route('/diagnostics/backfill', methods=['POST']) def trigger_backfill(): payload = request.get_json() or {} cluster = payload.get('cluster') start_ts = payload.get('start_ts') end_ts = payload.get('end_ts') required_runtime_fields = ['provider', 'region', 'tenant', 'project', 'o11y_cluster', 'env', 'cluster_type'] if not cluster: return jsonify({'status': 'error', 'message': 'cluster is required'}), 400 if start_ts is None or end_ts is None: return jsonify({'status': 'error', 'message': 'start_ts and end_ts are required'}), 400 missing_runtime = [k for k in required_runtime_fields if payload.get(k) is None] if missing_runtime: return jsonify({'status': 'error', 'message': f"missing required fields: {', '.join(missing_runtime)}"}), 400 try: start_ts = int(start_ts) end_ts = int(end_ts) except Exception: return jsonify({'status': 'error', 'message': 'start_ts/end_ts must be integers'}), 400 if start_ts >= end_ts: return jsonify({'status': 'error', 'message': 'start_ts must be less than end_ts'}), 400 with self._backfill_lock: handler = self._backfill_handler if handler is None: return jsonify({'status': 'error', 'message': 'backfill handler not registered'}), 503 job_id = uuid.uuid4().hex self._set_backfill_job( job_id, cluster=cluster, start_ts=start_ts, end_ts=end_ts, status='running', created_at=time.time(), ) def _run(): try: handler(payload) self._set_backfill_job(job_id, status='success', finished_at=time.time()) except Exception as e: self._set_backfill_job(job_id, status='failed', error=str(e), finished_at=time.time()) threading.Thread(target=_run, daemon=True).start() return jsonify({'status': 'success', 'job_id': job_id}), 202 @app.route('/diagnostics/backfill/', methods=['GET']) def get_backfill_job(job_id: str): job = self._get_backfill_job(job_id) if job is None: return jsonify({'status': 'error', 'message': f'job {job_id} not found'}), 404 return jsonify({'status': 'success', 'data': job}) # Start the server in a background thread def run_server(): app.run(host='0.0.0.0', port=8000) server_thread = threading.Thread(target=run_server, daemon=True) server_thread.start() def get_oldest_processed_hour(self) -> Optional[Tuple[str, int]]: """Get the cluster with the oldest processed hour and its timestamp. Returns: Tuple of (cluster, hour_timestamp) or None if no data """ with self._lock: if not self._cluster_hours: return None # Find the cluster with the oldest hour oldest_cluster = min(self._cluster_hours.items(), key=lambda x: x[1]) return oldest_cluster def last_update_timestamp_for_cluster(self, cluster: str) -> Optional[float]: """Last metrics update time for cluster, or None if no series exists.""" for family in self.last_update.collect(): for sample in family.samples: if sample.labels.get("cluster") == cluster: return float(sample.value) return None def total_logs_written_1m(self) -> int: """Sum counter values across all cluster labels (excludes _created samples).""" total = 0 for family in self.logs_written_1m.collect(): for sample in family.samples: if sample.name.endswith("_created"): continue total += int(sample.value) return total def reset_logs_written_1m(self) -> None: """Clear all labeled series for the 1-minute logs counter.""" labelnames = getattr(self.logs_written_1m, "_labelnames", ()) seen: Set[Tuple[str, ...]] = set() for family in self.logs_written_1m.collect(): for sample in family.samples: if not sample.labels: continue key = tuple(sample.labels.get(ln) for ln in labelnames) if None in key or key in seen: continue seen.add(key) try: self.logs_written_1m.remove(*key) except KeyError: pass def cluster_has_last_update_series(self, cluster: str) -> bool: """Whether last_update has a time series for this cluster.""" return self.last_update_timestamp_for_cluster(cluster) is not None # Global metrics instance metrics = Metrics()