"""Prometheus metrics endpoint for external monitoring.""" import platform import secrets from fastapi import APIRouter, Depends, Header, HTTPException, Response from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from backend.app.core.config import APP_VERSION from backend.app.core.database import get_db from backend.app.models.print_log import PrintLogEntry from backend.app.models.print_queue import PrintQueueItem from backend.app.models.printer import Printer from backend.app.models.settings import Settings from backend.app.services.printer_manager import printer_manager, supports_chamber_temp router = APIRouter(tags=["metrics"]) async def get_prometheus_settings(db: AsyncSession) -> tuple[bool, str]: """Get Prometheus settings from database.""" result = await db.execute(select(Settings).where(Settings.key.in_(["prometheus_enabled", "prometheus_token"]))) settings_dict = {s.key: s.value for s in result.scalars().all()} enabled = settings_dict.get("prometheus_enabled", "false").lower() == "true" token = settings_dict.get("prometheus_token", "") return enabled, token def format_labels(**labels: str) -> str: """Format label key-value pairs for Prometheus.""" if not labels: return "" pairs = [f'{k}="{v}"' for k, v in labels.items() if v is not None] return "{" + ",".join(pairs) + "}" def state_to_numeric(state: str) -> int: """Convert printer state string to numeric value.""" state_map = { "unknown": 0, "IDLE": 1, "RUNNING": 2, "PAUSE": 3, "FINISH": 4, "FAILED": 5, "PREPARE": 6, "SLICING": 7, } return state_map.get(state, 0) @router.get("/metrics", response_class=Response) async def get_metrics( db: AsyncSession = Depends(get_db), authorization: str | None = Header(None), ): """ Prometheus metrics endpoint. Returns metrics in Prometheus text exposition format. Requires prometheus_enabled setting to be true. If prometheus_token is set, requires Bearer token authentication. """ # Check if enabled enabled, token = await get_prometheus_settings(db) if not enabled: raise HTTPException(status_code=404, detail="Prometheus metrics not enabled") # Check authentication if token is set if token: if not authorization: raise HTTPException(status_code=401, detail="Authorization required") if not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="Bearer token required") provided_token = authorization[7:] # Remove "Bearer " prefix # Constant-time comparison closes the byte-by-byte timing oracle that # plain ``!=`` opens on a LAN-attached attacker (audit finding I2). if not secrets.compare_digest(provided_token.encode("utf-8"), token.encode("utf-8")): raise HTTPException(status_code=401, detail="Invalid token") lines: list[str] = [] # ========================================================================= # Build info # ========================================================================= lines.append("# HELP bambuddy_build_info Build and version information") lines.append("# TYPE bambuddy_build_info gauge") build_labels = format_labels( version=APP_VERSION, python_version=platform.python_version(), platform=platform.system(), architecture=platform.machine(), ) lines.append(f"bambuddy_build_info{build_labels} 1") # ========================================================================= # Printer metrics # ========================================================================= # Get all printers from DB result = await db.execute(select(Printer).where(Printer.is_active == True)) # noqa: E712 printers = list(result.scalars().all()) # Build lookup for printer info printer_info = {p.id: p for p in printers} # Get all connected printer statuses all_statuses = printer_manager.get_all_statuses() # Printer connection status lines.append("# HELP bambuddy_printer_connected Printer connection status (1=connected, 0=disconnected)") lines.append("# TYPE bambuddy_printer_connected gauge") for printer in printers: status = all_statuses.get(printer.id) connected = 1 if status and status.connected else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, model=printer.model or "unknown", ) lines.append(f"bambuddy_printer_connected{labels} {connected}") # Printer state lines.append("") lines.append( "# HELP bambuddy_printer_state Printer state (0=unknown, 1=idle, 2=running, 3=pause, 4=finish, 5=failed, 6=prepare, 7=slicing)" ) lines.append("# TYPE bambuddy_printer_state gauge") for printer in printers: status = all_statuses.get(printer.id) state_val = state_to_numeric(status.state) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_printer_state{labels} {state_val}") # Print progress lines.append("") lines.append("# HELP bambuddy_print_progress Current print progress (0-100)") lines.append("# TYPE bambuddy_print_progress gauge") for printer in printers: status = all_statuses.get(printer.id) progress = status.progress if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_print_progress{labels} {progress:.1f}") # Remaining time lines.append("") lines.append("# HELP bambuddy_print_remaining_seconds Estimated remaining print time in seconds") lines.append("# TYPE bambuddy_print_remaining_seconds gauge") for printer in printers: status = all_statuses.get(printer.id) remaining = status.remaining_time * 60 if status else 0 # Convert minutes to seconds labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_print_remaining_seconds{labels} {remaining}") # Layer progress lines.append("") lines.append("# HELP bambuddy_print_layer_current Current layer number") lines.append("# TYPE bambuddy_print_layer_current gauge") for printer in printers: status = all_statuses.get(printer.id) layer = status.layer_num if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_print_layer_current{labels} {layer}") lines.append("") lines.append("# HELP bambuddy_print_layer_total Total layers in current print") lines.append("# TYPE bambuddy_print_layer_total gauge") for printer in printers: status = all_statuses.get(printer.id) total = status.total_layers if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_print_layer_total{labels} {total}") # ========================================================================= # Temperature metrics # ========================================================================= lines.append("") lines.append("# HELP bambuddy_bed_temp_celsius Current bed temperature") lines.append("# TYPE bambuddy_bed_temp_celsius gauge") for printer in printers: status = all_statuses.get(printer.id) temp = status.temperatures.get("bed", 0) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_bed_temp_celsius{labels} {temp:.1f}") lines.append("") lines.append("# HELP bambuddy_bed_target_celsius Target bed temperature") lines.append("# TYPE bambuddy_bed_target_celsius gauge") for printer in printers: status = all_statuses.get(printer.id) temp = status.temperatures.get("bed_target", 0) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_bed_target_celsius{labels} {temp:.1f}") lines.append("") lines.append("# HELP bambuddy_nozzle_temp_celsius Current nozzle temperature") lines.append("# TYPE bambuddy_nozzle_temp_celsius gauge") for printer in printers: status = all_statuses.get(printer.id) # Primary nozzle temp = status.temperatures.get("nozzle", 0) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, nozzle="0", ) lines.append(f"bambuddy_nozzle_temp_celsius{labels} {temp:.1f}") # Second nozzle if present if status and "nozzle_2" in status.temperatures: temp2 = status.temperatures.get("nozzle_2", 0) labels2 = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, nozzle="1", ) lines.append(f"bambuddy_nozzle_temp_celsius{labels2} {temp2:.1f}") lines.append("") lines.append("# HELP bambuddy_nozzle_target_celsius Target nozzle temperature") lines.append("# TYPE bambuddy_nozzle_target_celsius gauge") for printer in printers: status = all_statuses.get(printer.id) temp = status.temperatures.get("nozzle_target", 0) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, nozzle="0", ) lines.append(f"bambuddy_nozzle_target_celsius{labels} {temp:.1f}") if status and "nozzle_2_target" in status.temperatures: temp2 = status.temperatures.get("nozzle_2_target", 0) labels2 = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, nozzle="1", ) lines.append(f"bambuddy_nozzle_target_celsius{labels2} {temp2:.1f}") lines.append("") lines.append( "# HELP bambuddy_chamber_temp_celsius Current chamber temperature (only for models with chamber sensor)" ) lines.append("# TYPE bambuddy_chamber_temp_celsius gauge") for printer in printers: # Only report chamber temp for models that have a real sensor if not supports_chamber_temp(printer.model): continue status = all_statuses.get(printer.id) temp = status.temperatures.get("chamber", 0) if status else 0 labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_chamber_temp_celsius{labels} {temp:.1f}") # ========================================================================= # Fan speeds # ========================================================================= lines.append("") lines.append("# HELP bambuddy_fan_speed_percent Fan speed percentage") lines.append("# TYPE bambuddy_fan_speed_percent gauge") for printer in printers: status = all_statuses.get(printer.id) if not status: continue # Part cooling fan if "part_fan" in status.temperatures: val = status.temperatures["part_fan"] labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, fan="part", ) lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}") # Aux fan if "aux_fan" in status.temperatures: val = status.temperatures["aux_fan"] labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, fan="aux", ) lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}") # Chamber fan if "chamber_fan" in status.temperatures: val = status.temperatures["chamber_fan"] labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, fan="chamber", ) lines.append(f"bambuddy_fan_speed_percent{labels} {val:.1f}") # ========================================================================= # WiFi signal # ========================================================================= lines.append("") lines.append("# HELP bambuddy_wifi_signal_dbm WiFi signal strength in dBm") lines.append("# TYPE bambuddy_wifi_signal_dbm gauge") for printer in printers: status = all_statuses.get(printer.id) if status and status.wifi_signal is not None: labels = format_labels( printer_id=str(printer.id), printer_name=printer.name, serial=printer.serial_number, ) lines.append(f"bambuddy_wifi_signal_dbm{labels} {status.wifi_signal}") # ========================================================================= # Print statistics (from database) # ========================================================================= # Total prints by status — count print events from PrintLogEntry so # reprints contribute new rows instead of overwriting the source archive # (#1378). lines.append("") lines.append("# HELP bambuddy_prints_total Total number of prints by result") lines.append("# TYPE bambuddy_prints_total counter") result = await db.execute(select(PrintLogEntry.status, func.count(PrintLogEntry.id)).group_by(PrintLogEntry.status)) for print_result, count in result.all(): result_label = print_result or "unknown" labels = format_labels(result=result_label) lines.append(f"bambuddy_prints_total{labels} {count}") # Total prints per printer lines.append("") lines.append("# HELP bambuddy_printer_prints_total Total prints per printer") lines.append("# TYPE bambuddy_printer_prints_total counter") result = await db.execute( select(PrintLogEntry.printer_id, func.count(PrintLogEntry.id)).group_by(PrintLogEntry.printer_id) ) for printer_id, count in result.all(): if printer_id and printer_id in printer_info: p = printer_info[printer_id] labels = format_labels( printer_id=str(printer_id), printer_name=p.name, serial=p.serial_number, ) lines.append(f"bambuddy_printer_prints_total{labels} {count}") # Total filament used — sum per-run actuals from PrintLogEntry. lines.append("") lines.append("# HELP bambuddy_filament_used_grams Total filament used in grams") lines.append("# TYPE bambuddy_filament_used_grams counter") result = await db.execute(select(func.coalesce(func.sum(PrintLogEntry.filament_used_grams), 0))) total_filament = result.scalar() or 0 lines.append(f"bambuddy_filament_used_grams {total_filament:.1f}") # Total print time — sum per-run elapsed durations. lines.append("") lines.append("# HELP bambuddy_print_time_seconds Total print time in seconds") lines.append("# TYPE bambuddy_print_time_seconds counter") result = await db.execute(select(func.coalesce(func.sum(PrintLogEntry.duration_seconds), 0))) total_time = result.scalar() or 0 lines.append(f"bambuddy_print_time_seconds {total_time}") # ========================================================================= # Queue metrics # ========================================================================= lines.append("") lines.append("# HELP bambuddy_queue_pending Number of pending queue items") lines.append("# TYPE bambuddy_queue_pending gauge") result = await db.execute(select(func.count(PrintQueueItem.id)).where(PrintQueueItem.status == "pending")) pending_count = result.scalar() or 0 lines.append(f"bambuddy_queue_pending {pending_count}") lines.append("") lines.append("# HELP bambuddy_queue_printing Number of currently printing queue items") lines.append("# TYPE bambuddy_queue_printing gauge") result = await db.execute(select(func.count(PrintQueueItem.id)).where(PrintQueueItem.status == "printing")) printing_count = result.scalar() or 0 lines.append(f"bambuddy_queue_printing {printing_count}") # ========================================================================= # System metrics # ========================================================================= lines.append("") lines.append("# HELP bambuddy_printers_connected Number of connected printers") lines.append("# TYPE bambuddy_printers_connected gauge") connected_count = sum(1 for s in all_statuses.values() if s.connected) lines.append(f"bambuddy_printers_connected {connected_count}") lines.append("") lines.append("# HELP bambuddy_printers_total Total number of configured printers") lines.append("# TYPE bambuddy_printers_total gauge") lines.append(f"bambuddy_printers_total {len(printers)}") # Add trailing newline lines.append("") content = "\n".join(lines) return Response(content=content, media_type="text/plain; version=0.0.4; charset=utf-8")