| """Append-only, fail-open production telemetry. |
| |
| This module is deliberately *not* part of the media/control path: |
| |
| * Public calls only update in-memory counters or put a small record on a |
| bounded queue. They never write to disk, never wait for I/O, and never raise |
| into a relay/manager thread. |
| * A single daemon writer appends compact JSON lines. If storage is slow or |
| unavailable, telemetry records are dropped and counted; the stream is not. |
| * There is intentionally no rotate/truncate/delete API. Records live under the |
| persistent data root (``<DATA_ROOT>/telemetry/telemetry.jsonl``) and are |
| removed only by an explicit, manual operator action outside this code. |
| * No media, signed URLs, stream keys, cookies, tokens, or credentials are |
| recorded. Potentially sensitive keys/values are stripped defensively. |
| |
| Set ``TELEMETRY_ENABLED=0`` to disable collection without touching streaming. |
| """ |
|
|
| import json |
| import math |
| import os |
| import queue |
| import re |
| import threading |
| import time |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
|
|
| _QUEUE_MAX = 4096 |
| _SENSITIVE_PARTS = ( |
| "url", "token", "cookie", "authorization", "sessionid", "password", |
| "secret", "stream_key", "rtmp_key", "access_key", "refresh_token", |
| ) |
| _URL_RE = re.compile(r"(?i)\b(?:rtmps?|https?)://\S+") |
| _SECRET_ASSIGNMENT_RE = re.compile( |
| r"(?i)\b(sessionid|token|stream[_ -]?key|authorization|sign|signature)\s*[=:]\s*[^\s;&]+" |
| ) |
|
|
|
|
| def _iso_now(): |
| return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") |
|
|
|
|
| def _enabled(): |
| return str(os.getenv("TELEMETRY_ENABLED", "1")).strip().lower() not in ( |
| "0", "false", "off", "no", "" |
| ) |
|
|
|
|
| def _default_path(): |
| explicit = str(os.getenv("TELEMETRY_PATH", "") or "").strip() |
| if explicit: |
| return Path(explicit).expanduser() |
|
|
| events_path = str(os.getenv("STREAM_EVENTS_PATH", "") or "").strip() |
| if events_path: |
| events = Path(events_path).expanduser() |
| root = events.parent.parent if events.parent.name == "logs" else events.parent |
| return root / "telemetry" / "telemetry.jsonl" |
|
|
| data_dir = str(os.getenv("DATA_DIR", "") or "").strip() |
| if data_dir: |
| return Path(data_dir).expanduser() / "telemetry" / "telemetry.jsonl" |
| if Path("/data").is_dir(): |
| return Path("/data/tiktok-relay/telemetry/telemetry.jsonl") |
| return Path("./storage/telemetry/telemetry.jsonl") |
|
|
|
|
| def _safe(value, depth=0): |
| if depth > 5: |
| return "…" |
| if isinstance(value, dict): |
| out = {} |
| for key, item in list(value.items())[:80]: |
| name = str(key)[:80] |
| lowered = name.lower() |
| if any(part in lowered for part in _SENSITIVE_PARTS): |
| continue |
| out[name] = _safe(item, depth + 1) |
| return out |
| if isinstance(value, (list, tuple)): |
| return [_safe(item, depth + 1) for item in value[:30]] |
| if isinstance(value, str): |
| value = _URL_RE.sub("«hidden-url»", value) |
| value = _SECRET_ASSIGNMENT_RE.sub(r"\1=«hidden»", value) |
| return value[:300] |
| if isinstance(value, bool) or value is None: |
| return value |
| if isinstance(value, (int, float)): |
| return value if math.isfinite(float(value)) else None |
| return _safe(str(value), depth + 1) |
|
|
|
|
| def http_status_from_exception(error): |
| """Best-effort HTTP status from curl/httpx/requests-style exceptions.""" |
| response = getattr(error, "response", None) |
| for candidate in ( |
| getattr(response, "status_code", None), |
| getattr(error, "status_code", None), |
| getattr(error, "status", None), |
| ): |
| try: |
| status = int(candidate) |
| except (TypeError, ValueError): |
| continue |
| if 100 <= status < 600: |
| return status |
| |
| |
| match = re.search(r"\bHTTP(?:/[\d.]+)?\s+([1-5]\d\d)\b", str(error or ""), re.I) |
| if match: |
| try: |
| return int(match.group(1)) |
| except ValueError: |
| return 0 |
| return 0 |
|
|
|
|
| class _CpuSampler: |
| """Cached cgroup CPU sampler; reads tiny proc files at most every 10s.""" |
|
|
| def __init__(self): |
| self._lock = threading.Lock() |
| self._last_at = 0.0 |
| self._last_usage_seconds = None |
| self._cached = {} |
|
|
| @staticmethod |
| def _read_text(path): |
| try: |
| return Path(path).read_text(encoding="utf-8", errors="ignore") |
| except OSError: |
| return "" |
|
|
| def _usage_seconds(self): |
| text = self._read_text("/sys/fs/cgroup/cpu.stat") |
| for line in text.splitlines(): |
| if line.startswith("usage_usec"): |
| try: |
| return int(line.split()[1]) / 1_000_000.0 |
| except (IndexError, ValueError): |
| pass |
| for path in ( |
| "/sys/fs/cgroup/cpuacct/cpuacct.usage", |
| "/sys/fs/cgroup/cpu,cpuacct/cpuacct.usage", |
| ): |
| text = self._read_text(path).strip() |
| if text: |
| try: |
| return int(text) / 1_000_000_000.0 |
| except ValueError: |
| pass |
| return None |
|
|
| def _quota_cpus(self): |
| text = self._read_text("/sys/fs/cgroup/cpu.max").split() |
| if len(text) >= 2 and text[0] != "max": |
| try: |
| return max(0.01, int(text[0]) / int(text[1])) |
| except (ValueError, ZeroDivisionError): |
| pass |
| try: |
| quota = int(self._read_text("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").strip()) |
| period = int(self._read_text("/sys/fs/cgroup/cpu/cpu.cfs_period_us").strip()) |
| if quota > 0 and period > 0: |
| return quota / period |
| except ValueError: |
| pass |
| return float(os.cpu_count() or 1) |
|
|
| def fields(self): |
| now = time.monotonic() |
| with self._lock: |
| if self._cached and now - self._last_at < 10.0: |
| return dict(self._cached) |
| out = {} |
| try: |
| out["load1"] = round(float(os.getloadavg()[0]), 3) |
| except (AttributeError, OSError, ValueError): |
| pass |
| usage = self._usage_seconds() |
| if usage is not None and self._last_usage_seconds is not None: |
| elapsed = now - self._last_at |
| if elapsed > 0: |
| cores = max(0.0, (usage - self._last_usage_seconds) / elapsed) |
| quota = self._quota_cpus() |
| out["cpu_cores_used"] = round(cores, 4) |
| out["cpu_pct_of_quota"] = round(min(999.0, 100.0 * cores / quota), 2) if quota else None |
| if usage is not None: |
| self._last_usage_seconds = usage |
| |
| |
| self._last_at = now |
| self._cached = dict(out) |
| return out |
|
|
|
|
| class Telemetry: |
| """In-memory session aggregation plus a non-blocking append-only writer.""" |
|
|
| def __init__(self, path=None): |
| self.path = Path(path).expanduser() if path else _default_path() |
| self.enabled = _enabled() |
| self._queue = queue.Queue(maxsize=_QUEUE_MAX) if self.enabled else None |
| self._lock = threading.RLock() |
| self._writer_lock = threading.Lock() |
| self._writer = None |
| self._sessions = {} |
| self._sequence = 0 |
| self.dropped_records = 0 |
| self.write_errors = 0 |
| self._cpu = _CpuSampler() |
|
|
| |
| def _ensure_writer(self): |
| if not self.enabled: |
| return False |
| writer = self._writer |
| if writer is not None and writer.is_alive(): |
| return True |
| with self._writer_lock: |
| writer = self._writer |
| if writer is not None and writer.is_alive(): |
| return True |
| self._writer = threading.Thread( |
| target=self._writer_loop, name="telemetry-writer", daemon=True |
| ) |
| self._writer.start() |
| return True |
|
|
| def _writer_loop(self): |
| while True: |
| record = self._queue.get() |
| try: |
| line = json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n" |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| with self.path.open("a", encoding="utf-8") as handle: |
| handle.write(line) |
| handle.flush() |
| if record.get("type") in ("session_finish", "session_start"): |
| try: |
| os.fsync(handle.fileno()) |
| except OSError: |
| pass |
| try: |
| os.chmod(self.path, 0o600) |
| except OSError: |
| pass |
| except Exception: |
| with self._lock: |
| self.write_errors += 1 |
| finally: |
| self._queue.task_done() |
|
|
| def _enqueue(self, record): |
| if not self.enabled: |
| return False |
| try: |
| if not self._ensure_writer(): |
| return False |
| self._queue.put_nowait(_safe(record)) |
| return True |
| except queue.Full: |
| with self._lock: |
| self.dropped_records += 1 |
| return False |
| except Exception: |
| return False |
|
|
| |
| @staticmethod |
| def _new_stat(value): |
| return {"count": 1, "sum": float(value), "min": float(value), "max": float(value), "last": float(value)} |
|
|
| def _update_session_locked(self, session, fields): |
| now = time.monotonic() |
| stats = session["stats"] |
| last = session["last"] |
| for name, value in (fields or {}).items(): |
| name = str(name)[:80] |
| if isinstance(value, bool) or value is None: |
| if value is not None: |
| last[name] = value |
| continue |
| if isinstance(value, (int, float)) and math.isfinite(float(value)): |
| number = float(value) |
| stat = stats.get(name) |
| if stat is None: |
| stats[name] = self._new_stat(number) |
| else: |
| stat["count"] += 1 |
| stat["sum"] += number |
| stat["min"] = min(stat["min"], number) |
| stat["max"] = max(stat["max"], number) |
| stat["last"] = number |
| last[name] = number |
| if name == "ffmpeg_speed": |
| if number < 0.95: |
| session["counters"]["ffmpeg_speed_below_0_95_samples"] = ( |
| session["counters"].get("ffmpeg_speed_below_0_95_samples", 0) + 1 |
| ) |
| if number < 1.0: |
| session["counters"]["ffmpeg_speed_below_1_samples"] = ( |
| session["counters"].get("ffmpeg_speed_below_1_samples", 0) + 1 |
| ) |
| elif isinstance(value, (str, int, float)): |
| last[name] = _safe(value) |
|
|
| minute = int(max(0.0, now - session["started_monotonic"]) // 60) |
| bucket = session["minutes"].setdefault( |
| str(minute), |
| {"samples": 0, "ffmpeg_speed_sum": 0.0, "ffmpeg_speed_samples": 0, |
| "ffmpeg_speed_below_0_95": 0, "source_bps_sum": 0.0, "source_bps_samples": 0}, |
| ) |
| bucket["samples"] += 1 |
| if isinstance((fields or {}).get("ffmpeg_speed"), (int, float)): |
| speed = float(fields["ffmpeg_speed"]) |
| bucket["ffmpeg_speed_sum"] += speed |
| bucket["ffmpeg_speed_samples"] += 1 |
| if speed < 0.95: |
| bucket["ffmpeg_speed_below_0_95"] += 1 |
| if isinstance((fields or {}).get("source_bps"), (int, float)): |
| bucket["source_bps_sum"] += float(fields["source_bps"]) |
| bucket["source_bps_samples"] += 1 |
| session["last_observed_at"] = _iso_now() |
| session["last_observed_monotonic"] = now |
|
|
| def _summary_locked(self, session, *, finished=False, reason="", extra=None, include_minutes=True): |
| now = time.monotonic() |
| stats = {} |
| for name, stat in session["stats"].items(): |
| count = int(stat["count"]) |
| stats[name] = { |
| "count": count, |
| "avg": round(stat["sum"] / count, 4) if count else None, |
| "min": round(stat["min"], 4), |
| "max": round(stat["max"], 4), |
| "last": round(stat["last"], 4), |
| } |
| minutes = [] |
| if include_minutes: |
| for minute, bucket in sorted(session["minutes"].items(), key=lambda item: int(item[0])): |
| speed_samples = int(bucket["ffmpeg_speed_samples"]) |
| source_samples = int(bucket["source_bps_samples"]) |
| minutes.append({ |
| "minute": int(minute), |
| "samples": int(bucket["samples"]), |
| "ffmpeg_speed_avg": round(bucket["ffmpeg_speed_sum"] / speed_samples, 4) if speed_samples else None, |
| "ffmpeg_speed_below_0_95": int(bucket["ffmpeg_speed_below_0_95"]), |
| "source_bps_avg": round(bucket["source_bps_sum"] / source_samples, 1) if source_samples else None, |
| }) |
| return _safe({ |
| "session_key": session["key"], |
| "broadcaster": session["broadcaster"], |
| "started_at": session["started_at"], |
| "ended_at": _iso_now() if finished else None, |
| "duration_seconds": round(max(0.0, now - session["started_monotonic"]), 3), |
| "meta": session["meta"], |
| "counters": dict(session["counters"]), |
| "stats": stats, |
| "last": dict(session["last"]), |
| "minutes": minutes, |
| "minutes_omitted": not include_minutes, |
| "reason": str(reason or "")[:200], |
| "extra": extra or {}, |
| "host": self._cpu.fields(), |
| "telemetry_dropped_records": self.dropped_records, |
| "telemetry_write_errors": self.write_errors, |
| }) |
|
|
| |
| def start_session(self, broadcaster, meta=None): |
| if not self.enabled: |
| return None |
| try: |
| user = str(broadcaster or "system")[:100] |
| with self._lock: |
| self._sequence += 1 |
| key = "%s:%d:%d" % (user, int(time.time() * 1000), self._sequence) |
| session = { |
| "key": key, |
| "broadcaster": user, |
| "started_at": _iso_now(), |
| "started_monotonic": time.monotonic(), |
| "meta": _safe(meta or {}), |
| "counters": {}, |
| "stats": {}, |
| "last": {}, |
| "minutes": {}, |
| "dedupe": {}, |
| "last_observed_at": None, |
| "last_observed_monotonic": time.monotonic(), |
| } |
| self._sessions[key] = session |
| summary = self._summary_locked(session) |
| self._enqueue({ |
| "type": "session_start", "recorded_at": _iso_now(), |
| "session_key": key, "broadcaster": user, "summary": summary, |
| }) |
| return key |
| except Exception: |
| return None |
|
|
| def observe(self, session_key, fields=None): |
| if not self.enabled or not session_key: |
| return False |
| try: |
| with self._lock: |
| session = self._sessions.get(session_key) |
| if session is None: |
| return False |
| self._update_session_locked(session, fields or {}) |
| return True |
| except Exception: |
| return False |
|
|
| def event(self, session_key, broadcaster, event, details=None, dedupe_key=None): |
| if not self.enabled: |
| return False |
| try: |
| name = str(event or "UNKNOWN")[:80] |
| safe_details = _safe(details or {}) |
| elapsed = None |
| with self._lock: |
| session = self._sessions.get(session_key) if session_key else None |
| if session is not None: |
| counters = session["counters"] |
| counters[name] = counters.get(name, 0) + 1 |
| if name == "SOURCE_OPEN_ERROR": |
| status = (safe_details or {}).get("http_status") |
| if status: |
| label = "source_open_http_%s" % status |
| counters[label] = counters.get(label, 0) + 1 |
| elapsed = round(max(0.0, time.monotonic() - session["started_monotonic"]), 3) |
| if dedupe_key: |
| previous = session["dedupe"].get(str(dedupe_key)) |
| if previous == safe_details: |
| label = name + "_suppressed" |
| counters[label] = counters.get(label, 0) + 1 |
| return True |
| session["dedupe"][str(dedupe_key)] = safe_details |
| return self._enqueue({ |
| "type": "event", "recorded_at": _iso_now(), "session_key": session_key, |
| "broadcaster": str(broadcaster or "system")[:100], "event": name, |
| "elapsed_seconds": elapsed, "details": safe_details, |
| }) |
| except Exception: |
| return False |
|
|
| def checkpoint(self, session_key, reason="periodic"): |
| if not self.enabled or not session_key: |
| return False |
| try: |
| with self._lock: |
| session = self._sessions.get(session_key) |
| if session is None: |
| return False |
| summary = self._summary_locked(session, reason=reason, include_minutes=False) |
| user = session["broadcaster"] |
| return self._enqueue({ |
| "type": "checkpoint", "recorded_at": _iso_now(), "session_key": session_key, |
| "broadcaster": user, "summary": summary, |
| }) |
| except Exception: |
| return False |
|
|
| def finish(self, session_key, reason="", extra=None): |
| if not self.enabled or not session_key: |
| return False |
| try: |
| with self._lock: |
| session = self._sessions.pop(session_key, None) |
| if session is None: |
| return False |
| summary = self._summary_locked( |
| session, finished=True, reason=reason, extra=_safe(extra or {}) |
| ) |
| user = session["broadcaster"] |
| return self._enqueue({ |
| "type": "session_finish", "recorded_at": _iso_now(), "session_key": session_key, |
| "broadcaster": user, "summary": summary, |
| }) |
| except Exception: |
| return False |
|
|
| def read_records(self, limit=200, broadcaster=None): |
| """Read recent records for the authenticated diagnostics API. |
| |
| This is an operator/read path only; it is never called by media code. |
| It reads a bounded tail so a long-lived append-only file cannot make |
| the request consume unbounded memory. |
| """ |
| try: |
| limit = max(1, min(int(limit), 5000)) |
| except (TypeError, ValueError): |
| limit = 200 |
| try: |
| if not self.path.exists(): |
| return [] |
| size = self.path.stat().st_size |
| max_bytes = 16 * 1024 * 1024 |
| offset = max(0, size - max_bytes) |
| with self.path.open("rb") as handle: |
| handle.seek(offset) |
| raw = handle.read() |
| if offset: |
| newline = raw.find(b"\n") |
| raw = raw[newline + 1:] if newline >= 0 else b"" |
| records = [] |
| wanted = str(broadcaster or "").strip().lstrip("@") |
| for line in raw.decode("utf-8", "ignore").splitlines(): |
| try: |
| item = json.loads(line) |
| except ValueError: |
| continue |
| if not isinstance(item, dict): |
| continue |
| if wanted: |
| summary = item.get("summary") or {} |
| if item.get("broadcaster") != wanted and summary.get("broadcaster") != wanted: |
| continue |
| records.append(item) |
| return records[-limit:] |
| except OSError: |
| return [] |
|
|
| def status(self): |
| with self._lock: |
| return { |
| "enabled": bool(self.enabled), |
| "file": "telemetry/telemetry.jsonl", |
| "active_sessions": len(self._sessions), |
| "dropped_records": self.dropped_records, |
| "write_errors": self.write_errors, |
| "queued_records": self._queue.qsize() if self._queue is not None else 0, |
| } |
|
|
|
|
| telemetry = Telemetry() |
|
|
|
|
| def start_session(broadcaster, meta=None): |
| try: |
| return telemetry.start_session(broadcaster, meta) |
| except Exception: |
| return None |
|
|
|
|
| def observe(session_key, fields=None): |
| try: |
| return telemetry.observe(session_key, fields) |
| except Exception: |
| return False |
|
|
|
|
| def record_event(session_key, broadcaster, event, details=None, dedupe_key=None): |
| try: |
| return telemetry.event(session_key, broadcaster, event, details, dedupe_key=dedupe_key) |
| except Exception: |
| return False |
|
|
|
|
| def checkpoint(session_key, reason="periodic"): |
| try: |
| return telemetry.checkpoint(session_key, reason) |
| except Exception: |
| return False |
|
|
|
|
| def finish(session_key, reason="", extra=None): |
| try: |
| return telemetry.finish(session_key, reason, extra) |
| except Exception: |
| return False |
|
|
|
|
| def read_records(limit=200, broadcaster=None): |
| try: |
| return telemetry.read_records(limit, broadcaster) |
| except Exception: |
| return [] |
|
|
|
|
| def status(): |
| try: |
| return telemetry.status() |
| except Exception: |
| return {"enabled": False, "file": "telemetry/telemetry.jsonl", "active_sessions": 0} |
|
|
