from prometheus_client import Counter, Histogram, Gauge, start_http_server import logging from typing import Optional logger = logging.getLogger(__name__) # Counters # Ingest is counted once, by SOURCE_ACTIVITY_TOTAL, which also carries the channel title. SOURCE_ACTIVITY_TOTAL = Counter( "copykar_source_activity_total", "Total posts ingested per source channel", ["channel_id", "title"] ) TARGET_ACTIVITY_TOTAL = Counter( "copykar_target_activity_total", "Total posts published per target channel", ["channel_id", "title"] ) AI_REQUESTS_TOTAL = Counter( "copykar_ai_requests_total", "Total AI API calls made", ["action", "status"] ) DUPLICATES_DETECTED_TOTAL = Counter( "copykar_duplicates_detected_total", "Total duplicate posts detected", ["method"] ) ADMIN_ACTIONS_TOTAL = Counter( "copykar_admin_actions_total", "Total review decisions by admins", ["action"] ) # Delivery is counted once, by TARGET_ACTIVITY_TOTAL. AUTO_ROUTED_POSTS_TOTAL = Counter( "copykar_auto_routed_posts_total", "Posts queued automatically by a source-to-target route", ["source_channel_id", "target_title"] ) ERRORS_TOTAL = Counter( "copykar_errors_total", "Total exceptions and errors caught across services", ["service", "error_type"] ) ERRORS_RESOLVED_TOTAL = Counter( "copykar_errors_resolved_total", "Errors an admin has marked as fixed", ["service", "error_type"] ) # Histograms AI_LATENCY_SECONDS = Histogram( "copykar_ai_latency_seconds", "Time taken for AI API operations", ["action"], buckets=(0.5, 1.0, 2.0, 3.0, 5.0, 8.0, 12.0, 20.0, 30.0, 45.0, 60.0, 90.0, 120.0, 180.0) ) # Gauges QUEUE_POSTS_GAUGE = Gauge( "copykar_posts_queue_gauge", "Number of posts currently in various queue states", ["status"] ) ERRORS_OPEN_GAUGE = Gauge( "copykar_errors_open", "Unresolved errors currently recorded, by service and exception type", ["service", "error_type"] ) ERRORS_OPEN_TOTAL_GAUGE = Gauge( "copykar_errors_open_total", "Total unresolved errors across all services" ) import shutil import asyncio # Disk Space Gauges per Mount Point DISK_TOTAL_BYTES = Gauge( "copykar_disk_total_bytes", "Total disk capacity in bytes", ["mountpoint", "device"] ) DISK_USED_BYTES = Gauge( "copykar_disk_used_bytes", "Used disk space in bytes", ["mountpoint", "device"] ) DISK_FREE_BYTES = Gauge( "copykar_disk_free_bytes", "Free/available disk space in bytes", ["mountpoint", "device"] ) DISK_FREE_PERCENT = Gauge( "copykar_disk_free_percent", "Percentage of free disk space", ["mountpoint", "device"] ) import os VOLUME_DEFINITIONS = [ { "mountpoint": "/", "device": "/dev/sda1", "container_paths": ["/host_os/home", "/home"] }, { "mountpoint": "/projects", "device": "/dev/sda2", "container_paths": ["/host_os/projects", "/projects"] } ] def update_disk_metrics(): """Update Prometheus gauges with real host OS filesystem disk usage for verified mounted volumes.""" try: recorded_mounts = set() for v in VOLUME_DEFINITIONS: mnt = v["mountpoint"] dev = v["device"] for path in v["container_paths"]: if os.path.exists(path) and os.path.isdir(path): try: total, used, free = shutil.disk_usage(path) DISK_TOTAL_BYTES.labels(mountpoint=mnt, device=dev).set(total) DISK_USED_BYTES.labels(mountpoint=mnt, device=dev).set(used) DISK_FREE_BYTES.labels(mountpoint=mnt, device=dev).set(free) free_pct = (free / total * 100.0) if total > 0 else 0.0 DISK_FREE_PERCENT.labels(mountpoint=mnt, device=dev).set(free_pct) recorded_mounts.add(mnt) break except Exception as err: logger.debug(f"Error measuring disk usage for {path}: {err}") # Dynamically check any mounted external media drives under /host_os/media media_root = "/host_os/media" if os.path.exists(media_root) and os.path.isdir(media_root): try: for user in os.listdir(media_root): user_dir = os.path.join(media_root, user) if os.path.isdir(user_dir): for drive in os.listdir(user_dir): drive_path = os.path.join(user_dir, drive) if os.path.isdir(drive_path): total, used, free = shutil.disk_usage(drive_path) mnt_label = f"/run/media/{user}/{drive}" DISK_TOTAL_BYTES.labels(mountpoint=mnt_label, device="external").set(total) DISK_USED_BYTES.labels(mountpoint=mnt_label, device="external").set(used) DISK_FREE_BYTES.labels(mountpoint=mnt_label, device="external").set(free) free_pct = (free / total * 100.0) if total > 0 else 0.0 DISK_FREE_PERCENT.labels(mountpoint=mnt_label, device="external").set(free_pct) except Exception as e: logger.debug(f"Error scanning external media mounts: {e}") # If neither specific mount matched, fallback to root if not recorded_mounts: total, used, free = shutil.disk_usage("/") DISK_TOTAL_BYTES.labels(mountpoint="/", device="/dev/sda1").set(total) DISK_USED_BYTES.labels(mountpoint="/", device="/dev/sda1").set(used) DISK_FREE_BYTES.labels(mountpoint="/", device="/dev/sda1").set(free) free_pct = (free / total * 100.0) if total > 0 else 0.0 DISK_FREE_PERCENT.labels(mountpoint="/", device="/dev/sda1").set(free_pct) except Exception as e: logger.debug(f"Error updating host disk metrics: {e}") async def _disk_metrics_loop(interval: int = 15): """Background task to keep disk metrics updated.""" while True: try: update_disk_metrics() except Exception as e: logger.debug(f"Disk metrics loop error: {e}") await asyncio.sleep(interval) # The overall queue depth is sum(copykar_posts_queue_gauge) - no separate total series, # which previously made the dashboard report the queue twice. def start_metrics_server(port: int = 8008): try: start_http_server(port) update_disk_metrics() try: loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task(_disk_metrics_loop()) except Exception: pass logger.info(f"Prometheus metrics server running on port {port}") except Exception as e: logger.error(f"Failed to start Prometheus metrics server: {e}")