From b42de4e6692cb8a42db85e34d4fea7a7951826e9 Mon Sep 17 00:00:00 2001 From: POWERFULMOVES <142271328+POWERFULMOVES@users.noreply.github.com> Date: Mon, 29 Sep 2025 19:43:19 -0400 Subject: [PATCH] feat(metrics): extend publisher telemetry rollups --- pmoves/docs/NEXT_STEPS.md | 8 +- pmoves/docs/ROADMAP.md | 6 +- pmoves/docs/TELEMETRY_ROI.md | 64 ++++++ pmoves/services/common/telemetry.py | 228 +++++++++++++++++++++ pmoves/services/publisher-discord/main.py | 168 ++++++++++++++-- pmoves/services/publisher/README.md | 3 + pmoves/services/publisher/publisher.py | 229 +--------------------- 7 files changed, 462 insertions(+), 244 deletions(-) create mode 100644 pmoves/docs/TELEMETRY_ROI.md create mode 100644 pmoves/services/common/telemetry.py diff --git a/pmoves/docs/NEXT_STEPS.md b/pmoves/docs/NEXT_STEPS.md index 3fe82c08ff..cbf3819334 100644 --- a/pmoves/docs/NEXT_STEPS.md +++ b/pmoves/docs/NEXT_STEPS.md @@ -2,7 +2,7 @@ _Last updated: 2025-09-26 (geometry cache sync)_ -_Last updated: 2025-10-01_ +_Last updated: 2025-10-05_ ## Immediate @@ -13,15 +13,15 @@ _Last updated: 2025-10-01_ - [ ] Activate the n8n approval poller and echo publisher workflows once secrets are loaded; document the activation + first successful run. - [ ] Confirm Jellyfin credentials (API key and optional user id) allow library enumeration; note any dependency gaps that require new guardrails. - [ ] Validate that enriched publisher metadata propagates into Agent Zero and Discord events; schedule a backfill for legacy records if fields are missing. -- [ ] Hit the publisher `/metrics` endpoint and capture the turnaround/latency summary for the runbook. -- [ ] Confirm Supabase `publisher_metrics_rollup` rows are created with engagement + cost payloads and link the ROI dashboard query. +- [ ] Hit the publisher and publisher-discord `/metrics` endpoints and capture the turnaround/latency summary for the runbook. +- [ ] Confirm Supabase `publisher_metrics_rollup` and `publisher_discord_metrics` rows are created with engagement + cost payloads and link the ROI dashboard query. - [ ] Record step-by-step evidence in `SESSION_IMPLEMENTATION_PLAN.md` while executing the operational reminders list. ### 2. Jellyfin Publisher Reliability - [x] Add a scheduled refresh or webhook trigger so Jellyfin libraries update after publisher runs; include cron/webhook settings in `services/publisher/README.md`. - [ ] Expand error/reporting hooks so failures surface with actionable messages (Jellyfin HTTP errors, dependency mismatches, asset gaps). - [ ] Backfill historic Jellyfin entries with enriched metadata and confirm downstream consumers (Agent Zero, Discord) render the new fields. -- [ ] Plot baseline ROI visuals (turnaround vs engagement vs cost) using the Supabase rollup table and document interpretation guidance in the dashboard notes. +- [ ] Plot baseline ROI visuals (turnaround vs engagement vs cost) using the Supabase rollup tables and incorporate the guidance captured in `TELEMETRY_ROI.md` into the dashboard notes. ### 3. Graph & Retrieval Enhancements (Kickoff M3) - [ ] Seed Neo4j with the brand alias dictionary (DARKXSIDE, POWERFULMOVES, plus pending community submissions) and record Cypher script locations (draft plan in `SESSION_IMPLEMENTATION_PLAN.md`). diff --git a/pmoves/docs/ROADMAP.md b/pmoves/docs/ROADMAP.md index ac5e306d6f..cdd7e1ecc7 100644 --- a/pmoves/docs/ROADMAP.md +++ b/pmoves/docs/ROADMAP.md @@ -1,5 +1,5 @@ # PMOVES v5 • ROADMAP -_Last updated: 2025-10-01_ +_Last updated: 2025-10-05_ ## Vision A production-ready, self-hostable orchestration mesh for creative + agent workloads across GPU boxes and Jetsons: **hybrid Hi‑RAG**, **Supabase Studio**, **n8n orchestration**, **Jellyfin publishing**, and **graph-aware retrieval**. @@ -19,7 +19,7 @@ A production-ready, self-hostable orchestration mesh for creative + agent worklo | ✅ | ComfyUI ↔ MinIO Presign microservice | `services/presign/api.py` provides presigned PUT/GET/POST helpers for MinIO/S3. | | ✅ | Render Webhook (Comfy → Supabase Studio) | `services/render-webhook/webhook.py` inserts submissions into `studio_board` with optional auto-approval. | | 🚧 | Publisher (Jellyfin) | `services/publisher/publisher.py` consumes approval events and refreshes Jellyfin; optional dependency guards and envelope fallback landed, but richer metadata handling and error reporting are still pending. | -| ✅ | Publisher telemetry & ROI rollups | `/metrics` endpoint plus Supabase rollups from `services/publisher/publisher.py` make turnaround/latency/cost telemetry queryable for dashboards. | +| ✅ | Publisher telemetry & ROI rollups | `/metrics` feeds from `services/publisher/publisher.py` and `services/publisher-discord/main.py` expose turnaround/latency/cost telemetry, with Supabase rollups powering the ROI dashboards documented in `pmoves/docs/TELEMETRY_ROI.md`. | | ✅ | PDF/MinIO ingestion | `services/pdf-ingest/app.py` pulls PDFs from MinIO, extracts text, forwards chunks, and emits ingest events. | | ⏳ | n8n flows (Discord/webhooks) | `n8n/flows/*.json` only define placeholder workflows; Supabase pollers and Discord actions must be configured. | | 🚧 | Jellyfin library refresh hook + Discord rich cards | Jellyfin refresh occurs in the publisher, and `services/publisher-discord` formats embeds, but published-event wiring and asset deep links remain. Automation activation plan logged in `pmoves/docs/SESSION_IMPLEMENTATION_PLAN.md`. | @@ -27,7 +27,7 @@ A production-ready, self-hostable orchestration mesh for creative + agent worklo **Outstanding to close M2:** - publisher metadata/envelope polish — namespace-aware filenames, dependency guards, and fallback envelopes merged; monitor adoption and backfill historic assets if needed - add published-event Discord embeds via `content.published.v1`; execution plan staged in `SESSION_IMPLEMENTATION_PLAN.md` -- wire Supabase ROI dashboards to the new publisher telemetry rollups; document interpretation guidance alongside ROI reporting. +- wire Supabase ROI dashboards to the new publisher telemetry rollups; document interpretation guidance alongside ROI reporting (**see `docs/TELEMETRY_ROI.md` for the latest walkthrough**). - build the Supabase→Discord automation inside the n8n exports and track discrete workflow validation steps in the implementation log - execute the Supabase → Agent Zero → Discord activation checklist (`pmoves/docs/SUPABASE_DISCORD_AUTOMATION.md`) and log the validation timestamp (see operational reminders captured in the implementation plan) diff --git a/pmoves/docs/TELEMETRY_ROI.md b/pmoves/docs/TELEMETRY_ROI.md new file mode 100644 index 0000000000..da7a2353af --- /dev/null +++ b/pmoves/docs/TELEMETRY_ROI.md @@ -0,0 +1,64 @@ +# Telemetry & ROI Dashboards + +_Last updated: 2025-10-05_ + +## Overview + +Publisher-facing services now expose real-time telemetry so operators can +correlate turnaround, approval latency, engagement, and cost. Two lightweight +HTTP endpoints provide quick snapshots while Supabase rollup tables persist the +event-level detail required for historical dashboards. + +| Service | Endpoint | Purpose | +| --- | --- | --- | +| `services/publisher/publisher.py` | `http://:9095/metrics` | Aggregated counts/averages for artifact downloads, Jellyfin refreshes, turnaround, approval latency, engagement, and cost inputs. | +| `services/publisher-discord/main.py` | `http://:/metrics` | Discord delivery counters plus the same turnaround/approval telemetry derived from published events. | + +Both endpoints return JSON with `telemetry` blocks shaped like: + +```json +{ + "turnaround_samples": 12, + "avg_turnaround_seconds": 5400.25, + "approval_latency_samples": 12, + "avg_approval_latency_seconds": 900.5, + "engagement_totals": {"views": 1800, "ctr": 12.5}, + "cost_totals": {"storage_gb": 6.2, "processing_minutes": 44.0} +} +``` + +Use the metrics endpoints for smoke tests and runbooks; dashboards should query +Supabase directly to avoid scraping service processes. + +## Supabase Rollups + +| Table | Source | Description | +| --- | --- | --- | +| `publisher_metrics_rollup` | `services/publisher/publisher.py` | One row per published artifact, keyed by `artifact_uri`, capturing turnaround/approval latency plus engagement and cost payloads forwarded from approvals. | +| `publisher_discord_metrics` | `services/publisher-discord/main.py` | Mirrors publisher telemetry for Discord notifications and adds `webhook_success` + `channel` flags to monitor downstream delivery health. | + +Both tables are upserted using `services/common/supabase.py::upsert_row`. Use +`PUBLISHER_METRICS_CONFLICT` and `DISCORD_METRICS_CONFLICT` environment +variables to tune the `ON CONFLICT` keys when running migrations. + +## Interpreting ROI Dashboards + +1. **Turnaround vs. Approval Latency** – Plot `avg_turnaround_seconds` and + `avg_approval_latency_seconds` trends. Rising approval latency usually points + to reviewer bottlenecks; rising turnaround means ingest-to-publish automation + needs tuning. +2. **Engagement Ratios** – Divide `engagement_totals` (views, CTR, likes) by the + corresponding `cost_totals` (processing minutes, storage GB, egress). High + engagement with low cost indicates good ROI; low engagement + high cost flags + candidates for pruning or repackaging. +3. **Channel Health** – Join `publisher_metrics_rollup` and + `publisher_discord_metrics` on `artifact_uri` (or slug) to confirm Discord + notifications land for the same assets that hit Jellyfin. Investigate rows + where `webhook_success` is false to catch stale credentials or rate limits. +4. **Namespace Drill-down** – Group by `namespace` to understand which creator + lanes drive the most engagement per unit of spend. Feed these insights into + the sprint prioritisation matrix. + +Document snapshot queries alongside dashboards so the next operator can run +validations quickly (see `docs/NEXT_STEPS.md` for the checklist). + diff --git a/pmoves/services/common/telemetry.py b/pmoves/services/common/telemetry.py new file mode 100644 index 0000000000..87ff21b324 --- /dev/null +++ b/pmoves/services/common/telemetry.py @@ -0,0 +1,228 @@ +"""Shared telemetry helpers for publisher-style services.""" + +from __future__ import annotations + +import datetime as _dt +from dataclasses import asdict, dataclass, field +from typing import Any, Dict, Optional, Tuple + + +@dataclass +class PublisherMetrics: + """Aggregate metrics for publisher style services.""" + + downloads: int = 0 + download_failures: int = 0 + refresh_attempts: int = 0 + refresh_success: int = 0 + refresh_failures: int = 0 + turnaround_samples: int = 0 + total_turnaround_seconds: float = 0.0 + max_turnaround_seconds: float = 0.0 + approval_latency_samples: int = 0 + total_approval_latency_seconds: float = 0.0 + max_approval_latency_seconds: float = 0.0 + engagement_events: int = 0 + engagement_totals: Dict[str, float] = field(default_factory=dict) + cost_events: int = 0 + cost_totals: Dict[str, float] = field(default_factory=dict) + + def record_download_success(self) -> None: + self.downloads += 1 + + def record_download_failure(self) -> None: + self.download_failures += 1 + + def record_refresh_attempt(self) -> None: + self.refresh_attempts += 1 + + def record_refresh_success(self) -> None: + self.refresh_success += 1 + + def record_refresh_failure(self) -> None: + self.refresh_failures += 1 + + def record_turnaround(self, seconds: Optional[float]) -> None: + if seconds is None or seconds < 0: + return + self.turnaround_samples += 1 + self.total_turnaround_seconds += seconds + if seconds > self.max_turnaround_seconds: + self.max_turnaround_seconds = seconds + + def record_approval_latency(self, seconds: Optional[float]) -> None: + if seconds is None or seconds < 0: + return + self.approval_latency_samples += 1 + self.total_approval_latency_seconds += seconds + if seconds > self.max_approval_latency_seconds: + self.max_approval_latency_seconds = seconds + + def record_engagement(self, engagement: Dict[str, float]) -> None: + if not engagement: + return + self.engagement_events += 1 + for key, value in engagement.items(): + try: + numeric = float(value) + except (TypeError, ValueError): + continue + self.engagement_totals[key] = self.engagement_totals.get(key, 0.0) + numeric + + def record_cost(self, cost: Dict[str, float]) -> None: + if not cost: + return + self.cost_events += 1 + for key, value in cost.items(): + try: + numeric = float(value) + except (TypeError, ValueError): + continue + self.cost_totals[key] = self.cost_totals.get(key, 0.0) + numeric + + def summary(self) -> Dict[str, Any]: + data = asdict(self) + if self.turnaround_samples: + data["avg_turnaround_seconds"] = self.total_turnaround_seconds / self.turnaround_samples + if self.approval_latency_samples: + data["avg_approval_latency_seconds"] = ( + self.total_approval_latency_seconds / self.approval_latency_samples + ) + return data + + +@dataclass +class PublishTelemetry: + published_at: _dt.datetime + turnaround_seconds: Optional[float] + approval_latency_seconds: Optional[float] + engagement: Dict[str, float] + cost: Dict[str, float] + + def to_meta(self) -> Dict[str, Any]: + meta: Dict[str, Any] = { + "published_at": self.published_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"), + } + if self.turnaround_seconds is not None: + meta["turnaround_seconds"] = self.turnaround_seconds + if self.approval_latency_seconds is not None: + meta["approval_to_publish_seconds"] = self.approval_latency_seconds + if self.engagement: + meta["engagement"] = self.engagement + if self.cost: + meta["cost"] = self.cost + return meta + + def to_rollup_row(self, *, artifact_uri: str, namespace: str, slug: str) -> Dict[str, Any]: + return { + "artifact_uri": artifact_uri, + "namespace": namespace, + "slug": slug, + "published_at": self.published_at.isoformat(), + "turnaround_seconds": self.turnaround_seconds, + "approval_latency_seconds": self.approval_latency_seconds, + "engagement": self.engagement or None, + "cost": self.cost or None, + } + + +def _parse_iso8601(value: Optional[Any]) -> Optional[_dt.datetime]: + if not value or not isinstance(value, str): + return None + value = value.strip() + if not value: + return None + try: + if value.endswith("Z"): + value = value[:-1] + "+00:00" + return _dt.datetime.fromisoformat(value) + except ValueError: + return None + + +def _coerce_numeric(value: Any) -> Optional[float]: + try: + if value is None: + return None + numeric = float(value) + if numeric != numeric: # NaN + return None + return numeric + except (TypeError, ValueError): + return None + + +def _extract_first(meta: Dict[str, Any], keys: Tuple[str, ...]) -> Optional[str]: + for key in keys: + candidate = meta.get(key) + if isinstance(candidate, str) and candidate: + return candidate + return None + + +def compute_publish_telemetry( + incoming_meta: Optional[Dict[str, Any]], + event_ts: Optional[str], + published_at: _dt.datetime, +) -> PublishTelemetry: + meta = incoming_meta or {} + start_keys = ( + "ingest_started_at", + "submitted_at", + "created_at", + "capture_completed_at", + ) + approval_keys = ( + "approval_granted_at", + "approved_at", + "approval_completed_at", + ) + + start_ts = _parse_iso8601(_extract_first(meta, start_keys)) + approval_ts = _parse_iso8601(_extract_first(meta, approval_keys)) + event_timestamp = _parse_iso8601(event_ts) + + turnaround_seconds: Optional[float] = None + if start_ts is not None: + turnaround_seconds = (published_at - start_ts).total_seconds() + + approval_latency_seconds: Optional[float] = None + reference_ts = approval_ts or event_timestamp + if reference_ts is not None: + approval_latency_seconds = (published_at - reference_ts).total_seconds() + + engagement: Dict[str, float] = {} + for key in ("engagement", "analytics", "metrics"): + candidate = meta.get(key) + if isinstance(candidate, dict): + for metric_key, value in candidate.items(): + numeric = _coerce_numeric(value) + if numeric is None: + continue + engagement[metric_key] = engagement.get(metric_key, 0.0) + numeric + + cost: Dict[str, float] = {} + for key in ("cost", "spend", "usage"): + candidate = meta.get(key) + if isinstance(candidate, dict): + for cost_key, value in candidate.items(): + numeric = _coerce_numeric(value) + if numeric is None: + continue + cost[cost_key] = cost.get(cost_key, 0.0) + numeric + + return PublishTelemetry( + published_at=published_at, + turnaround_seconds=turnaround_seconds, + approval_latency_seconds=approval_latency_seconds, + engagement=engagement, + cost=cost, + ) + + +__all__ = [ + "PublisherMetrics", + "PublishTelemetry", + "compute_publish_telemetry", +] + diff --git a/pmoves/services/publisher-discord/main.py b/pmoves/services/publisher-discord/main.py index 3196ca8c50..a53a4ecc9a 100644 --- a/pmoves/services/publisher-discord/main.py +++ b/pmoves/services/publisher-discord/main.py @@ -1,11 +1,24 @@ -import os, json, asyncio, logging +import asyncio +import datetime +import json +import logging +import os +import re from collections import Counter -from typing import Dict, Any, Iterable, Optional +from typing import Any, Dict, Iterable, Optional import httpx -from fastapi import FastAPI, Body, HTTPException +from fastapi import Body, FastAPI, HTTPException from nats.aio.client import Client as NATS +try: # pragma: no cover - optional Supabase helper + from services.common import supabase as supabase_common +except Exception: # pragma: no cover - supabase is optional for local/dev + supabase_common = None # type: ignore[assignment] + +from services.common.telemetry import PublisherMetrics, PublishTelemetry, compute_publish_telemetry + + app = FastAPI(title="Publisher-Discord", version="0.1.0") DISCORD_WEBHOOK_URL = os.environ.get("DISCORD_WEBHOOK_URL", "") @@ -16,9 +29,12 @@ "DISCORD_SUBJECTS", "ingest.file.added.v1,ingest.transcript.ready.v1,ingest.summary.ready.v1,ingest.chapters.ready.v1,content.published.v1", ).split(",") +DISCORD_METRICS_TABLE = os.environ.get("DISCORD_METRICS_TABLE", "publisher_discord_metrics") +DISCORD_METRICS_CONFLICT = os.environ.get("DISCORD_METRICS_CONFLICT", "published_event_id") _nc: Optional[NATS] = None -_metrics = Counter() +_webhook_counters = Counter() +_telemetry_metrics = PublisherMetrics() logger = logging.getLogger("publisher_discord") @@ -69,22 +85,113 @@ def _pick_thumbnail(payload: Dict[str, Any]) -> Optional[str]: return ranked[0][1] return None + +def _coerce_text(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, str): + value = value.strip() + return value or None + return str(value) + + +def _safe_slug(*values: Optional[str]) -> str: + for value in values: + if not value: + continue + slug = re.sub(r"[^a-z0-9]+", "-", str(value).lower()).strip("-") + if slug: + return slug + return "discord-event" + + +def _webhook_snapshot() -> Dict[str, int]: + return { + "webhook_success": _webhook_counters.get("discord_webhook_success", 0), + "webhook_failures": _webhook_counters.get("discord_webhook_failures", 0), + "webhook_missing": _webhook_counters.get("discord_webhook_missing", 0), + } + + +def _record_publish_telemetry(telemetry: PublishTelemetry) -> None: + _telemetry_metrics.record_turnaround(telemetry.turnaround_seconds) + _telemetry_metrics.record_approval_latency(telemetry.approval_latency_seconds) + _telemetry_metrics.record_engagement(telemetry.engagement) + _telemetry_metrics.record_cost(telemetry.cost) + + +async def _persist_discord_rollup( + telemetry: PublishTelemetry, + payload: Dict[str, Any], + envelope: Dict[str, Any], + webhook_success: bool, +) -> None: + if supabase_common is None: + logger.debug("Supabase client unavailable; skipping Discord metrics rollup persistence") + return + + artifact_uri = _coerce_text(payload.get("artifact_uri")) or _coerce_text(payload.get("content_url")) + published_event_id = _coerce_text(envelope.get("id")) + if not artifact_uri: + artifact_uri = f"discord::{published_event_id or _safe_slug(payload.get('title'), payload.get('slug'))}" + + namespace = _coerce_text(payload.get("namespace") or payload.get("workspace") or "pmoves") or "pmoves" + slug = _safe_slug( + payload.get("slug"), + payload.get("title"), + payload.get("published_path"), + published_event_id, + ) + + row = telemetry.to_rollup_row( + artifact_uri=artifact_uri, + namespace=namespace, + slug=slug, + ) + row.update( + { + "published_event_id": published_event_id, + "event_topic": _coerce_text(envelope.get("topic") or envelope.get("subject")), + "channel": "discord", + "webhook_success": webhook_success, + } + ) + + try: + await asyncio.to_thread( + supabase_common.upsert_row, + DISCORD_METRICS_TABLE, + row, + DISCORD_METRICS_CONFLICT or None, + ) + except Exception as exc: # pragma: no cover - external dependency + logger.warning( + "Failed to persist Discord metrics rollup", + extra={"table": DISCORD_METRICS_TABLE, "row": row}, + exc_info=exc, + ) + @app.get("/healthz") async def healthz(): return { "ok": True, "webhook": bool(DISCORD_WEBHOOK_URL), - "metrics": { - "webhook_success": _metrics.get("discord_webhook_success", 0), - "webhook_failures": _metrics.get("discord_webhook_failures", 0), - "webhook_missing": _metrics.get("discord_webhook_missing", 0), - }, + "metrics": _webhook_snapshot(), + "telemetry": _telemetry_metrics.summary(), + } + + +@app.get("/metrics") +async def metrics(): + return { + "webhook": _webhook_snapshot(), + "telemetry": _telemetry_metrics.summary(), } async def _post_discord(content: Optional[str], embeds: Optional[list] = None, retries: int = 3): if not DISCORD_WEBHOOK_URL: logger.warning("discord_webhook_missing", extra={"event": "discord_webhook_missing"}) - _metrics["discord_webhook_missing"] += 1 + _webhook_counters["discord_webhook_missing"] += 1 return False payload = {"username": DISCORD_USERNAME} if DISCORD_AVATAR_URL: @@ -111,7 +218,7 @@ async def _post_discord(content: Optional[str], embeds: Optional[list] = None, r backoff = min(backoff * 2.0, 8.0) continue if r.status_code in (200, 204): - _metrics["discord_webhook_success"] += 1 + _webhook_counters["discord_webhook_success"] += 1 return True if r.status_code == 429: try: @@ -134,13 +241,13 @@ async def _post_discord(content: Optional[str], embeds: Optional[list] = None, r "body": r.text[:256], }, ) - _metrics["discord_webhook_failures"] += 1 + _webhook_counters["discord_webhook_failures"] += 1 return False logger.warning( "discord_webhook_failed", extra={"event": "discord_webhook_failed", "status_code": None, "attempt": retries}, ) - _metrics["discord_webhook_failures"] += 1 + _webhook_counters["discord_webhook_failures"] += 1 return False def _format_event(name: str, payload: Dict[str, Any]) -> Dict[str, Any]: @@ -238,11 +345,24 @@ async def startup(): async def handler(msg): try: data = json.loads(msg.data.decode("utf-8")) - name = data.get("topic") or msg.subject - payload = data.get("payload") or data + envelope: Dict[str, Any] = data if isinstance(data, dict) else {} except Exception: - name = msg.subject - payload = {"raw": msg.data.decode("utf-8",errors="ignore")} + envelope = {} + + name = envelope.get("topic") or msg.subject + payload = envelope.get("payload") if isinstance(envelope.get("payload"), dict) else envelope or {} + if not isinstance(payload, dict): + payload = {"raw": msg.data.decode("utf-8", errors="ignore")} + + meta = payload.get("meta") if isinstance(payload.get("meta"), dict) else None + published_at = datetime.datetime.now(datetime.timezone.utc) + telemetry = compute_publish_telemetry( + meta, + envelope.get("ts") if isinstance(envelope, dict) else None, + published_at, + ) + _record_publish_telemetry(telemetry) + rendered = _format_event(name, payload) ok = await _post_discord(rendered.get("content"), rendered.get("embeds")) if not ok: @@ -254,6 +374,20 @@ async def handler(msg): "nats_subject": msg.subject, }, ) + + await _persist_discord_rollup(telemetry, payload, envelope if isinstance(envelope, dict) else {}, ok) + logger.info( + "discord_event_processed", + extra={ + "subject": name, + "nats_subject": msg.subject, + "webhook_success": ok, + "metrics": { + "webhook": _webhook_snapshot(), + "telemetry": _telemetry_metrics.summary(), + }, + }, + ) for subj in SUBJECTS: s = subj.strip() if not s: diff --git a/pmoves/services/publisher/README.md b/pmoves/services/publisher/README.md index 5127c6e324..3f7325ef63 100644 --- a/pmoves/services/publisher/README.md +++ b/pmoves/services/publisher/README.md @@ -91,6 +91,9 @@ each artifact, aggregated by namespace. Average turnaround and approval latency highlight operational friction; spikes should trigger reviews of automation queues or manual approval load. +See `pmoves/docs/TELEMETRY_ROI.md` for step-by-step guidance on charting the +rollup tables and pairing them with Discord delivery telemetry. + ## Local Smoke Test 1. Ensure the PMOVES stack is running (`make up`). diff --git a/pmoves/services/publisher/publisher.py b/pmoves/services/publisher/publisher.py index 909054a325..377554f0fd 100644 --- a/pmoves/services/publisher/publisher.py +++ b/pmoves/services/publisher/publisher.py @@ -10,7 +10,7 @@ import pathlib import re import unicodedata -from dataclasses import asdict, dataclass, field +from dataclasses import asdict from typing import Any, Dict, Optional, Tuple, TYPE_CHECKING from urllib.parse import urljoin @@ -75,6 +75,12 @@ def envelope( except Exception: # pragma: no cover - supabase is optional for local/dev testing supabase_client = None # type: ignore[assignment] +from services.common.telemetry import ( + PublisherMetrics, + PublishTelemetry, + compute_publish_telemetry, +) + NATS_URL = os.environ.get("NATS_URL", "nats://nats:4222") MINIO_ENDPOINT = os.environ.get("MINIO_ENDPOINT", "minio:9000") @@ -109,88 +115,6 @@ def _configure_logging() -> None: ) -@dataclass -class PublisherMetrics: - downloads: int = 0 - download_failures: int = 0 - refresh_attempts: int = 0 - refresh_success: int = 0 - refresh_failures: int = 0 - turnaround_samples: int = 0 - total_turnaround_seconds: float = 0.0 - max_turnaround_seconds: float = 0.0 - approval_latency_samples: int = 0 - total_approval_latency_seconds: float = 0.0 - max_approval_latency_seconds: float = 0.0 - engagement_events: int = 0 - engagement_totals: Dict[str, float] = field(default_factory=dict) - cost_events: int = 0 - cost_totals: Dict[str, float] = field(default_factory=dict) - - def record_download_success(self) -> None: - self.downloads += 1 - - def record_download_failure(self) -> None: - self.download_failures += 1 - - def record_refresh_attempt(self) -> None: - self.refresh_attempts += 1 - - def record_refresh_success(self) -> None: - self.refresh_success += 1 - - def record_refresh_failure(self) -> None: - self.refresh_failures += 1 - - def record_turnaround(self, seconds: Optional[float]) -> None: - if seconds is None or seconds < 0: - return - self.turnaround_samples += 1 - self.total_turnaround_seconds += seconds - if seconds > self.max_turnaround_seconds: - self.max_turnaround_seconds = seconds - - def record_approval_latency(self, seconds: Optional[float]) -> None: - if seconds is None or seconds < 0: - return - self.approval_latency_samples += 1 - self.total_approval_latency_seconds += seconds - if seconds > self.max_approval_latency_seconds: - self.max_approval_latency_seconds = seconds - - def record_engagement(self, engagement: Dict[str, float]) -> None: - if not engagement: - return - self.engagement_events += 1 - for key, value in engagement.items(): - try: - numeric = float(value) - except (TypeError, ValueError): - continue - self.engagement_totals[key] = self.engagement_totals.get(key, 0.0) + numeric - - def record_cost(self, cost: Dict[str, float]) -> None: - if not cost: - return - self.cost_events += 1 - for key, value in cost.items(): - try: - numeric = float(value) - except (TypeError, ValueError): - continue - self.cost_totals[key] = self.cost_totals.get(key, 0.0) + numeric - - def summary(self) -> Dict[str, Any]: - data = asdict(self) - if self.turnaround_samples: - data["avg_turnaround_seconds"] = self.total_turnaround_seconds / self.turnaround_samples - if self.approval_latency_samples: - data["avg_approval_latency_seconds"] = ( - self.total_approval_latency_seconds / self.approval_latency_samples - ) - return data - - METRICS = PublisherMetrics() _METRICS_SERVER: Optional[asyncio.AbstractServer] = None @@ -202,44 +126,6 @@ class DownloadError(Exception): class JellyfinRefreshError(Exception): """Raised when Jellyfin fails to refresh or respond.""" - - -@dataclass -class PublishTelemetry: - published_at: datetime.datetime - turnaround_seconds: Optional[float] - approval_latency_seconds: Optional[float] - engagement: Dict[str, float] - cost: Dict[str, float] - - def to_meta(self) -> Dict[str, Any]: - meta: Dict[str, Any] = { - "published_at": self.published_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"), - } - if self.turnaround_seconds is not None: - meta["turnaround_seconds"] = self.turnaround_seconds - if self.approval_latency_seconds is not None: - meta["approval_to_publish_seconds"] = self.approval_latency_seconds - if self.engagement: - meta["engagement"] = self.engagement - if self.cost: - meta["cost"] = self.cost - return meta - - def to_rollup_row(self, *, artifact_uri: str, namespace: str, slug: str) -> Dict[str, Any]: - return { - "artifact_uri": artifact_uri, - "namespace": namespace, - "slug": slug, - "published_at": self.published_at.isoformat(), - "turnaround_seconds": self.turnaround_seconds, - "approval_latency_seconds": self.approval_latency_seconds, - "engagement": self.engagement or None, - "cost": self.cost or None, - } - - - def _utc_now_iso() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat() @@ -481,100 +367,6 @@ async def download_with_retries(minio: MinioClientType, bucket: str, key: str, d raise DownloadError(f"Failed to download s3://{bucket}/{key}") from last_error -def _parse_iso8601(value: Optional[Any]) -> Optional[datetime.datetime]: - if not value or not isinstance(value, str): - return None - value = value.strip() - if not value: - return None - try: - if value.endswith("Z"): - value = value[:-1] + "+00:00" - return datetime.datetime.fromisoformat(value) - except ValueError: - return None - - -def _coerce_numeric(value: Any) -> Optional[float]: - try: - if value is None: - return None - numeric = float(value) - if numeric != numeric: # NaN - return None - return numeric - except (TypeError, ValueError): - return None - - -def _extract_first(meta: Dict[str, Any], keys: Tuple[str, ...]) -> Optional[str]: - for key in keys: - candidate = meta.get(key) - if isinstance(candidate, str) and candidate: - return candidate - return None - - -def compute_publish_telemetry( - incoming_meta: Optional[Dict[str, Any]], - event_ts: Optional[str], - published_at: datetime.datetime, -) -> PublishTelemetry: - meta = incoming_meta or {} - start_keys = ( - "ingest_started_at", - "submitted_at", - "created_at", - "capture_completed_at", - ) - approval_keys = ( - "approval_granted_at", - "approved_at", - "approval_completed_at", - ) - - start_ts = _parse_iso8601(_extract_first(meta, start_keys)) - approval_ts = _parse_iso8601(_extract_first(meta, approval_keys)) - event_timestamp = _parse_iso8601(event_ts) - - turnaround_seconds: Optional[float] = None - if start_ts is not None: - turnaround_seconds = (published_at - start_ts).total_seconds() - - approval_latency_seconds: Optional[float] = None - reference_ts = approval_ts or event_timestamp - if reference_ts is not None: - approval_latency_seconds = (published_at - reference_ts).total_seconds() - - engagement: Dict[str, float] = {} - for key in ("engagement", "analytics", "metrics"): - candidate = meta.get(key) - if isinstance(candidate, dict): - for metric_key, value in candidate.items(): - numeric = _coerce_numeric(value) - if numeric is None: - continue - engagement[metric_key] = engagement.get(metric_key, 0.0) + numeric - - cost: Dict[str, float] = {} - for key in ("cost", "spend", "usage"): - candidate = meta.get(key) - if isinstance(candidate, dict): - for cost_key, value in candidate.items(): - numeric = _coerce_numeric(value) - if numeric is None: - continue - cost[cost_key] = cost.get(cost_key, 0.0) + numeric - - return PublishTelemetry( - published_at=published_at, - turnaround_seconds=turnaround_seconds, - approval_latency_seconds=approval_latency_seconds, - engagement=engagement, - cost=cost, - ) - - async def persist_publish_rollup(row: Dict[str, Any]) -> None: if supabase_common is None: logger.debug("Supabase client unavailable; skipping metrics rollup persistence") @@ -922,13 +714,10 @@ async def handle(msg): "event_id": env.get("id"), "correlation_id": env.get("correlation_id"), "public_url": published_payload.get("public_url"), - - "metrics": METRICS.summary(), - + "metrics_summary": METRICS.summary(), "publish_event_id": publish_event_id, "reviewer": reviewer, - "metrics": asdict(METRICS), - + "metrics_state": asdict(METRICS), }, )