metrics: daily snapshot of members and usage, backfillable, plus a read route
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.
This commit is contained in:
1 parent
403a773f53
commit
70ae435867
4 files changed
+350
-1
No files matched your search
@@ -21,6 +21,10 @@
|
|||||||
"rag_refresh": {
|
"rag_refresh": {
|
||||||
"schedule": "30 3 * * *",
|
"schedule": "30 3 * * *",
|
||||||
"command": "/app/code/rag_refresh.py"
|
"command": "/app/code/rag_refresh.py"
|
||||||
|
},
|
||||||
|
"metrics_snapshot": {
|
||||||
|
"schedule": "15 3 * * *",
|
||||||
|
"command": "/app/code/metrics_snapshot.py"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
+2
-1
@@ -10,7 +10,8 @@ RUN pip install --no-cache-dir -r requirements.txt
|
|||||||
COPY app/ ./app/
|
COPY app/ ./app/
|
||||||
COPY sweep.sh /app/code/sweep.sh
|
COPY sweep.sh /app/code/sweep.sh
|
||||||
COPY rag_refresh.py /app/code/rag_refresh.py
|
COPY rag_refresh.py /app/code/rag_refresh.py
|
||||||
RUN chmod +x /app/code/sweep.sh /app/code/rag_refresh.py
|
COPY metrics_snapshot.py /app/code/metrics_snapshot.py
|
||||||
|
RUN chmod +x /app/code/sweep.sh /app/code/rag_refresh.py /app/code/metrics_snapshot.py
|
||||||
|
|
||||||
# Run as root (the base image default). Cloudron's sandboxing (read-only
|
# Run as root (the base image default). Cloudron's sandboxing (read-only
|
||||||
# rootfs, AppArmor, dropped capabilities) still applies to root inside the
|
# rootfs, AppArmor, dropped capabilities) still applies to root inside the
|
||||||
|
|||||||
+60
@@ -607,6 +607,24 @@ def get_db() -> sqlite3.Connection:
|
|||||||
"active INTEGER DEFAULT 1"
|
"active INTEGER DEFAULT 1"
|
||||||
")"
|
")"
|
||||||
)
|
)
|
||||||
|
# Daily metrics snapshot, written by /app/code/metrics_snapshot.py from the
|
||||||
|
# scheduler and read by the admin panel + dashboard. Cumulative columns
|
||||||
|
# (members_*, newsletter_subs, matrix_space_members) are counts AS OF that
|
||||||
|
# day; spend_usd/requests are THAT day's traffic. A NULL column means the
|
||||||
|
# source could not be read that day — charts must show a gap, never a zero.
|
||||||
|
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"
|
||||||
|
")"
|
||||||
|
)
|
||||||
# Member-managed API keys (created via the broker). Each row is one API key
|
# Member-managed API keys (created via the broker). Each row is one API key
|
||||||
# under the member's team, tracked so we can list/revoke by name without
|
# under the member's team, tracked so we can list/revoke by name without
|
||||||
# exposing the chat key. `sk_token` is the full sk- key (revealed once).
|
# exposing the chat key. `sk_token` is the full sk- key (revealed once).
|
||||||
@@ -3036,6 +3054,48 @@ async def admin_overview(request: Request):
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@app.get("/admin/metrics/series")
|
||||||
|
@limiter.limit("30/minute")
|
||||||
|
async def admin_metrics_series(request: Request, days: int = 90):
|
||||||
|
"""Daily metrics series for charts.
|
||||||
|
|
||||||
|
Authenticated by the X-Admin-Token header. Returns oldest-first rows from
|
||||||
|
`metrics_daily` for the last `days` days, for the admin panel and the member
|
||||||
|
dashboard to chart from one source of truth. A NULL column means that source
|
||||||
|
could not be read on that day: render a gap, not a zero.
|
||||||
|
|
||||||
|
Cumulative columns (members_*, newsletter_subs, matrix_space_members) are
|
||||||
|
counts as of that day; spend_usd/requests are that day's traffic only.
|
||||||
|
Backfilled days carry a LOWER BOUND for members_total (members provisioned
|
||||||
|
without a welcome email are absent from the history).
|
||||||
|
"""
|
||||||
|
_verify_admin_token(request)
|
||||||
|
limit = max(1, min(int(days or 90), 730))
|
||||||
|
conn = get_db()
|
||||||
|
try:
|
||||||
|
rows = conn.execute(
|
||||||
|
"SELECT day, members_active, members_total, members_new, newsletter_subs, "
|
||||||
|
"matrix_space_members, spend_usd, requests, recorded_at "
|
||||||
|
"FROM metrics_daily ORDER BY day DESC LIMIT ?",
|
||||||
|
(limit,),
|
||||||
|
).fetchall()
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
cols = [
|
||||||
|
"day",
|
||||||
|
"members_active",
|
||||||
|
"members_total",
|
||||||
|
"members_new",
|
||||||
|
"newsletter_subs",
|
||||||
|
"matrix_space_members",
|
||||||
|
"spend_usd",
|
||||||
|
"requests",
|
||||||
|
"recorded_at",
|
||||||
|
]
|
||||||
|
series = [dict(zip(cols, row)) for row in reversed(rows)]
|
||||||
|
return {"series": series, "count": len(series), "days": limit}
|
||||||
|
|
||||||
|
|
||||||
@app.get("/admin/model-usage")
|
@app.get("/admin/model-usage")
|
||||||
@limiter.limit("30/minute")
|
@limiter.limit("30/minute")
|
||||||
async def admin_model_usage(request: Request, days: int = 0):
|
async def admin_model_usage(request: Request, days: int = 0):
|
||||||
|
|||||||
@@ -0,0 +1,284 @@
|
|||||||
|
#!/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()
|
||||||
Reference in new issue
Block a user