from __future__ import annotations import gzip import json import os import re from collections import Counter, defaultdict from dataclasses import dataclass, field from datetime import date, datetime, timedelta from pathlib import Path from typing import Iterable from zoneinfo import ZoneInfo from fastapi import FastAPI from fastapi.responses import PlainTextResponse SESSIONS_DIR = Path(os.environ.get("OPENCLAW_SESSIONS_DIR", "/root/.openclaw/agents/main/sessions")) TIMEZONE = ZoneInfo(os.environ.get("OPENCLAW_EXPORTER_TZ", "Europe/Berlin")) SESSIONS_INDEX_PATH = Path(os.environ.get("OPENCLAW_SESSIONS_INDEX", str(SESSIONS_DIR / "sessions.json"))) USAGE_COST_CACHE_PATH = Path(os.environ.get("OPENCLAW_USAGE_COST_CACHE", str(SESSIONS_DIR / ".usage-cost-cache.json"))) HISTORY_DAYS = max(1, int(os.environ.get("OPENCLAW_HISTORY_DAYS", "30"))) HISTORY_HOURS = max(1, int(os.environ.get("OPENCLAW_HISTORY_HOURS", "168"))) UUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}") app = FastAPI(title="openclaw-usage-exporter") @dataclass class Totals: total: int = 0 input: int = 0 output: int = 0 cache_read: int = 0 cache_write: int = 0 reasoning: int = 0 entries: int = 0 cost_input: float = 0.0 cost_output: float = 0.0 cost_cache_read: float = 0.0 cost_cache_write: float = 0.0 cost_total: float = 0.0 def add_usage(self, usage: dict) -> tuple[int, int, int, int, int, int]: total_tokens = _as_int(usage.get("totalTokens")) input_tokens = _as_int(usage.get("input")) output_tokens = _as_int(usage.get("output")) cache_read_tokens = _as_int(usage.get("cacheRead")) cache_write_tokens = _as_int(usage.get("cacheWrite")) reasoning_tokens = _as_int(usage.get("reasoningTokens")) self.entries += 1 self.total += total_tokens self.input += input_tokens self.output += output_tokens self.cache_read += cache_read_tokens self.cache_write += cache_write_tokens self.reasoning += reasoning_tokens cost = usage.get("cost") or {} if isinstance(cost, dict): self.cost_input += _as_float(cost.get("input")) self.cost_output += _as_float(cost.get("output")) self.cost_cache_read += _as_float(cost.get("cacheRead")) self.cost_cache_write += _as_float(cost.get("cacheWrite")) self.cost_total += _as_float(cost.get("total")) return ( total_tokens, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens, ) @dataclass class UsageReport: today: Totals = field(default_factory=Totals) all_time: Totals = field(default_factory=Totals) files_scanned: int = 0 lines_scanned: int = 0 parse_errors: int = 0 message_records_today: int = 0 message_records_all_time: int = 0 non_message_records_today: int = 0 non_message_records_all_time: int = 0 unique_session_ids_today: set[str] = field(default_factory=set) unique_models_today: set[str] = field(default_factory=set) unique_providers_today: set[str] = field(default_factory=set) role_counts_today: Counter[str] = field(default_factory=Counter) stop_reason_counts_today: Counter[tuple[str, str, str]] = field(default_factory=Counter) tool_call_counts_today: Counter[str] = field(default_factory=Counter) error_type_counts_today: Counter[str] = field(default_factory=Counter) session_message_counts_today: Counter[tuple[str, str]] = field(default_factory=Counter) session_error_counts_today: Counter[str] = field(default_factory=Counter) session_tool_call_counts_today: Counter[tuple[str, str]] = field(default_factory=Counter) tokens_by_model_today: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) input_tokens_by_model_today: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) output_tokens_by_model_today: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) reasoning_tokens_by_model_today: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) session_tokens_today: defaultdict[str, int] = field(default_factory=lambda: defaultdict(int)) session_input_tokens_today: defaultdict[str, int] = field(default_factory=lambda: defaultdict(int)) session_output_tokens_today: defaultdict[str, int] = field(default_factory=lambda: defaultdict(int)) tokens_by_model_all_time: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) input_tokens_by_model_all_time: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) output_tokens_by_model_all_time: defaultdict[tuple[str, str], int] = field(default_factory=lambda: defaultdict(int)) session_tokens_all_time: defaultdict[str, int] = field(default_factory=lambda: defaultdict(int)) day_buckets: defaultdict[str, Totals] = field(default_factory=lambda: defaultdict(Totals)) hour_buckets: defaultdict[str, Totals] = field(default_factory=lambda: defaultdict(Totals)) @dataclass class SessionIndexReport: visible_count: int = 0 total_count: int = 0 has_more: bool = False sessions: list[dict] = field(default_factory=list) @dataclass class UsageCostCacheReport: version: int | None = None updated_at_ms: int | None = None files_tracked: int = 0 def _as_int(value: object) -> int: if value is None: return 0 if isinstance(value, bool): return int(value) if isinstance(value, (int, float)): return int(value) return 0 def _as_float(value: object) -> float: if value is None: return 0.0 if isinstance(value, bool): return float(value) if isinstance(value, (int, float)): return float(value) return 0.0 def _today_in_tz() -> date: return datetime.now(TIMEZONE).date() def _now_in_tz() -> datetime: return datetime.now(TIMEZONE) def _sanitize_label(value: object, default: str = "unknown") -> str: if value is None: return default text = str(value).strip() return text or default def _prom_value(value: str) -> str: return value.replace("\\", r"\\").replace("\n", r"\n").replace('"', r'\"') def _metric_line(name: str, value: object, **labels: object) -> str: if labels: rendered = ",".join(f'{key}="{_prom_value(_sanitize_label(val))}"' for key, val in sorted(labels.items())) return f"{name}{{{rendered}}} {value}" return f"{name} {value}" def _open_session_file(path: Path): if path.suffix == ".gz": return gzip.open(path, "rt", encoding="utf-8") return path.open("r", encoding="utf-8") def _session_id_from_path(path: Path) -> str | None: match = UUID_RE.search(path.name) if match: return match.group(0).lower() return None def _iter_session_paths() -> Iterable[Path]: for path in sorted(SESSIONS_DIR.glob("*.jsonl*")): if ".trajectory." in path.name: continue yield path def collect_usage() -> UsageReport: report = UsageReport() now = _now_in_tz() today = now.date() day_cutoff = today - timedelta(days=HISTORY_DAYS - 1) hour_cutoff = now.replace(minute=0, second=0, microsecond=0) - timedelta(hours=HISTORY_HOURS - 1) for path in _iter_session_paths(): report.files_scanned += 1 session_id_from_path = _session_id_from_path(path) try: with _open_session_file(path) as handle: for line in handle: report.lines_scanned += 1 if not line.strip(): continue try: record = json.loads(line) except json.JSONDecodeError: report.parse_errors += 1 continue message = record.get("message") or {} timestamp_ms = message.get("timestamp", record.get("timestamp")) if not isinstance(timestamp_ms, (int, float)): continue dt = datetime.fromtimestamp(timestamp_ms / 1000, TIMEZONE) usage_date = dt.date() is_today = usage_date == today if record.get("type") != "message": report.non_message_records_all_time += 1 if is_today: report.non_message_records_today += 1 continue report.message_records_all_time += 1 if is_today: report.message_records_today += 1 role = _sanitize_label(message.get("role")) model = _sanitize_label(message.get("model")) provider = _sanitize_label(message.get("provider")) stop_reason = _sanitize_label(message.get("stopReason")) tool_name = _sanitize_label(message.get("toolName")) error_type = _sanitize_label(message.get("errorType")) is_error = bool(message.get("isError")) usage = message.get("usage") session_id = session_id_from_path if is_today: report.role_counts_today[role] += 1 if model != "unknown": report.unique_models_today.add(model) if provider != "unknown": report.unique_providers_today.add(provider) if session_id: report.unique_session_ids_today.add(session_id) report.session_message_counts_today[(session_id, role)] += 1 if stop_reason != "unknown": report.stop_reason_counts_today[(stop_reason, model, provider)] += 1 if tool_name != "unknown": report.tool_call_counts_today[tool_name] += 1 if session_id: report.session_tool_call_counts_today[(session_id, tool_name)] += 1 if is_error: report.error_type_counts_today[error_type] += 1 if session_id: report.session_error_counts_today[session_id] += 1 if not isinstance(usage, dict): continue total_tokens, input_tokens, output_tokens, _, _, reasoning_tokens = report.all_time.add_usage(usage) report.tokens_by_model_all_time[(model, provider)] += total_tokens report.input_tokens_by_model_all_time[(model, provider)] += input_tokens report.output_tokens_by_model_all_time[(model, provider)] += output_tokens if session_id: report.session_tokens_all_time[session_id] += total_tokens if usage_date >= day_cutoff: report.day_buckets[usage_date.isoformat()].add_usage(usage) hour_bucket = dt.replace(minute=0, second=0, microsecond=0) if hour_bucket >= hour_cutoff: report.hour_buckets[hour_bucket.isoformat()].add_usage(usage) if is_today: report.today.add_usage(usage) report.tokens_by_model_today[(model, provider)] += total_tokens report.input_tokens_by_model_today[(model, provider)] += input_tokens report.output_tokens_by_model_today[(model, provider)] += output_tokens report.reasoning_tokens_by_model_today[(model, provider)] += reasoning_tokens if session_id: report.session_tokens_today[session_id] += total_tokens report.session_input_tokens_today[session_id] += input_tokens report.session_output_tokens_today[session_id] += output_tokens except OSError: report.parse_errors += 1 continue return report def collect_session_index() -> SessionIndexReport: report = SessionIndexReport() try: data = json.loads(SESSIONS_INDEX_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return report if not isinstance(data, dict): return report if isinstance(data.get("sessions"), list): sessions = [session for session in data.get("sessions", []) if isinstance(session, dict)] report.visible_count = _as_int(data.get("count")) or len(sessions) report.total_count = _as_int(data.get("totalCount")) or len(sessions) report.has_more = bool(data.get("hasMore")) report.sessions = sessions return report sessions = [session for session in data.values() if isinstance(session, dict)] report.visible_count = len(sessions) report.total_count = len(sessions) report.has_more = False report.sessions = sessions return report def collect_usage_cost_cache() -> UsageCostCacheReport: report = UsageCostCacheReport() try: data = json.loads(USAGE_COST_CACHE_PATH.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return report if not isinstance(data, dict): return report files = data.get("files") report.version = _as_int(data.get("version")) if data.get("version") is not None else None report.updated_at_ms = _as_int(data.get("updatedAt")) if data.get("updatedAt") is not None else None report.files_tracked = len(files) if isinstance(files, dict) else 0 return report def _append_current_day_metrics(lines: list[str], usage: UsageReport, today: str, tz_name: str) -> None: lines.extend([ "# HELP openclaw_daily_total_tokens Total OpenClaw tokens recorded today.", "# TYPE openclaw_daily_total_tokens gauge", _metric_line("openclaw_daily_total_tokens", usage.today.total, date=today, tz=tz_name), "# HELP openclaw_daily_input_tokens OpenClaw input tokens recorded today.", "# TYPE openclaw_daily_input_tokens gauge", _metric_line("openclaw_daily_input_tokens", usage.today.input, date=today, tz=tz_name), "# HELP openclaw_daily_output_tokens OpenClaw output tokens recorded today.", "# TYPE openclaw_daily_output_tokens gauge", _metric_line("openclaw_daily_output_tokens", usage.today.output, date=today, tz=tz_name), "# HELP openclaw_daily_cache_read_tokens OpenClaw cache-read tokens recorded today.", "# TYPE openclaw_daily_cache_read_tokens gauge", _metric_line("openclaw_daily_cache_read_tokens", usage.today.cache_read, date=today, tz=tz_name), "# HELP openclaw_daily_cache_write_tokens OpenClaw cache-write tokens recorded today.", "# TYPE openclaw_daily_cache_write_tokens gauge", _metric_line("openclaw_daily_cache_write_tokens", usage.today.cache_write, date=today, tz=tz_name), "# HELP openclaw_daily_reasoning_tokens OpenClaw reasoning tokens recorded today.", "# TYPE openclaw_daily_reasoning_tokens gauge", _metric_line("openclaw_daily_reasoning_tokens", usage.today.reasoning, date=today, tz=tz_name), "# HELP openclaw_daily_usage_entries OpenClaw usage entries recorded today.", "# TYPE openclaw_daily_usage_entries gauge", _metric_line("openclaw_daily_usage_entries", usage.today.entries, date=today, tz=tz_name), "# HELP openclaw_daily_cost_input OpenClaw input cost recorded today.", "# TYPE openclaw_daily_cost_input gauge", _metric_line("openclaw_daily_cost_input", f"{usage.today.cost_input:.12g}", date=today, tz=tz_name), "# HELP openclaw_daily_cost_output OpenClaw output cost recorded today.", "# TYPE openclaw_daily_cost_output gauge", _metric_line("openclaw_daily_cost_output", f"{usage.today.cost_output:.12g}", date=today, tz=tz_name), "# HELP openclaw_daily_cost_cache_read OpenClaw cache-read cost recorded today.", "# TYPE openclaw_daily_cost_cache_read gauge", _metric_line("openclaw_daily_cost_cache_read", f"{usage.today.cost_cache_read:.12g}", date=today, tz=tz_name), "# HELP openclaw_daily_cost_cache_write OpenClaw cache-write cost recorded today.", "# TYPE openclaw_daily_cost_cache_write gauge", _metric_line("openclaw_daily_cost_cache_write", f"{usage.today.cost_cache_write:.12g}", date=today, tz=tz_name), "# HELP openclaw_daily_cost_total OpenClaw total cost recorded today.", "# TYPE openclaw_daily_cost_total gauge", _metric_line("openclaw_daily_cost_total", f"{usage.today.cost_total:.12g}", date=today, tz=tz_name), "# HELP openclaw_daily_message_records Total message records seen today.", "# TYPE openclaw_daily_message_records gauge", _metric_line("openclaw_daily_message_records", usage.message_records_today, date=today, tz=tz_name), "# HELP openclaw_daily_non_message_records Total non-message records seen today.", "# TYPE openclaw_daily_non_message_records gauge", _metric_line("openclaw_daily_non_message_records", usage.non_message_records_today, date=today, tz=tz_name), "# HELP openclaw_daily_unique_sessions Number of unique session ids seen today.", "# TYPE openclaw_daily_unique_sessions gauge", _metric_line("openclaw_daily_unique_sessions", len(usage.unique_session_ids_today), date=today, tz=tz_name), "# HELP openclaw_daily_unique_models Number of unique models seen today.", "# TYPE openclaw_daily_unique_models gauge", _metric_line("openclaw_daily_unique_models", len(usage.unique_models_today), date=today, tz=tz_name), "# HELP openclaw_daily_unique_providers Number of unique providers seen today.", "# TYPE openclaw_daily_unique_providers gauge", _metric_line("openclaw_daily_unique_providers", len(usage.unique_providers_today), date=today, tz=tz_name), ]) def _append_counter_metrics(lines: list[str], usage: UsageReport) -> None: lines.extend([ "# HELP openclaw_total_tokens_total All-time OpenClaw tokens from session usage logs.", "# TYPE openclaw_total_tokens_total counter", _metric_line("openclaw_total_tokens_total", usage.all_time.total), "# HELP openclaw_input_tokens_total All-time OpenClaw input tokens from session usage logs.", "# TYPE openclaw_input_tokens_total counter", _metric_line("openclaw_input_tokens_total", usage.all_time.input), "# HELP openclaw_output_tokens_total All-time OpenClaw output tokens from session usage logs.", "# TYPE openclaw_output_tokens_total counter", _metric_line("openclaw_output_tokens_total", usage.all_time.output), "# HELP openclaw_cache_read_tokens_total All-time OpenClaw cache-read tokens from session usage logs.", "# TYPE openclaw_cache_read_tokens_total counter", _metric_line("openclaw_cache_read_tokens_total", usage.all_time.cache_read), "# HELP openclaw_cache_write_tokens_total All-time OpenClaw cache-write tokens from session usage logs.", "# TYPE openclaw_cache_write_tokens_total counter", _metric_line("openclaw_cache_write_tokens_total", usage.all_time.cache_write), "# HELP openclaw_reasoning_tokens_total All-time OpenClaw reasoning tokens from session usage logs.", "# TYPE openclaw_reasoning_tokens_total counter", _metric_line("openclaw_reasoning_tokens_total", usage.all_time.reasoning), "# HELP openclaw_usage_entries_total All-time OpenClaw usage entries from session usage logs.", "# TYPE openclaw_usage_entries_total counter", _metric_line("openclaw_usage_entries_total", usage.all_time.entries), "# HELP openclaw_cost_input_usd_total All-time OpenClaw input cost in USD from session usage logs.", "# TYPE openclaw_cost_input_usd_total counter", _metric_line("openclaw_cost_input_usd_total", f"{usage.all_time.cost_input:.12g}"), "# HELP openclaw_cost_output_usd_total All-time OpenClaw output cost in USD from session usage logs.", "# TYPE openclaw_cost_output_usd_total counter", _metric_line("openclaw_cost_output_usd_total", f"{usage.all_time.cost_output:.12g}"), "# HELP openclaw_cost_cache_read_usd_total All-time OpenClaw cache-read cost in USD from session usage logs.", "# TYPE openclaw_cost_cache_read_usd_total counter", _metric_line("openclaw_cost_cache_read_usd_total", f"{usage.all_time.cost_cache_read:.12g}"), "# HELP openclaw_cost_cache_write_usd_total All-time OpenClaw cache-write cost in USD from session usage logs.", "# TYPE openclaw_cost_cache_write_usd_total counter", _metric_line("openclaw_cost_cache_write_usd_total", f"{usage.all_time.cost_cache_write:.12g}"), "# HELP openclaw_cost_total_usd_total All-time OpenClaw total cost in USD from session usage logs.", "# TYPE openclaw_cost_total_usd_total counter", _metric_line("openclaw_cost_total_usd_total", f"{usage.all_time.cost_total:.12g}"), "# HELP openclaw_message_records_total All-time message records seen in session logs.", "# TYPE openclaw_message_records_total counter", _metric_line("openclaw_message_records_total", usage.message_records_all_time), "# HELP openclaw_non_message_records_total All-time non-message records seen in session logs.", "# TYPE openclaw_non_message_records_total counter", _metric_line("openclaw_non_message_records_total", usage.non_message_records_all_time), ]) def _append_bucket_metrics(lines: list[str], usage: UsageReport, tz_name: str) -> None: lines.extend([ "# HELP openclaw_day_total_tokens Daily token totals for recent retained days.", "# TYPE openclaw_day_total_tokens gauge", "# HELP openclaw_day_cost_total Daily cost totals for recent retained days.", "# TYPE openclaw_day_cost_total gauge", "# HELP openclaw_day_usage_entries Daily usage-entry totals for recent retained days.", "# TYPE openclaw_day_usage_entries gauge", ]) for bucket_date, totals in sorted(usage.day_buckets.items()): labels = {"date": bucket_date, "tz": tz_name} lines.append(_metric_line("openclaw_day_total_tokens", totals.total, **labels)) lines.append(_metric_line("openclaw_day_cost_total", f"{totals.cost_total:.12g}", **labels)) lines.append(_metric_line("openclaw_day_usage_entries", totals.entries, **labels)) lines.extend([ "# HELP openclaw_hour_total_tokens Hourly token totals for recent retained hours.", "# TYPE openclaw_hour_total_tokens gauge", "# HELP openclaw_hour_cost_total Hourly cost totals for recent retained hours.", "# TYPE openclaw_hour_cost_total gauge", "# HELP openclaw_hour_usage_entries Hourly usage-entry totals for recent retained hours.", "# TYPE openclaw_hour_usage_entries gauge", ]) for bucket_hour, totals in sorted(usage.hour_buckets.items()): labels = {"hour": bucket_hour, "tz": tz_name} lines.append(_metric_line("openclaw_hour_total_tokens", totals.total, **labels)) lines.append(_metric_line("openclaw_hour_cost_total", f"{totals.cost_total:.12g}", **labels)) lines.append(_metric_line("openclaw_hour_usage_entries", totals.entries, **labels)) def _append_breakdown_metrics(lines: list[str], usage: UsageReport, today: str, tz_name: str) -> None: lines.extend([ "# HELP openclaw_daily_messages_total Daily message count by role.", "# TYPE openclaw_daily_messages_total gauge", ]) for role, count in sorted(usage.role_counts_today.items()): lines.append(_metric_line("openclaw_daily_messages_total", count, date=today, tz=tz_name, role=role)) lines.extend([ "# HELP openclaw_daily_stop_reasons_total Daily assistant stop reasons by model/provider.", "# TYPE openclaw_daily_stop_reasons_total gauge", ]) for (stop_reason, model, provider), count in sorted(usage.stop_reason_counts_today.items()): lines.append(_metric_line("openclaw_daily_stop_reasons_total", count, date=today, tz=tz_name, stop_reason=stop_reason, model=model, provider=provider)) lines.extend([ "# HELP openclaw_daily_tool_calls_total Daily tool call count by tool name.", "# TYPE openclaw_daily_tool_calls_total gauge", ]) for tool_name, count in sorted(usage.tool_call_counts_today.items()): lines.append(_metric_line("openclaw_daily_tool_calls_total", count, date=today, tz=tz_name, tool_name=tool_name)) lines.extend([ "# HELP openclaw_daily_errors_total Daily message error count by error type.", "# TYPE openclaw_daily_errors_total gauge", ]) for error_type, count in sorted(usage.error_type_counts_today.items()): lines.append(_metric_line("openclaw_daily_errors_total", count, date=today, tz=tz_name, error_type=error_type)) lines.extend([ "# HELP openclaw_daily_model_total_tokens Daily total tokens by model/provider.", "# TYPE openclaw_daily_model_total_tokens gauge", ]) for (model, provider), count in sorted(usage.tokens_by_model_today.items()): lines.append(_metric_line("openclaw_daily_model_total_tokens", count, date=today, tz=tz_name, model=model, provider=provider)) lines.extend([ "# HELP openclaw_daily_model_input_tokens Daily input tokens by model/provider.", "# TYPE openclaw_daily_model_input_tokens gauge", ]) for (model, provider), count in sorted(usage.input_tokens_by_model_today.items()): lines.append(_metric_line("openclaw_daily_model_input_tokens", count, date=today, tz=tz_name, model=model, provider=provider)) lines.extend([ "# HELP openclaw_daily_model_output_tokens Daily output tokens by model/provider.", "# TYPE openclaw_daily_model_output_tokens gauge", ]) for (model, provider), count in sorted(usage.output_tokens_by_model_today.items()): lines.append(_metric_line("openclaw_daily_model_output_tokens", count, date=today, tz=tz_name, model=model, provider=provider)) lines.extend([ "# HELP openclaw_daily_model_reasoning_tokens Daily reasoning tokens by model/provider.", "# TYPE openclaw_daily_model_reasoning_tokens gauge", ]) for (model, provider), count in sorted(usage.reasoning_tokens_by_model_today.items()): lines.append(_metric_line("openclaw_daily_model_reasoning_tokens", count, date=today, tz=tz_name, model=model, provider=provider)) lines.extend([ "# HELP openclaw_model_total_tokens_total All-time total tokens by model/provider.", "# TYPE openclaw_model_total_tokens_total counter", ]) for (model, provider), count in sorted(usage.tokens_by_model_all_time.items()): lines.append(_metric_line("openclaw_model_total_tokens_total", count, model=model, provider=provider)) lines.extend([ "# HELP openclaw_model_input_tokens_total All-time input tokens by model/provider.", "# TYPE openclaw_model_input_tokens_total counter", ]) for (model, provider), count in sorted(usage.input_tokens_by_model_all_time.items()): lines.append(_metric_line("openclaw_model_input_tokens_total", count, model=model, provider=provider)) lines.extend([ "# HELP openclaw_model_output_tokens_total All-time output tokens by model/provider.", "# TYPE openclaw_model_output_tokens_total counter", ]) for (model, provider), count in sorted(usage.output_tokens_by_model_all_time.items()): lines.append(_metric_line("openclaw_model_output_tokens_total", count, model=model, provider=provider)) lines.extend([ "# HELP openclaw_daily_session_total_tokens Daily total tokens by session id.", "# TYPE openclaw_daily_session_total_tokens gauge", ]) for session_id, count in sorted(usage.session_tokens_today.items()): lines.append(_metric_line("openclaw_daily_session_total_tokens", count, date=today, tz=tz_name, session_id=session_id)) lines.extend([ "# HELP openclaw_daily_session_input_tokens Daily input tokens by session id.", "# TYPE openclaw_daily_session_input_tokens gauge", ]) for session_id, count in sorted(usage.session_input_tokens_today.items()): lines.append(_metric_line("openclaw_daily_session_input_tokens", count, date=today, tz=tz_name, session_id=session_id)) lines.extend([ "# HELP openclaw_daily_session_output_tokens Daily output tokens by session id.", "# TYPE openclaw_daily_session_output_tokens gauge", ]) for session_id, count in sorted(usage.session_output_tokens_today.items()): lines.append(_metric_line("openclaw_daily_session_output_tokens", count, date=today, tz=tz_name, session_id=session_id)) lines.extend([ "# HELP openclaw_session_total_tokens_total All-time total tokens by session id.", "# TYPE openclaw_session_total_tokens_total counter", ]) for session_id, count in sorted(usage.session_tokens_all_time.items()): lines.append(_metric_line("openclaw_session_total_tokens_total", count, session_id=session_id)) lines.extend([ "# HELP openclaw_daily_session_messages_total Daily message count by session id and role.", "# TYPE openclaw_daily_session_messages_total gauge", ]) for (session_id, role), count in sorted(usage.session_message_counts_today.items()): lines.append(_metric_line("openclaw_daily_session_messages_total", count, date=today, tz=tz_name, session_id=session_id, role=role)) lines.extend([ "# HELP openclaw_daily_session_errors_total Daily error count by session id.", "# TYPE openclaw_daily_session_errors_total gauge", ]) for session_id, count in sorted(usage.session_error_counts_today.items()): lines.append(_metric_line("openclaw_daily_session_errors_total", count, date=today, tz=tz_name, session_id=session_id)) lines.extend([ "# HELP openclaw_daily_session_tool_calls_total Daily tool call count by session id and tool name.", "# TYPE openclaw_daily_session_tool_calls_total gauge", ]) for (session_id, tool_name), count in sorted(usage.session_tool_call_counts_today.items()): lines.append(_metric_line("openclaw_daily_session_tool_calls_total", count, date=today, tz=tz_name, session_id=session_id, tool_name=tool_name)) def _append_exporter_metrics(lines: list[str], usage: UsageReport, session_index: SessionIndexReport, usage_cost_cache: UsageCostCacheReport, today: str, tz_name: str) -> None: lines.extend([ "# HELP openclaw_exporter_files_scanned Session files scanned for metrics generation.", "# TYPE openclaw_exporter_files_scanned gauge", _metric_line("openclaw_exporter_files_scanned", usage.files_scanned, date=today, tz=tz_name), "# HELP openclaw_exporter_lines_scanned Session lines scanned for metrics generation.", "# TYPE openclaw_exporter_lines_scanned gauge", _metric_line("openclaw_exporter_lines_scanned", usage.lines_scanned, date=today, tz=tz_name), "# HELP openclaw_exporter_parse_errors Parse or read errors while generating metrics.", "# TYPE openclaw_exporter_parse_errors gauge", _metric_line("openclaw_exporter_parse_errors", usage.parse_errors, date=today, tz=tz_name), "# HELP openclaw_exporter_history_days_retained Retained day-bucket history window.", "# TYPE openclaw_exporter_history_days_retained gauge", _metric_line("openclaw_exporter_history_days_retained", HISTORY_DAYS), "# HELP openclaw_exporter_history_hours_retained Retained hour-bucket history window.", "# TYPE openclaw_exporter_history_hours_retained gauge", _metric_line("openclaw_exporter_history_hours_retained", HISTORY_HOURS), "# HELP openclaw_sessions_visible_count Session entries currently present in sessions.json output.", "# TYPE openclaw_sessions_visible_count gauge", _metric_line("openclaw_sessions_visible_count", session_index.visible_count), "# HELP openclaw_sessions_total_count Total stored sessions reported by sessions.json.", "# TYPE openclaw_sessions_total_count gauge", _metric_line("openclaw_sessions_total_count", session_index.total_count), "# HELP openclaw_sessions_has_more Whether sessions.json was truncated when written.", "# TYPE openclaw_sessions_has_more gauge", _metric_line("openclaw_sessions_has_more", int(session_index.has_more)), "# HELP openclaw_usage_cost_cache_files_tracked Number of files tracked in .usage-cost-cache.json.", "# TYPE openclaw_usage_cost_cache_files_tracked gauge", _metric_line("openclaw_usage_cost_cache_files_tracked", usage_cost_cache.files_tracked), ]) if usage_cost_cache.version is not None: lines.extend([ "# HELP openclaw_usage_cost_cache_version Version field from .usage-cost-cache.json.", "# TYPE openclaw_usage_cost_cache_version gauge", _metric_line("openclaw_usage_cost_cache_version", usage_cost_cache.version), ]) if usage_cost_cache.updated_at_ms is not None: lines.extend([ "# HELP openclaw_usage_cost_cache_updated_timestamp_seconds Last update time of .usage-cost-cache.json.", "# TYPE openclaw_usage_cost_cache_updated_timestamp_seconds gauge", _metric_line("openclaw_usage_cost_cache_updated_timestamp_seconds", f"{usage_cost_cache.updated_at_ms / 1000:.3f}"), ]) def _append_session_inventory_metrics(lines: list[str], session_index: SessionIndexReport) -> None: lines.extend([ "# HELP openclaw_session_info Non-sensitive session metadata from sessions.json. Session key is intentionally omitted.", "# TYPE openclaw_session_info gauge", "# HELP openclaw_session_age_seconds Current age of a stored session from sessions.json.", "# TYPE openclaw_session_age_seconds gauge", "# HELP openclaw_session_updated_timestamp_seconds Last update time of a stored session from sessions.json.", "# TYPE openclaw_session_updated_timestamp_seconds gauge", "# HELP openclaw_session_started_timestamp_seconds Session start time from sessions.json.", "# TYPE openclaw_session_started_timestamp_seconds gauge", "# HELP openclaw_session_last_interaction_timestamp_seconds Last interaction time from sessions.json.", "# TYPE openclaw_session_last_interaction_timestamp_seconds gauge", "# HELP openclaw_session_total_tokens Stored session totalTokens from sessions.json.", "# TYPE openclaw_session_total_tokens gauge", "# HELP openclaw_session_context_tokens Stored session contextTokens from sessions.json.", "# TYPE openclaw_session_context_tokens gauge", "# HELP openclaw_session_runtime_seconds Stored runtimeMs from sessions.json.", "# TYPE openclaw_session_runtime_seconds gauge", "# HELP openclaw_session_estimated_cost_usd Stored estimatedCostUsd from sessions.json.", "# TYPE openclaw_session_estimated_cost_usd gauge", "# HELP openclaw_session_compaction_count Stored compactionCount from sessions.json.", "# TYPE openclaw_session_compaction_count gauge", "# HELP openclaw_session_usage_family_session_count Stored usageFamilySessionIds count from sessions.json.", "# TYPE openclaw_session_usage_family_session_count gauge", ]) for session in sorted(session_index.sessions, key=lambda item: _sanitize_label(item.get("sessionId"))): session_id = _sanitize_label(session.get("sessionId")) usage_family_session_ids = session.get("usageFamilySessionIds") labels = { "session_id": session_id, "status": _sanitize_label(session.get("status")), "chat_type": _sanitize_label(session.get("chatType")), "last_channel": _sanitize_label(session.get("lastChannel")), "model": _sanitize_label(session.get("model")), "provider": _sanitize_label(session.get("modelProvider")), "auth_profile_override": _sanitize_label(session.get("authProfileOverride")), "system_sent": str(bool(session.get("systemSent"))).lower(), "aborted_last_run": str(bool(session.get("abortedLastRun"))).lower(), "total_tokens_fresh": str(bool(session.get("totalTokensFresh"))).lower(), } lines.append(_metric_line("openclaw_session_info", 1, **labels)) lines.append(_metric_line("openclaw_session_age_seconds", f"{_as_int(session.get('ageMs')) / 1000:.3f}", session_id=session_id)) lines.append(_metric_line("openclaw_session_updated_timestamp_seconds", f"{_as_int(session.get('updatedAt')) / 1000:.3f}", session_id=session_id)) lines.append(_metric_line("openclaw_session_started_timestamp_seconds", f"{_as_int(session.get('sessionStartedAt')) / 1000:.3f}", session_id=session_id)) lines.append(_metric_line("openclaw_session_last_interaction_timestamp_seconds", f"{_as_int(session.get('lastInteractionAt')) / 1000:.3f}", session_id=session_id)) lines.append(_metric_line("openclaw_session_total_tokens", _as_int(session.get("totalTokens")), session_id=session_id)) lines.append(_metric_line("openclaw_session_context_tokens", _as_int(session.get("contextTokens")), session_id=session_id)) lines.append(_metric_line("openclaw_session_runtime_seconds", f"{_as_int(session.get('runtimeMs')) / 1000:.3f}", session_id=session_id)) lines.append(_metric_line("openclaw_session_estimated_cost_usd", f"{_as_float(session.get('estimatedCostUsd')):.12g}", session_id=session_id)) lines.append(_metric_line("openclaw_session_compaction_count", _as_int(session.get("compactionCount")), session_id=session_id)) lines.append(_metric_line("openclaw_session_usage_family_session_count", len(usage_family_session_ids) if isinstance(usage_family_session_ids, list) else 0, session_id=session_id)) @app.get("/metrics", response_class=PlainTextResponse) def metrics() -> str: usage = collect_usage() session_index = collect_session_index() usage_cost_cache = collect_usage_cost_cache() today = _today_in_tz().isoformat() tz_name = TIMEZONE.key lines: list[str] = [] _append_current_day_metrics(lines, usage, today, tz_name) _append_counter_metrics(lines, usage) _append_bucket_metrics(lines, usage, tz_name) _append_breakdown_metrics(lines, usage, today, tz_name) _append_exporter_metrics(lines, usage, session_index, usage_cost_cache, today, tz_name) _append_session_inventory_metrics(lines, session_index) lines.append("") return "\n".join(lines)