# (c) cavaliba.com - data - telemetry.py import hashlib import hmac import math import random import re import secrets import statistics import time from collections import Counter, defaultdict from contextvars import ContextVar from datetime import UTC, datetime, timedelta from django.core.cache import cache as django_cache from django.db import IntegrityError, transaction from django.db.models import Count, Max, Min from django.utils import timezone from app_data.models import ( DataEAV, DataFile, DataInstance, DataRegistry, DataRevision, DataTask, DataTelemetryAggregate, DataTelemetryRaw, DataTelemetryUnique, ) from app_data.registry import registry_get_key, registry_set_key from app_home.configuration import get_configuration from app_home.log import log2 from app_home.models import CavalibaEvent # ---------------------------------------------------------------------------------- # Vocabulary # ---------------------------------------------------------------------------------- TYPES = ["counter", "gauge", "timer", "set", "histogram", "unique"] PERIODS = ["minute", "hour", "day", "week", "month", "quarter", "year"] # first function is the default one for the type TYPE_FUNCTIONS = { "counter": ["sum", "count"], "gauge": ["last", "min", "max", "mean", "count"], "timer": ["p95", "count", "sum", "min", "max", "mean", "median", "p90", "p99"], "set": ["breakdown", "count", "distinct"], "histogram": ["buckets", "count", "sum", "min", "max", "mean"], "unique": ["distinct"], } UNIQUE_PERIODS = ["day", "week", "month"] DEFAULT_BUCKETS = [1, 10, 100, 1000, 10000] DEFAULT_MAX_IDS = 20 KEY_REGEX = re.compile(r"^[a-z0-9_]+(\.[a-z0-9_-]+)*$") KEY_MAX_LENGTH = 128 ID_REGEX = re.compile(r"^[\w.:/@-]*$") ID_MAX_LENGTH = 64 SVALUE_MAX_LENGTH = 128 # ---------------------------------------------------------------------------------- # Declared keys (authoritative for type/options) ; undeclared keys use the call type # ---------------------------------------------------------------------------------- TELEMETRY_KEYS = { "auth.login": { "type": "counter", "description": "login or new federated session, id=auth mode", }, "auth.login.failure": {"type": "counter", "description": "login form failure, id=local"}, "users.active": {"type": "unique", "description": "distinct active users"}, "api.request": {"type": "counter", "description": "API call, id=endpoint", "max_ids": 50}, "task.duration": { "type": "timer", "unit": "ms", "description": "celery task duration, id=task", "max_ids": 50, }, # HTTP requests "view.request": { "type": "counter", "description": "web view hit, id=app.view", "max_ids": 50, }, "ajax.request": {"type": "counter", "description": "ajax call, id=endpoint", "max_ids": 50}, "http.duration": { "type": "timer", "unit": "ms", "description": "response time, id=app.view (empty id = all requests)", "max_ids": 50, }, "http.exception": { "type": "counter", "description": "unhandled view exception (5xx), id=app.view", "max_ids": 50, }, # engine "log.event": {"type": "counter", "description": "retained log event, id=level"}, "cache.get": { "type": "counter", "description": "cache read, id=hit_local|hit_shared|miss", }, "cache.session": { "type": "counter", "description": "aaa session cache, id=hit|miss|nocache", }, "data.write": { "type": "counter", "description": "instance write (create/update/delete), id=schema", "max_ids": 50, }, "data.write.action": { "type": "set", "description": "instance write breakdown per action", }, "data.objects": { "type": "gauge", "description": "instances per schema, id=schema (sampled)", "max_ids": 50, }, "data.objects.total": {"type": "gauge", "description": "total instances (sampled)"}, "data.rows": { "type": "gauge", "description": "table size, id=table (sampled)", "max_ids": 50, }, } # ---------------------------------------------------------------------------------- # Per-process caches # ---------------------------------------------------------------------------------- KNOWN_KEYS_TTL = 60 UNIQUE_MEMO_MAX = 100000 # registry key of the unique-member hash salt: generated once, shared by all processes SALT_REGISTRY_KEY = "TELEMETRY_SALT" _known_keys = {"keys": set(), "ts": 0.0} _unique_memo = {} _salt = {"value": None} _buffer = {} # buffered counters, see telemetry_bump() _request = ContextVar("telemetry_request", default=None) # per-request timer, see request_id() def reset_cache(): """Clear per-process caches (known keys, unique memo, salt, counter buffer).""" _known_keys["keys"] = set() _known_keys["ts"] = 0.0 _unique_memo.clear() _salt["value"] = None _buffer.clear() # ---------------------------------------------------------------------------------- # Time buckets # ---------------------------------------------------------------------------------- def bucket_start(dt, period): """Return the start of the period bucket containing dt (local time zone boundaries) as a UTC datetime: an ambiguous local time (DST fall-back hour) never compares equal to the same instant in another zone, so buckets are compared and keyed in UTC.""" dt = timezone.localtime(dt) if period == "minute": start = dt.replace(second=0, microsecond=0) elif period == "hour": start = dt.replace(minute=0, second=0, microsecond=0) elif period in ("day", "week", "month", "quarter", "year"): start = dt.replace(hour=0, minute=0, second=0, microsecond=0) if period == "week": start = start - timedelta(days=start.weekday()) elif period == "month": start = start.replace(day=1) elif period == "quarter": start = start.replace(month=(start.month - 1) // 3 * 3 + 1, day=1) elif period == "year": start = start.replace(month=1, day=1) else: raise ValueError(f"unknown period: {period}") return start.astimezone(UTC) # ---------------------------------------------------------------------------------- # Record # ---------------------------------------------------------------------------------- def telemetry(key, type=None, id="", value=None, option=None): """Record a telemetry event; return True if recorded or sampled out, False if dropped. Never raises.""" try: return _record(key, type=type, id=id, value=value, option=option or {}) except Exception as e: log2.debug("telemetry %s: error %s", key, e) return False def on_request_started(sender=None, environ=None, **kwargs): """request_started receiver: start the response timer of this request.""" _request.set({"t0": time.monotonic(), "id": ""}) def request_id(id): """Name the current request for telemetry (app.view, ajax url_name, api endpoint). Called by start_view / start_ajax / start_api; the name becomes the id of http.duration and http.exception.""" context = _request.get() if context is not None: context["id"] = id or "" def on_request_finished(sender=None, **kwargs): """request_finished receiver: record the response time, then flush the counters this request raised (view hits, cache access, log events).""" context = _request.get() if context is not None: _request.set(None) ms = (time.monotonic() - context["t0"]) * 1000 telemetry("http.duration", id=context["id"], value=ms) telemetry_flush() def on_request_exception(sender=None, request=None, **kwargs): """got_request_exception receiver: count an unhandled view exception (5xx).""" context = _request.get() telemetry_bump("http.exception", id=(context or {}).get("id", "")) def on_user_logged_in(sender, request=None, user=None, **kwargs): """Signal receiver (local login form): count a login.""" telemetry("auth.login", id="local") def on_user_login_failed(sender, credentials=None, request=None, **kwargs): """Signal receiver (local login form): count a login failure.""" telemetry("auth.login.failure", id="local") def _drop(key, reason): """Log a dropped event (transient) and return False.""" log2.debug("telemetry %s: dropped, %s", key, reason) return False def _record(key, type=None, id="", value=None, option=None): """Validate one event and write it to raw or unique storage.""" if get_configuration(appname="home", keyname="TELEMETRY_ENABLED") != "yes": return True if not isinstance(key, str) or len(key) > KEY_MAX_LENGTH or not KEY_REGEX.match(key): return _drop(key, "invalid_key") declared = TELEMETRY_KEYS.get(key) if declared: if type and type != declared["type"]: return _drop(key, "type_mismatch") type = declared["type"] else: declared = {} type = type or "counter" if type not in TYPES: return _drop(key, "invalid_type") id = "" if id is None else str(id).strip() if len(id) > ID_MAX_LENGTH or not ID_REGEX.match(id): return _drop(key, "invalid_id") sample = option.get("sample") if sample is not None: sample = float(sample) if not 0 < sample <= 1: return _drop(key, "invalid_sample") if random.random() >= sample: return True if type == "unique": return _record_unique(key, declared, value) svalue = "" if type == "set": svalue = "" if value is None else str(value).strip() if not svalue or len(svalue) > SVALUE_MAX_LENGTH: return _drop(key, "invalid_value") if "values" in declared and svalue not in declared["values"]: return _drop(key, "unknown_value") value = 1 else: value = _to_number(value) if value is None: return _drop(key, "invalid_value") if type == "counter" and sample is not None: value = value / sample if not _key_allowed(key): return _drop(key, "max_keys") _write_raw(key, type, id, value, svalue) return True def _to_number(value): """Return value as a finite float (default 1), or None if invalid.""" if value is None: return 1.0 if isinstance(value, bool): return None try: value = float(value) except (TypeError, ValueError): return None if not math.isfinite(value): return None return value def _key_allowed(key): """Check TELEMETRY_MAX_KEYS against a per-process key cache. A known key costs nothing: the cache is read from the DB only when an unknown key shows up, and at most once per KNOWN_KEYS_TTL, so a key refused over the quota cannot trigger a query per event. A new key under the quota is allowed and remembered right away, without a read; the key count per process is therefore approximate, as it already was.""" keys = _known_keys["keys"] if key in keys: return True now = time.monotonic() if not _known_keys["ts"] or now - _known_keys["ts"] > KNOWN_KEYS_TTL: keys = set(TELEMETRY_KEYS) keys.update(DataTelemetryAggregate.objects.values_list("key", flat=True).distinct()) keys.update(DataTelemetryRaw.objects.values_list("key", flat=True).distinct()) keys.update(DataTelemetryUnique.objects.values_list("key", flat=True).distinct()) _known_keys["keys"] = keys _known_keys["ts"] = now if key in keys: return True max_keys = int(get_configuration(appname="home", keyname="TELEMETRY_MAX_KEYS")) if len(keys) >= max_keys: return False keys.add(key) return True def _write_raw(key, type, id, value, svalue): """Insert one raw event in its own savepoint.""" with transaction.atomic(): DataTelemetryRaw.objects.create( key=key, distinct_id=id, type=type, value=value, svalue=svalue ) # ---------------------------------------------------------------------------------- # Buffered counters (keys hit many times per request) # ---------------------------------------------------------------------------------- BUFFER_MAX_ENTRIES = 500 def telemetry_bump(key, id="", n=1): """Add n to a process-local counter instead of writing an event: for keys hit many times per request (cache access, log events, view hits). No DB write, no configuration read, never raises. telemetry_flush() turns the counters into one raw event per (key, id), so counts land in the bucket of the flush, not of the event.""" try: entry = (key, "" if id is None else str(id)) if entry in _buffer: _buffer[entry] += n elif len(_buffer) < BUFFER_MAX_ENTRIES: _buffer[entry] = n except Exception: pass def telemetry_flush(): """Write the buffered counters as raw counter events; return the rows written. Called at every HTTP entry and exit, at the end of a celery task and by each aggregation pass. Never raises: a failed flush drops its counts.""" if not _buffer: return 0 counts = list(_buffer.items()) _buffer.clear() try: if get_configuration(appname="home", keyname="TELEMETRY_ENABLED") != "yes": return 0 objs = [] for (key, did), n in counts: if len(key) > KEY_MAX_LENGTH or not KEY_REGEX.match(key): continue if len(did) > ID_MAX_LENGTH or not ID_REGEX.match(did): continue if not _key_allowed(key): continue objs.append(DataTelemetryRaw(key=key, distinct_id=did, type="counter", value=n)) with transaction.atomic(): DataTelemetryRaw.objects.bulk_create(objs) return len(objs) except Exception as e: log2.debug("telemetry flush: error %s", e) return 0 # ---------------------------------------------------------------------------------- # Unique (high-cardinality distinct count) # ---------------------------------------------------------------------------------- def _unique_salt(): """Return the unique-member salt from the registry, created on first use. Not SECRET_KEY: it may be random per process (unset in dev) or rotated, which would count one member several times.""" if _salt["value"] is None: try: with transaction.atomic(): entry, _ = DataRegistry.objects.get_or_create( key=SALT_REGISTRY_KEY, defaults={"value": secrets.token_hex(32)} ) except IntegrityError: entry = DataRegistry.objects.get(key=SALT_REGISTRY_KEY) _salt["value"] = entry.value return _salt["value"] def unique_member(value): """Return a short salted hash of a member value; the value itself is never stored.""" digest = hmac.new(_unique_salt().encode(), str(value).encode(), hashlib.sha256) return digest.hexdigest()[:16] def _record_unique(key, declared, value): """Add a member to each open period bucket of a unique key.""" value = "" if value is None else str(value).strip() if not value: return _drop(key, "invalid_value") periods = [p for p in declared.get("periods", UNIQUE_PERIODS) if p in UNIQUE_PERIODS] if not _key_allowed(key): return _drop(key, "max_keys") member = unique_member(value) now = timezone.now() rows = [] for period in periods: start = bucket_start(now, period) if not _unique_memo_seen(key, period, start, member): rows.append(DataTelemetryUnique(key=key, period=period, start=start, member=member)) if rows: _write_unique(rows) for row in rows: _unique_memo_add(key, row.period, row.start, member) return True def _unique_memo_seen(key, period, start, member): """Return True if member is already recorded for this bucket by this process.""" entry = _unique_memo.get((key, period)) return entry is not None and entry[0] == start and member in entry[1] def _unique_memo_add(key, period, start, member): """Remember a recorded member; a new bucket or a full memo starts a fresh set.""" entry = _unique_memo.get((key, period)) if entry is None or entry[0] != start or len(entry[1]) >= UNIQUE_MEMO_MAX: entry = (start, set()) _unique_memo[(key, period)] = entry entry[1].add(member) def _write_unique(rows): """Insert unique members in their own savepoint; duplicates are ignored.""" with transaction.atomic(): DataTelemetryUnique.objects.bulk_create(rows, ignore_conflicts=True) # ---------------------------------------------------------------------------------- # Aggregation: raw -> minute/hour ; rollups day..year ; unique close ; raw purge # ---------------------------------------------------------------------------------- GRACE_SECONDS = 60 RAW_PERIODS = ["minute", "hour"] ROLLUP_SOURCE = {"day": "hour", "week": "day", "month": "day", "quarter": "month", "year": "month"} MAX_BUCKETS_PER_RUN = 240 RAW_CHUNK_BUCKETS = 60 RAW_HARD_KEEP_HOURS = 24 PERCENTILES = {"p90": 90, "p95": 95, "p99": 99} AGGREGATE_LOCK = "telemetry_aggregate_lock" AGGREGATE_LOCK_TIMEOUT = 300 def bucket_next(start, period): """Return the start of the bucket following the one that starts at start.""" if period == "minute": return bucket_start(start.astimezone(UTC) + timedelta(minutes=1), period) if period == "hour": return bucket_start(start.astimezone(UTC) + timedelta(hours=1), period) day = timezone.localtime(start).date() if period == "day": day = day + timedelta(days=1) elif period == "week": day = day + timedelta(days=7) else: month = day.month - 1 + {"month": 1, "quarter": 3, "year": 12}[period] day = day.replace(year=day.year + month // 12, month=month % 12 + 1, day=1) return timezone.make_aware(datetime.combine(day, datetime.min.time())).astimezone(UTC) def _wm_get(period): """Return the period watermark: start of the first bucket not yet finalized.""" value = registry_get_key(f"TELEMETRY_WM_{period.upper()}") if not value: return None try: return datetime.fromisoformat(value) except ValueError: return None def _wm_set(period, start): """Store the period watermark.""" registry_set_key(f"TELEMETRY_WM_{period.upper()}", start.isoformat()) def _bucket_range(lo, limit, period, max_buckets=MAX_BUCKETS_PER_RUN): """Return hi, the end of the run of closed buckets from lo, with hi <= limit.""" hi = lo for _ in range(max_buckets): nxt = bucket_next(hi, period) if nxt > limit: break hi = nxt return hi def telemetry_aggregate(now=None): """Run one aggregation pass; return counts per step, or None if a pass is running.""" if not django_cache.add(AGGREGATE_LOCK, 1, AGGREGATE_LOCK_TIMEOUT): return None try: now = now or timezone.now() counts = {"flush": telemetry_flush()} for period in RAW_PERIODS: counts[period] = _aggregate_raw(period, now) for period in ROLLUP_SOURCE: counts[period] = _aggregate_rollup(period, now) counts["unique"] = _close_unique(now) counts["raw_purge"] = _purge_raw(now) finally: django_cache.delete(AGGREGATE_LOCK) log2.debug("telemetry aggregate %s", counts) return counts def _aggregate_raw(period, now): """Finalize closed minute/hour buckets from raw events; return the rows written. A catch-up run is split in windows of RAW_CHUNK_BUCKETS buckets, each finalized and watermarked on its own, so the events held in memory stay bounded.""" closed = bucket_start(now - timedelta(seconds=GRACE_SECONDS), period) lo = _wm_get(period) if lo is None: first = DataTelemetryRaw.objects.order_by("timestamp").values_list("timestamp", flat=True) first = first.first() lo = bucket_start(first, period) if first else closed _wm_set(period, lo) hi = _bucket_range(lo, closed, period) rows = 0 while lo < hi: chunk = _bucket_range(lo, hi, period, max_buckets=RAW_CHUNK_BUCKETS) rows += _aggregate_raw_window(period, lo, chunk) lo = chunk return rows def _aggregate_raw_window(period, lo, hi): """Finalize the buckets of one [lo, hi) window from raw events; return the rows written. The distinct_id="" row of a key covers every event of that key, whatever its id: a sum over all ids for a counter, but a mix of dimensions for a gauge (last = the id written last), which is why a gauge total belongs in a key of its own.""" raws = DataTelemetryRaw.objects.filter(timestamp__gte=lo, timestamp__lt=hi) raws = raws.order_by("timestamp", "id") raws = raws.values_list("timestamp", "key", "distinct_id", "type", "value", "svalue") types = {} groups = defaultdict(list) mixed = 0 for ts, key, did, typ, value, svalue in raws.iterator(): if types.setdefault(key, typ) != typ: mixed += 1 continue groups[(bucket_start(ts, period), key)].append((did, value, svalue)) if mixed: log2.debug("telemetry aggregate %s: %s event(s) dropped, type changed", period, mixed) objs = [] for (start, key), events in groups.items(): typ = types[key] objs.append(_agg_from_events(key, "", typ, period, start, events)) by_id = defaultdict(list) for event in events: if event[0]: by_id[event[0]].append(event) for did, evs in _fold_ids(key, by_id, len).items(): objs.append(_agg_from_events(key, did, typ, period, start, evs)) _replace_aggregates(period, lo, hi, objs) return len(objs) def _aggregate_rollup(period, now): """Finalize closed buckets of period from its source period; return the rows written.""" source = ROLLUP_SOURCE[period] src_wm = _wm_get(source) if src_wm is None: return 0 lo = _wm_get(period) if lo is None: first = DataTelemetryAggregate.objects.filter(period=source).exclude(type="unique") first = first.order_by("start").values_list("start", flat=True).first() lo = bucket_start(first or src_wm, period) _wm_set(period, lo) hi = _bucket_range(lo, src_wm, period) if hi <= lo: return 0 children = DataTelemetryAggregate.objects.filter(period=source, start__gte=lo, start__lt=hi) children = children.exclude(type="unique").order_by("start") groups = defaultdict(lambda: defaultdict(list)) for child in children.iterator(): groups[(bucket_start(child.start, period), child.key)][child.distinct_id].append(child) objs = [] for (start, key), by_id in groups.items(): total = by_id.pop("", []) if total: objs.append(_agg_merge(key, "", period, start, total)) for did, kids in _fold_ids(key, by_id, _count_weight).items(): objs.append(_agg_merge(key, did, period, start, kids)) _replace_aggregates(period, lo, hi, objs) return len(objs) def _count_weight(kids): """Fold weight of a list of child aggregates: their total count.""" return sum(k.count for k in kids) def _fold_ids(key, by_id, weight): """Keep the max_ids heaviest ids of a key; fold the others into _other.""" max_ids = TELEMETRY_KEYS.get(key, {}).get("max_ids", DEFAULT_MAX_IDS) if len(by_id) <= max_ids: return by_id ranked = sorted(by_id, key=lambda did: weight(by_id[did]), reverse=True) folded = {did: by_id[did] for did in ranked[:max_ids]} other = folded.setdefault("_other", []) for did in ranked[max_ids:]: other.extend(by_id[did]) return folded def _replace_aggregates(period, lo, hi, objs): """Atomically replace the [lo, hi) aggregates of period and move its watermark to hi.""" with transaction.atomic(): old = DataTelemetryAggregate.objects.filter(period=period, start__gte=lo, start__lt=hi) old.exclude(type="unique").delete() DataTelemetryAggregate.objects.bulk_create(objs, batch_size=500) _wm_set(period, hi) def _percentile(ordered, p): """Nearest-rank percentile of a sorted, non-empty list.""" return ordered[max(0, math.ceil(p / 100 * len(ordered)) - 1)] def _histogram(key, values): """Count values per bucket: le_ is (previous b, b], inf is above the last b.""" bounds = TELEMETRY_KEYS.get(key, {}).get("buckets", DEFAULT_BUCKETS) data = {f"le_{b:g}": 0 for b in bounds} data["inf"] = 0 for value in values: label = next((f"le_{b:g}" for b in bounds if value <= b), "inf") data[label] += 1 return data def _agg_from_events(key, did, typ, period, start, events): """Build an aggregate row from raw events [(id, value, svalue)].""" agg = DataTelemetryAggregate( key=key, distinct_id=did, type=typ, period=period, start=start, count=len(events) ) if typ == "set": agg.data = dict(Counter(event[2] for event in events)) agg.distinct = len(agg.data) return agg values = [event[1] for event in events] agg.sum = sum(values) agg.min = min(values) agg.max = max(values) agg.mean = agg.sum / agg.count agg.last = values[-1] if typ == "timer": ordered = sorted(values) agg.median = statistics.median(ordered) for name, p in PERCENTILES.items(): setattr(agg, name, _percentile(ordered, p)) if typ == "histogram": agg.data = _histogram(key, values) return agg def _agg_merge(key, did, period, start, kids): """Build an aggregate row from child aggregates; percentiles are count-weighted means.""" kids = sorted(kids, key=lambda k: k.start) typ = kids[0].type agg = DataTelemetryAggregate( key=key, distinct_id=did, type=typ, period=period, start=start, count=sum(k.count for k in kids), ) # fmt: skip if typ in ("set", "histogram"): data = Counter() for kid in kids: data.update(kid.data or {}) agg.data = dict(data) if typ == "set": agg.distinct = len(agg.data) return agg numeric = [k for k in kids if k.count and k.min is not None] if not numeric: return agg ncount = sum(k.count for k in numeric) agg.sum = sum(k.sum or 0 for k in numeric) agg.min = min(k.min for k in numeric) agg.max = max(k.max for k in numeric) agg.mean = agg.sum / ncount if ncount else None agg.last = numeric[-1].last if typ == "timer": for name in ["median", *PERCENTILES]: pairs = [(k.count, getattr(k, name)) for k in numeric if getattr(k, name) is not None] if pairs: setattr(agg, name, sum(c * v for c, v in pairs) / sum(c for c, _ in pairs)) agg.approx = len(numeric) > 1 or any(k.approx for k in numeric) return agg def _close_unique(now): """Write the distinct count of closed unique buckets as aggregates; drop their members.""" rows = 0 for period in UNIQUE_PERIODS: closed = bucket_start(now - timedelta(seconds=GRACE_SECONDS), period) buckets = DataTelemetryUnique.objects.filter(period=period, start__lt=closed) buckets = buckets.values("key", "start").annotate(n=Count("id")) for bucket in buckets: with transaction.atomic(): DataTelemetryAggregate.objects.update_or_create( key=bucket["key"], distinct_id="", period=period, start=bucket["start"], defaults={"type": "unique", "count": bucket["n"], "distinct": bucket["n"]}, ) # fmt: skip DataTelemetryUnique.objects.filter( key=bucket["key"], period=period, start=bucket["start"] ).delete() rows += 1 return rows def _purge_raw(now): """Delete raw events older than TELEMETRY_RAW_KEEP_MINUTES, never before every raw period has aggregated them (lowest watermark: minute may lag behind hour on a catch-up run), always beyond RAW_HARD_KEEP_HOURS; return the count.""" keep = int(get_configuration(appname="home", keyname="TELEMETRY_RAW_KEEP_MINUTES")) hard = now - timedelta(hours=RAW_HARD_KEEP_HOURS) watermarks = [_wm_get(period) for period in RAW_PERIODS] if any(wm is None for wm in watermarks): cutoff = hard else: cutoff = min(now - timedelta(minutes=keep), *watermarks) cutoff = max(cutoff, hard) return DataTelemetryRaw.objects.filter(timestamp__lt=cutoff).delete()[0] # ---------------------------------------------------------------------------------- # Object quantities (gauges sampled by a periodic task) # ---------------------------------------------------------------------------------- SAMPLED_TABLES = { "instance": DataInstance, "eav": DataEAV, "revision": DataRevision, "file": DataFile, "task": DataTask, "log_event": CavalibaEvent, "telemetry_raw": DataTelemetryRaw, "telemetry_agg": DataTelemetryAggregate, } def telemetry_sample(): """Record the object quantity gauges: instances per schema (data.objects, id=schema), their total (data.objects.total) and the size of the main tables (data.rows, id=table). Called by the sampler task; return the number of events recorded. Checks TELEMETRY_ENABLED up front: the counts below are the expensive part, no point running them for events that would be dropped.""" if get_configuration(appname="home", keyname="TELEMETRY_ENABLED") != "yes": return 0 count = 0 total = 0 rows = DataInstance.objects.values("classname").annotate(n=Count("id")) for row in rows: total += row["n"] count += telemetry("data.objects", id=row["classname"], value=row["n"]) count += telemetry("data.objects.total", value=total) for name, model in SAMPLED_TABLES.items(): count += telemetry("data.rows", id=name, value=model.objects.count()) log2.debug("telemetry sample: %s gauge(s)", count) return count # ---------------------------------------------------------------------------------- # Retention # ---------------------------------------------------------------------------------- def telemetry_purge(keep_days=None, now=None): """Apply TELEMETRY_KEEP_DAYS_ (0 = forever) to aggregates; keep_days {period: days} overrides the configuration. Return the count removed.""" now = now or timezone.now() keep_days = keep_days or {} count = 0 for period in PERIODS: days = keep_days.get(period) if days is None: keyname = f"TELEMETRY_KEEP_DAYS_{period.upper()}" days = int(get_configuration(appname="home", keyname=keyname)) if days > 0: old = DataTelemetryAggregate.objects.filter( period=period, start__lt=now - timedelta(days=days) ) count += old.delete()[0] log2.ok("telemetry.purge", count=count) return count # ---------------------------------------------------------------------------------- # Query (shared by UI and API) # ---------------------------------------------------------------------------------- MAX_POINTS = 1000 MAX_IDS_LISTED = 100 RAW_DEFAULT_LIMIT = 100 RAW_MAX_LIMIT = 1000 RELATIVE_REGEX = re.compile(r"^now(?:([+-])(\d+)([smhdw]))?$") RELATIVE_UNITS = {"s": "seconds", "m": "minutes", "h": "hours", "d": "days", "w": "weeks"} class TelemetryError(ValueError): """Invalid telemetry query (bad period, function, time, range).""" class TelemetryNotFound(TelemetryError): """Unknown telemetry key.""" def parse_time(text=None, now=None): """Parse now, now-15m/2h/1d/1w, 2026-10-01, 2026/10/01, 2026-10-01T10:00 into an aware datetime (naive values are local time). Relative offsets are real elapsed time.""" now = (now or timezone.now()).astimezone(UTC) if isinstance(text, datetime): dt = text else: text = (text or "now").strip() match = RELATIVE_REGEX.match(text.lower()) if match: sign, amount, unit = match.groups() if not sign: return now delta = timedelta(**{RELATIVE_UNITS[unit]: int(amount)}) return now + delta if sign == "+" else now - delta try: dt = datetime.fromisoformat(text.replace("/", "-")) except ValueError: raise TelemetryError(f"invalid time: {text}") from None if timezone.is_naive(dt): dt = timezone.make_aware(dt) return dt def key_type(key): """Return the type of a declared or seen key, or None if unknown.""" if key in TELEMETRY_KEYS: return TELEMETRY_KEYS[key]["type"] for model in (DataTelemetryAggregate, DataTelemetryRaw): typ = model.objects.filter(key=key).values_list("type", flat=True).first() if typ: return typ if DataTelemetryUnique.objects.filter(key=key).exists(): return "unique" return None def key_periods(key, typ): """Return the periods available for a key.""" if typ == "unique": periods = TELEMETRY_KEYS.get(key, {}).get("periods", UNIQUE_PERIODS) return [p for p in UNIQUE_PERIODS if p in periods] return list(PERIODS) def get_keys(): """Return declared and seen keys: [{key, type, description, declared}] sorted by key.""" seen = {} for model in (DataTelemetryAggregate, DataTelemetryRaw): for key, typ in model.objects.values_list("key", "type").distinct(): seen.setdefault(key, typ) for key in DataTelemetryUnique.objects.values_list("key", flat=True).distinct(): seen.setdefault(key, "unique") reply = [] for key in sorted(set(TELEMETRY_KEYS) | set(seen)): declared = TELEMETRY_KEYS.get(key) reply.append({ "key": key, "type": declared["type"] if declared else seen[key], "description": declared.get("description", "") if declared else "", "declared": bool(declared), }) # fmt: skip return reply def get_key_detail(key): """Return type, functions, periods with available range, and ids of a key.""" typ = key_type(key) if not typ: raise TelemetryNotFound(f"unknown key: {key}") declared = TELEMETRY_KEYS.get(key, {}) ranges = {} rows = DataTelemetryAggregate.objects.filter(key=key, distinct_id="").values("period") for row in rows.annotate(first=Min("start"), last=Max("start"), count=Count("id")): ranges[row["period"]] = {"first": row["first"], "last": row["last"], "count": row["count"]} ids = DataTelemetryAggregate.objects.filter(key=key).exclude(distinct_id="") ids = ids.values_list("distinct_id", flat=True).distinct().order_by("distinct_id") return { "key": key, "type": typ, "description": declared.get("description", ""), "unit": declared.get("unit", ""), "declared": bool(declared), "functions": TYPE_FUNCTIONS[typ], "default_function": TYPE_FUNCTIONS[typ][0], "periods": key_periods(key, typ), "ranges": ranges, "ids": list(ids[:MAX_IDS_LISTED]), } def _check_query(key, function, period): """Validate key/function/period; return (type, function).""" typ = key_type(key) if not typ: raise TelemetryNotFound(f"unknown key: {key}") function = function or TYPE_FUNCTIONS[typ][0] if function not in TYPE_FUNCTIONS[typ]: raise TelemetryError(f"invalid function for {typ}: {function}") if period not in key_periods(key, typ): raise TelemetryError(f"invalid period for {key}: {period}") return typ, function def _ordered_data(key, data): """Return breakdown/buckets data with histogram buckets in bound order.""" data = data or {} def bound(label): """Numeric sort key of a bucket label; inf and non-bucket labels last.""" try: return (0, float(label[3:])) if label.startswith("le_") else (1, 0.0) except ValueError: return (1, 0.0) if all(label.startswith("le_") or label == "inf" for label in data): return {label: data[label] for label in sorted(data, key=bound)} return dict(sorted(data.items(), key=lambda item: -item[1])) def _row_value(key, typ, function, row): """Extract a function value from an aggregate row; None row gives the empty value.""" if row is None: if function in ("breakdown", "buckets"): return {} if function in ("count", "distinct") or typ == "counter": return 0 return None if function in ("breakdown", "buckets"): return _ordered_data(key, row.data) return getattr(row, function) def _unique_live(key, period, first, last): """Return {start: member count} of not-yet-closed unique buckets in [first, last].""" rows = DataTelemetryUnique.objects.filter( key=key, period=period, start__gte=first, start__lte=last ) return {r["start"]: r["n"] for r in rows.values("start").annotate(n=Count("id"))} def get_value(key, function=None, period="minute", at="now", id=""): """Return {start, value} of the bucket containing at.""" typ, function = _check_query(key, function, period) start = bucket_start(parse_time(at), period) row = DataTelemetryAggregate.objects.filter( key=key, distinct_id=id, period=period, start=start ).first() if row is None and typ == "unique" and not id: live = _unique_live(key, period, start, start) if live: return {"start": start, "value": live[start]} return {"start": start, "value": _row_value(key, typ, function, row)} def get_series(key, function=None, period="hour", start="now-1d", end="now", id=""): """Return [{start, value}] for every bucket from start to end (inclusive).""" typ, function = _check_query(key, function, period) first = bucket_start(parse_time(start), period) last = bucket_start(parse_time(end), period) if last < first: raise TelemetryError("end before start") starts = [first] while starts[-1] < last: if len(starts) >= MAX_POINTS: raise TelemetryError(f"more than {MAX_POINTS} points, use a larger period") starts.append(bucket_next(starts[-1], period)) rows = DataTelemetryAggregate.objects.filter( key=key, distinct_id=id, period=period, start__gte=first, start__lte=last ) by_start = {row.start: row for row in rows} live = {} if typ == "unique" and not id: live = _unique_live(key, period, first, last) reply = [] for bucket in starts: row = by_start.get(bucket) if row is None and bucket in live: value = live[bucket] else: value = _row_value(key, typ, function, row) reply.append({"start": bucket, "value": value}) return reply def get_id_breakdown(key, function=None, period="hour", start="now-1d", end="now"): """Return [{id, value}] per distinct_id of a key over a range, heaviest first: the top views, the busiest schemas, cache hit vs miss. Buckets are combined per id the way the function allows: added (sum, count), min, max, count-weighted (mean, median, p9x) or taken from the latest bucket (last).""" typ, function = _check_query(key, function, period) if function in ("breakdown", "buckets", "distinct"): raise TelemetryError(f"no id breakdown for {function}") first = bucket_start(parse_time(start), period) last = bucket_start(parse_time(end), period) if last < first: raise TelemetryError("end before start") rows = DataTelemetryAggregate.objects.filter( key=key, period=period, start__gte=first, start__lte=last ).exclude(distinct_id="") by_id = defaultdict(list) for row in rows.order_by("start"): value = getattr(row, function) if value is not None: by_id[row.distinct_id].append((row.count, value)) reply = [] for did, values in by_id.items(): if function in ("sum", "count"): value = sum(v for _, v in values) elif function == "min": value = min(v for _, v in values) elif function == "max": value = max(v for _, v in values) elif function == "last": value = values[-1][1] else: value = sum(c * v for c, v in values) / (sum(c for c, _ in values) or 1) reply.append({"id": did, "value": value}) reply.sort(key=lambda item: -item["value"]) return reply[:MAX_IDS_LISTED] def get_raw(limit=RAW_DEFAULT_LIMIT, key=None): """Return the most recent raw events, newest first.""" limit = max(1, min(int(limit), RAW_MAX_LIMIT)) qs = DataTelemetryRaw.objects.order_by("-timestamp", "-id") if key: qs = qs.filter(key=key) fields = ("timestamp", "key", "distinct_id", "type", "value", "svalue") return list(qs.values(*fields)[:limit]) # ---------------------------------------------------------------------------------- # Widgets and dashboards (hardcoded for now) # ---------------------------------------------------------------------------------- # A widget spec: w (value|table|barchart|idtable), key, function, period, id, label and # either at (value) or start/end (table, barchart, idtable). Open minute/hour/day buckets # are not aggregated yet, so value widgets point at the previous bucket (unique keys are # live). An idtable shows one row per distinct_id of the key over the range. WIDGET_TYPES = ["value", "table", "barchart", "idtable"] WIDGETS = { "logins_last_hour": { "w": "value", "label": "Logins (last hour)", "key": "auth.login", "period": "hour", "at": "now-1h", }, "login_failures_last_hour": { "w": "value", "label": "Login failures (last hour)", "key": "auth.login.failure", "period": "hour", "at": "now-1h", }, "logins_hourly": { "w": "barchart", "label": "Logins per hour", "key": "auth.login", "period": "hour", "start": "now-1d", "end": "now", }, "login_failures_hourly": { "w": "barchart", "label": "Login failures per hour", "key": "auth.login.failure", "period": "hour", "start": "now-1d", "end": "now", }, "login_failures_daily": { "w": "table", "label": "Login failures per day", "key": "auth.login.failure", "period": "day", "start": "now-7d", "end": "now", }, "users_today": {"w": "value", "label": "Active users today", "key": "users.active", "period": "day"}, "users_week": {"w": "value", "label": "Active users this week", "key": "users.active", "period": "week"}, "users_month": {"w": "value", "label": "Active users this month", "key": "users.active", "period": "month"}, "users_daily": { "w": "barchart", "label": "Active users per day", "key": "users.active", "period": "day", "start": "now-30d", "end": "now", }, "api_requests_last_hour": { "w": "value", "label": "API requests (last hour)", "key": "api.request", "period": "hour", "at": "now-1h", }, "api_requests_hourly": { "w": "barchart", "label": "API requests per hour", "key": "api.request", "period": "hour", "start": "now-1d", "end": "now", }, "task_duration_p95_hourly": { "w": "barchart", "label": "Task duration p95 per hour (ms)", "key": "task.duration", "function": "p95", "period": "hour", "start": "now-1d", "end": "now", }, "task_runs_daily": { "w": "table", "label": "Task runs per day", "key": "task.duration", "function": "count", "period": "day", "start": "now-7d", "end": "now", }, # performance "response_time_p95_hourly": { "w": "barchart", "label": "Response time p95 per hour (ms)", "key": "http.duration", "function": "p95", "period": "hour", "start": "now-1d", "end": "now", }, "response_time_by_view": { "w": "idtable", "label": "Response time p95 per view (ms)", "key": "http.duration", "function": "p95", "period": "hour", "start": "now-1d", "end": "now", }, "requests_hourly": { "w": "barchart", "label": "View hits per hour", "key": "view.request", "period": "hour", "start": "now-1d", "end": "now", }, "top_views": { "w": "idtable", "label": "Top views (1 day)", "key": "view.request", "period": "hour", "start": "now-1d", "end": "now", }, "cache_get": { "w": "idtable", "label": "Cache hit / miss (1 day)", "key": "cache.get", "period": "hour", "start": "now-1d", "end": "now", }, "cache_session": { "w": "idtable", "label": "Session cache hit / miss (1 day)", "key": "cache.session", "period": "hour", "start": "now-1d", "end": "now", }, # log "log_events_hourly": { "w": "barchart", "label": "Log events per hour", "key": "log.event", "period": "hour", "start": "now-1d", "end": "now", }, "log_events_by_level": { "w": "idtable", "label": "Log events per level (7 days)", "key": "log.event", "period": "day", "start": "now-7d", "end": "now", }, "http_exceptions_daily": { "w": "table", "label": "View exceptions per day", "key": "http.exception", "period": "day", "start": "now-7d", "end": "now", }, # data "revisions_hourly": { "w": "barchart", "label": "Writes per hour", "key": "data.write", "period": "hour", "start": "now-1d", "end": "now", }, "revisions_daily": { "w": "table", "label": "Writes per day", "key": "data.write", "period": "day", "start": "now-7d", "end": "now", }, "writes_by_schema": { "w": "idtable", "label": "Writes per schema (7 days)", "key": "data.write", "period": "day", "start": "now-7d", "end": "now", }, "writes_by_action": { "w": "table", "label": "Writes per action (7 days)", "key": "data.write.action", "function": "breakdown", "period": "day", "start": "now-7d", "end": "now", }, "objects_total": { "w": "value", "label": "Objects", "key": "data.objects.total", "period": "hour", "at": "now-1h", }, "objects_by_schema": { "w": "idtable", "label": "Objects per schema", "key": "data.objects", "function": "last", "period": "hour", "start": "now-1d", "end": "now", }, "objects_daily": { "w": "barchart", "label": "Objects per day", "key": "data.objects.total", "period": "day", "start": "now-30d", "end": "now", }, "table_rows": { "w": "idtable", "label": "Table sizes", "key": "data.rows", "function": "last", "period": "hour", "start": "now-1d", "end": "now", }, } # fmt: skip DASHBOARDS = { "security": { "label": "Security", "refresh": 60, "widgets": [ "logins_last_hour", "login_failures_last_hour", "users_today", "logins_hourly", "login_failures_hourly", "login_failures_daily", ], }, "users": { "label": "Users", "refresh": 60, "widgets": ["users_today", "users_week", "users_month", "users_daily"], }, "operations": { "label": "Operations", "refresh": 60, "widgets": [ "api_requests_last_hour", "api_requests_hourly", "task_duration_p95_hourly", "task_runs_daily", "log_events_hourly", "log_events_by_level", "http_exceptions_daily", ], }, "performance": { "label": "Performance", "refresh": 60, "widgets": [ "response_time_p95_hourly", "response_time_by_view", "requests_hourly", "top_views", "cache_get", "cache_session", ], }, "data": { "label": "Data", "refresh": 60, "widgets": [ "objects_total", "objects_by_schema", "table_rows", "revisions_hourly", "revisions_daily", "writes_by_schema", "writes_by_action", "objects_daily", ], }, } # fmt: skip WIDGET_PARAMS = ["w", "key", "function", "period", "id", "at", "start", "end", "label"] def format_start(dt, period): """Short human label of a bucket start for its period.""" dt = timezone.localtime(dt) if period == "minute": return dt.strftime("%m-%d %H:%M") if period == "hour": return dt.strftime("%m-%d %H:00") if period == "day": return dt.strftime("%Y-%m-%d") if period == "week": return dt.strftime("%G-W%V") if period == "month": return dt.strftime("%Y-%m") if period == "quarter": return f"{dt.year}-Q{(dt.month - 1) // 3 + 1}" return str(dt.year) def format_number(value): """Display a numeric value: integers without decimals, others with 2.""" if value is None: return "-" if isinstance(value, float) and value.is_integer(): return str(int(value)) if isinstance(value, float): return f"{value:.2f}" return str(value) def widget_spec(params): """Build a widget spec from a WIDGETS name (params["name"]) or from explicit params.""" name = (params.get("name") or "").strip() if name: if name not in WIDGETS: raise TelemetryNotFound(f"unknown widget: {name}") return dict(WIDGETS[name]) return {k: params.get(k, "").strip() for k in WIDGET_PARAMS if params.get(k, "").strip()} def widget_data(spec): """Resolve a widget spec into render data: value, or rows/labels/values over a range. Breakdown/buckets over a range are summed into one row per category.""" w = spec.get("w") or "value" if w not in WIDGET_TYPES: raise TelemetryError(f"invalid widget: {w}") key = spec.get("key") or "" typ = key_type(key) if not typ: raise TelemetryNotFound(f"unknown key: {key}") function = spec.get("function") or TYPE_FUNCTIONS[typ][0] period = spec.get("period") or ("hour" if "hour" in key_periods(key, typ) else "day") did = spec.get("id", "") label = spec.get("label") or " ยท ".join(x for x in (key, did, function, period) if x) data = { "w": w, "key": key, "type": typ, "function": function, "period": period, "id": did, "label": label, "unit": TELEMETRY_KEYS.get(key, {}).get("unit", ""), } # fmt: skip if w == "value": result = get_value(key, function, period, spec.get("at") or "now", did) data["start"] = format_start(result["start"], period) if isinstance(result["value"], dict): data["rows"] = [{"label": k, "value": v} for k, v in result["value"].items()] data["value"] = format_number(sum(result["value"].values())) else: data["value"] = format_number(result["value"]) return data start = spec.get("start") or spec.get("at") or "now-1d" end = spec.get("end") or spec.get("at") or "now" if w == "idtable": breakdown = get_id_breakdown(key, function, period, start, end) rows = [{"label": row["id"], "value": row["value"]} for row in breakdown] data["range"] = f"{format_start(bucket_start(parse_time(start), period), period)} - " data["range"] += format_start(bucket_start(parse_time(end), period), period) for row in rows: row["display"] = format_number(row["value"]) data["rows"] = rows data["table_rows"] = rows return data series = get_series(key, function, period, start, end, did) if function in ("breakdown", "buckets"): merged = Counter() for point in series: merged.update(point["value"]) rows = [{"label": k, "value": v} for k, v in _ordered_data(key, dict(merged)).items()] data["range"] = f"{format_start(series[0]['start'], period)} - " data["range"] += format_start(series[-1]["start"], period) else: rows = [{"label": format_start(p["start"], period), "value": p["value"]} for p in series] for row in rows: row["display"] = format_number(row["value"]) data["rows"] = rows data["table_rows"] = rows if "range" in data else rows[::-1] data["chart"] = { "label": label, "labels": [row["label"] for row in rows], "values": [row["value"] for row in rows], } return data