211 lines
6.9 KiB
Python
211 lines
6.9 KiB
Python
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}")
|
|
|
|
|