inference-bot
70ae435867
The co-op had no longitudinal data: every number was an ad-hoc query, so
nothing compounded and no single source owned "how many members do we have,
and when did that change". That cost us a wrong figure in the newsletter ("six
new members" — actually 21; the six were forum joins).
New: metrics_snapshot.py writes one row per day into the portal's SQLite store
(the same members.db), and GET /admin/metrics/series returns them oldest-first
for the panel and dashboard to chart later.
- Cumulative columns (members_active, members_total, members_new,
newsletter_subs, matrix_space_members) are counts AS OF that day;
spend_usd/requests are THAT day's traffic.
- FAIL CLOSED PER METRIC: an unreadable source yields NULL, never a zero — a
zero reads as "nobody" and would bake a lie into a chart. Failures print to
stdout, and a scheduler task's stdout lands in the app log.
- Listmonk's cached subscriber_count is NOT trusted (it reads 142 for a list
holding 71): the count comes from paging the subscribers endpoint.
- Member growth comes from welcome_sent_at, the provisioning timestamp —
the closest thing the portal stores to a join date. Manual/founder members
have no welcome email, so backfilled days are a documented LOWER BOUND;
today's row carries the exact count.
- Runs as an in-app scheduler task (addons.scheduler, 03:15 UTC) rather than
an agent cron: member-facing plumbing should not depend on the agent VM.
- Backfill is idempotent; member growth is reconstructible back to
September and spend/requests back to the gateway's first request.
Charts land later, on the surfaces, once the shape of the series is reviewed.
285 lines
11 KiB
Python
285 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Daily metrics snapshot for the Inference Cooperative portal.
|
|
|
|
Writes one row per day into the portal's SQLite store (`metrics_daily`, the same
|
|
`members.db` that holds the members table) so the admin panel and the member
|
|
dashboard can chart trends from one source of truth.
|
|
|
|
metrics_snapshot.py record / refresh TODAY's row
|
|
metrics_snapshot.py --backfill rebuild history (idempotent)
|
|
|
|
Rules this file follows deliberately:
|
|
|
|
- **Fail closed per metric.** A source that cannot be read yields NULL for its
|
|
columns, never a zero. A zero reads as "nobody", which is a lie that would
|
|
survive into a chart; NULL reads as "not measured". Each failure is printed,
|
|
and the task's stdout lands in the app log (`cloudron logs --app <id>`).
|
|
- **Cumulative vs daily.** `members_active`, `members_total`, `newsletter_subs`
|
|
and `matrix_space_members` are counts AS OF that day; `spend_usd` and
|
|
`requests` are THAT day's traffic. Charts must treat them accordingly.
|
|
- **Runs in Cloudron's scheduler container**, which gets the platform env
|
|
(declared app env + addon vars) but NOT variables the app's own start.sh
|
|
sourced. Everything needed here is a declared app env var: LITELLM_DB_*,
|
|
LISTMONK_*, MATRIX_BOT_TOKEN, LISTMONK_LIST_ID.
|
|
- `members_new` uses `welcome_sent_at`, the provisioning timestamp — the closest
|
|
thing the portal stores to a join date. Manual members have no welcome email,
|
|
so their joins are absent from the growth curve (noted, not guessed).
|
|
"""
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
DB_PATH = os.environ.get("DB_PATH", "/app/data/members.db")
|
|
LISTMONK_BASE = os.environ.get("LISTMONK_BASE", "https://newsletter.inference.coop")
|
|
LISTMONK_LIST_ID = int(os.environ.get("LISTMONK_LIST_ID") or 3)
|
|
MATRIX_BASE = os.environ.get("MATRIX_BASE", "https://matrix.inference.coop")
|
|
SPACE = os.environ.get("MATRIX_SPACE_ID", "!FKOD2sSqeYvEq9jzr8jDL9ogf3TdJCEI1k4IsnBZsCM")
|
|
|
|
|
|
def today() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
|
|
def connect() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(DB_PATH, timeout=20)
|
|
conn.execute(
|
|
"""CREATE TABLE IF NOT EXISTS metrics_daily (
|
|
day TEXT PRIMARY KEY,
|
|
members_active INTEGER,
|
|
members_total INTEGER,
|
|
members_new INTEGER,
|
|
newsletter_subs INTEGER,
|
|
matrix_space_members INTEGER,
|
|
spend_usd REAL,
|
|
requests INTEGER,
|
|
recorded_at TEXT)"""
|
|
)
|
|
return conn
|
|
|
|
|
|
def _get_json(url: str, headers: dict | None = None):
|
|
req = urllib.request.Request(url, headers=headers or {})
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read() or b"{}")
|
|
|
|
|
|
# ---------------------------------------------------------------- sources
|
|
|
|
def from_sqlite() -> dict:
|
|
"""Members, plus the provisioning dates that give the growth curve."""
|
|
conn = connect()
|
|
try:
|
|
active = conn.execute("SELECT COUNT(*) FROM members WHERE active=1").fetchone()[0]
|
|
total = conn.execute("SELECT COUNT(*) FROM members").fetchone()[0]
|
|
joins = [
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT welcome_sent_at FROM members WHERE welcome_sent_at IS NOT NULL AND welcome_sent_at != ''"
|
|
)
|
|
]
|
|
finally:
|
|
conn.close()
|
|
return {"active": active, "total": total, "joins": joins}
|
|
|
|
|
|
def from_listmonk() -> int | None:
|
|
"""Confirmed subscribers on the members list, counted rather than trusted.
|
|
|
|
Listmonk's cached `subscriber_count` on the list object is stale and
|
|
double-counted (it read 142 for a list that truly holds 71), so we page the
|
|
subscribers endpoint and count confirmed memberships ourselves.
|
|
"""
|
|
token, user = os.environ.get("LISTMONK_API_TOKEN"), os.environ.get("LISTMONK_API_USER") or "portal-sync"
|
|
if not token:
|
|
print("metrics: LISTMONK_API_TOKEN not set — newsletter_subs left NULL")
|
|
return None
|
|
try:
|
|
total, page = 0, 1
|
|
while True:
|
|
data = _get_json(
|
|
f"{LISTMONK_BASE}/api/subscribers?list_id={LISTMONK_LIST_ID}&per_page=100&page={page}",
|
|
{"Authorization": f"token {user}:{token}"},
|
|
) or {}
|
|
payload = data.get("data") or {}
|
|
rows = payload.get("results") or []
|
|
total += sum(1 for s in rows if (s.get("status") or "") == "confirmed")
|
|
if not rows or len(rows) < 100 or page >= 20:
|
|
break
|
|
page += 1
|
|
return total
|
|
except Exception as exc: # noqa: BLE001 - report, never guess
|
|
print(f"metrics: listmonk read failed ({exc}) — newsletter_subs left NULL")
|
|
return None
|
|
|
|
|
|
def from_matrix() -> int | None:
|
|
token = os.environ.get("MATRIX_BOT_TOKEN")
|
|
if not token:
|
|
print("metrics: MATRIX_BOT_TOKEN not set — matrix_space_members left NULL")
|
|
return None
|
|
try:
|
|
data = _get_json(
|
|
f"{MATRIX_BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(SPACE)}/joined_members",
|
|
{"Authorization": f"Bearer {token}"},
|
|
) or {}
|
|
return len(data.get("joined") or {})
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"metrics: matrix read failed ({exc}) — matrix_space_members left NULL")
|
|
return None
|
|
|
|
|
|
def spend_by_day(since: dt.date | None = None) -> dict[str, tuple[float, int]]:
|
|
"""{day: (spend_usd, requests)} straight from LiteLLM's spend logs.
|
|
|
|
Prefers `model_group` the way aggregate_model_usage does, but here we only
|
|
need volume, so every row in the window counts.
|
|
"""
|
|
try:
|
|
import psycopg2
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"metrics: psycopg2 unavailable ({exc})")
|
|
return {}
|
|
try:
|
|
conn = psycopg2.connect(
|
|
host=os.environ.get("LITELLM_DB_HOST"),
|
|
port=os.environ.get("LITELLM_DB_PORT"),
|
|
user=os.environ.get("LITELLM_DB_USER"),
|
|
password=os.environ.get("LITELLM_DB_PASSWORD"),
|
|
dbname=os.environ.get("LITELLM_DB_NAME"),
|
|
connect_timeout=10,
|
|
sslmode="prefer",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"metrics: LiteLLM DB connect failed ({exc}) — spend left NULL")
|
|
return {}
|
|
try:
|
|
where, params = "", []
|
|
if since:
|
|
where, params = 'WHERE "startTime" >= %s', [since.isoformat()]
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
f"""SELECT to_char("startTime", 'YYYY-MM-DD') AS day,
|
|
COALESCE(SUM(spend), 0), COUNT(*)
|
|
FROM "LiteLLM_SpendLogs" {where}
|
|
GROUP BY 1 ORDER BY 1""",
|
|
params,
|
|
)
|
|
out = {row[0]: (round(float(row[1]), 6), int(row[2])) for row in cur.fetchall()}
|
|
finally:
|
|
conn.close()
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- writes
|
|
|
|
def upsert(day: str, row: dict) -> None:
|
|
conn = connect()
|
|
try:
|
|
conn.execute(
|
|
"""INSERT INTO metrics_daily
|
|
(day, members_active, members_total, members_new, newsletter_subs,
|
|
matrix_space_members, spend_usd, requests, recorded_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(day) DO UPDATE SET
|
|
members_active=excluded.members_active,
|
|
members_total=excluded.members_total,
|
|
members_new=excluded.members_new,
|
|
newsletter_subs=excluded.newsletter_subs,
|
|
matrix_space_members=excluded.matrix_space_members,
|
|
spend_usd=excluded.spend_usd,
|
|
requests=excluded.requests,
|
|
recorded_at=excluded.recorded_at""",
|
|
(
|
|
day,
|
|
row.get("members_active"),
|
|
row.get("members_total"),
|
|
row.get("members_new"),
|
|
row.get("newsletter_subs"),
|
|
row.get("matrix_space_members"),
|
|
row.get("spend_usd"),
|
|
row.get("requests"),
|
|
dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
|
|
),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def snapshot(day: str | None = None) -> dict:
|
|
day = day or today()
|
|
m = from_sqlite()
|
|
subs = from_listmonk()
|
|
space = from_matrix()
|
|
spend = spend_by_day(since=dt.date.fromisoformat(day) - dt.timedelta(days=1))
|
|
today_spend = spend.get(day, (None, None))
|
|
row = {
|
|
"members_active": m["active"],
|
|
"members_total": m["total"],
|
|
"members_new": sum(1 for j in m["joins"] if str(j)[:10] == day),
|
|
"newsletter_subs": subs,
|
|
"matrix_space_members": space,
|
|
"spend_usd": today_spend[0],
|
|
"requests": today_spend[1],
|
|
}
|
|
upsert(day, row)
|
|
print(f"metrics: recorded {day} -> {json.dumps(row)}")
|
|
return row
|
|
|
|
|
|
def backfill() -> None:
|
|
"""Rebuild history: cumulative members from provisioning dates, spend by day."""
|
|
m = from_sqlite()
|
|
joins = sorted(str(j)[:10] for j in m["joins"])
|
|
spend = spend_by_day()
|
|
if not joins and not spend:
|
|
print("metrics: backfill found no history")
|
|
return
|
|
|
|
first = min([d for d in joins] + ([min(spend)] if spend else []))
|
|
last = today()
|
|
print(
|
|
f"metrics: backfilling {first}..{last} "
|
|
f"({len(joins)} provisioning dates, {len(spend)} spend days)"
|
|
)
|
|
day = dt.date.fromisoformat(first)
|
|
end = dt.date.fromisoformat(last)
|
|
written = 0
|
|
while day <= end:
|
|
key = day.isoformat()
|
|
# Cumulative members AS OF this day, from provisioning dates. Members
|
|
# provisioned without a welcome email (manual/founder adds) are not in
|
|
# this history, so backfilled days are a LOWER BOUND — today's row below
|
|
# carries the exact count.
|
|
members_total = sum(1 for j in joins if j <= key) if joins else None
|
|
row = {
|
|
"members_total": members_total,
|
|
"members_new": sum(1 for j in joins if j == key),
|
|
"spend_usd": spend.get(key, (None, None))[0],
|
|
"requests": spend.get(key, (None, None))[1],
|
|
}
|
|
if key == today():
|
|
# today's row carries the exact live metrics
|
|
row.update(
|
|
members_active=m["active"],
|
|
members_total=m["total"],
|
|
newsletter_subs=from_listmonk(),
|
|
matrix_space_members=from_matrix(),
|
|
)
|
|
upsert(key, row)
|
|
written += 1
|
|
day += dt.timedelta(days=1)
|
|
print(f"metrics: backfill wrote {written} day(s)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if "--backfill" in sys.argv:
|
|
backfill()
|
|
else:
|
|
snapshot()
|