feat: anonymous opt-out install telemetry beacon
Add a once-a-day anonymous heartbeat (PostHog) so maintainers can see active installs and version/OS breakdowns. Self-hosted friendly: opt out via TELEMETRY_ENABLED in .env or the Admin -> System toggle (applied without a restart), or by clearing the endpoint/key. - Share version resolution in bot/utils/app_version.py so the admin sidebar and the beacon report the same build version - TelemetryWorker sends an opaque install id plus coarse facts only (version, OS/arch, python, locale, enabled providers, user-count range); never tokens, domains or user data - Register the worker in main_worker.py behind a Redis single-flight lock - Expose TELEMETRY_* settings and an Admin -> System manifest toggle - Document the payload and opt-out in docs/configuration/telemetry.md - Cover bucketing, payload shape and anonymity with tests
This commit is contained in:
3252a8
@@ -0,0 +1,212 @@
|
||||
"""Anonymous install telemetry beacon (self-hosted friendly, opt-out).
|
||||
|
||||
Once per ``TELEMETRY_INTERVAL_HOURS`` the worker sends a single obfuscation-free
|
||||
but fully anonymous "heartbeat" to a PostHog ingestion endpoint so the project
|
||||
maintainer can see how many installs are active and which versions/OSes are in
|
||||
use. No personal data, bot tokens, domains or user identities are sent — only
|
||||
an opaque per-install UUID plus coarse environment facts.
|
||||
|
||||
Operators can opt out in three independent ways, any of which stops the beacon:
|
||||
* ``TELEMETRY_ENABLED=false`` in ``.env``
|
||||
* the *System → Anonymous install analytics* toggle in the web admin (stored
|
||||
as a DB override and re-read every tick, so no restart is required)
|
||||
* leaving ``TELEMETRY_ENDPOINT`` / ``TELEMETRY_API_KEY`` empty in the image
|
||||
|
||||
Delivery is strictly fire-and-forget: every failure is swallowed so telemetry
|
||||
can never delay, block or crash the worker.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import platform
|
||||
import uuid
|
||||
from typing import Any, Dict, List
|
||||
|
||||
import aiohttp
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from bot.infra.redis import redis_lock
|
||||
from bot.utils.app_version import resolve_app_version, resolve_app_version_tag
|
||||
from config.settings import Settings
|
||||
from db.dal import app_settings_dal, user_dal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
INSTALLATION_ID_KEY = "TELEMETRY_INSTALLATION_ID"
|
||||
TELEMETRY_ENABLED_KEY = "TELEMETRY_ENABLED"
|
||||
HEARTBEAT_EVENT = "installation_heartbeat"
|
||||
INITIAL_DELAY_SECONDS = 300
|
||||
HTTP_TIMEOUT_SECONDS = 10
|
||||
|
||||
# Report the user count as a coarse range so individual installs stay anonymous
|
||||
# and the property keeps a low cardinality for breakdowns.
|
||||
_USER_BUCKETS = (
|
||||
(0, "0"),
|
||||
(10, "1-10"),
|
||||
(50, "11-50"),
|
||||
(200, "51-200"),
|
||||
(1000, "201-1000"),
|
||||
(5000, "1001-5000"),
|
||||
)
|
||||
|
||||
|
||||
def _bucket_users(count: int) -> str:
|
||||
for upper, label in _USER_BUCKETS:
|
||||
if count <= upper:
|
||||
return label
|
||||
return "5000+"
|
||||
|
||||
|
||||
class TelemetryWorker:
|
||||
def __init__(self, settings: Settings, session_factory: sessionmaker):
|
||||
self.settings = settings
|
||||
self.session_factory = session_factory
|
||||
self._stopped = asyncio.Event()
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stopped.set()
|
||||
|
||||
def _delivery_configured(self) -> bool:
|
||||
return bool(
|
||||
str(self.settings.TELEMETRY_ENDPOINT or "").strip()
|
||||
and str(self.settings.TELEMETRY_API_KEY or "").strip()
|
||||
)
|
||||
|
||||
async def run(self) -> None:
|
||||
if not self._delivery_configured():
|
||||
logger.info("Telemetry endpoint/key not configured; anonymous beacon disabled")
|
||||
return
|
||||
logger.info(
|
||||
"Anonymous install telemetry is ON (endpoint=%s, every %sh). "
|
||||
"It sends an opaque install id, version, OS and a user-count range — "
|
||||
"no personal data. Opt out via TELEMETRY_ENABLED=false or "
|
||||
"Admin -> System -> Anonymous install analytics. "
|
||||
"See docs/configuration/telemetry.md.",
|
||||
self.settings.TELEMETRY_ENDPOINT,
|
||||
self.settings.TELEMETRY_INTERVAL_HOURS,
|
||||
)
|
||||
await self._sleep(INITIAL_DELAY_SECONDS)
|
||||
while not self._stopped.is_set():
|
||||
try:
|
||||
await self._beacon_tick()
|
||||
except Exception:
|
||||
logger.exception("Telemetry beacon tick failed")
|
||||
await self._sleep(self._interval_seconds())
|
||||
|
||||
def _interval_seconds(self) -> int:
|
||||
return max(1, int(self.settings.TELEMETRY_INTERVAL_HOURS or 24)) * 3600
|
||||
|
||||
async def _sleep(self, seconds: float) -> None:
|
||||
try:
|
||||
await asyncio.wait_for(self._stopped.wait(), timeout=seconds)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
async def _beacon_tick(self) -> None:
|
||||
# A short-lived lock keeps a single beacon per interval even when the
|
||||
# worker is scaled to several replicas. Without Redis the lock yields
|
||||
# True, which is correct for the common single-worker deployment.
|
||||
async with redis_lock(
|
||||
self.settings,
|
||||
"telemetry-beacon",
|
||||
ttl_seconds=max(60, self._interval_seconds() // 2),
|
||||
) as acquired:
|
||||
if not acquired:
|
||||
return
|
||||
async with self.session_factory() as session:
|
||||
if not await self._is_enabled(session):
|
||||
return
|
||||
installation_id = await self._get_or_create_installation_id(session)
|
||||
payload = await self._build_payload(session, installation_id)
|
||||
await session.commit()
|
||||
await self._send(payload)
|
||||
|
||||
async def _is_enabled(self, session: AsyncSession) -> bool:
|
||||
# The web admin writes the toggle as a DB override. The worker process
|
||||
# does not apply overrides onto its in-memory Settings, so read it
|
||||
# straight from the table; the env default applies when unset.
|
||||
present, value = await app_settings_dal.get_override_value(session, TELEMETRY_ENABLED_KEY)
|
||||
if present:
|
||||
return bool(value)
|
||||
return bool(self.settings.TELEMETRY_ENABLED)
|
||||
|
||||
async def _get_or_create_installation_id(self, session: AsyncSession) -> str:
|
||||
present, value = await app_settings_dal.get_override_value(session, INSTALLATION_ID_KEY)
|
||||
if present and value:
|
||||
return str(value)
|
||||
installation_id = str(uuid.uuid4())
|
||||
await app_settings_dal.upsert_override(
|
||||
session,
|
||||
key=INSTALLATION_ID_KEY,
|
||||
value=installation_id,
|
||||
updated_by=None,
|
||||
)
|
||||
return installation_id
|
||||
|
||||
def _enabled_payment_providers(self) -> List[str]:
|
||||
try:
|
||||
from bot.payment_providers import iter_provider_specs
|
||||
|
||||
providers = [
|
||||
str(spec.id)
|
||||
for spec in iter_provider_specs()
|
||||
if spec.is_effectively_enabled(self.settings)
|
||||
]
|
||||
return sorted(set(providers))
|
||||
except Exception:
|
||||
logger.debug("Telemetry: failed to enumerate payment providers", exc_info=True)
|
||||
return []
|
||||
|
||||
async def _build_payload(self, session: AsyncSession, installation_id: str) -> Dict[str, Any]:
|
||||
try:
|
||||
user_count = await user_dal.count_all_users(session)
|
||||
except Exception:
|
||||
logger.debug("Telemetry: failed to count users", exc_info=True)
|
||||
user_count = 0
|
||||
|
||||
version = resolve_app_version()
|
||||
version_tag = resolve_app_version_tag()
|
||||
# Person properties (``$set``) snapshot the latest state per install, so
|
||||
# "version breakdown" in PostHog is a person-property breakdown.
|
||||
person_props = {
|
||||
"app_version": version,
|
||||
"app_version_tag": version_tag,
|
||||
"os": platform.system().lower() or "unknown",
|
||||
"arch": platform.machine().lower() or "unknown",
|
||||
"python_version": platform.python_version(),
|
||||
"locale": str(self.settings.DEFAULT_LANGUAGE or ""),
|
||||
"users_bucket": _bucket_users(int(user_count or 0)),
|
||||
"webapp_enabled": bool(self.settings.WEBAPP_ENABLED),
|
||||
"panel_configured": bool(str(self.settings.PANEL_API_URL or "").strip()),
|
||||
"payment_providers": self._enabled_payment_providers(),
|
||||
}
|
||||
properties = {
|
||||
**person_props,
|
||||
"$lib": "remnawave-minishop",
|
||||
"$lib_version": version,
|
||||
"$set": person_props,
|
||||
}
|
||||
return {
|
||||
"api_key": str(self.settings.TELEMETRY_API_KEY or "").strip(),
|
||||
"event": HEARTBEAT_EVENT,
|
||||
"distinct_id": installation_id,
|
||||
"properties": properties,
|
||||
}
|
||||
|
||||
async def _send(self, payload: Dict[str, Any]) -> None:
|
||||
url = str(self.settings.TELEMETRY_ENDPOINT or "").strip().rstrip("/") + "/capture/"
|
||||
timeout = aiohttp.ClientTimeout(total=HTTP_TIMEOUT_SECONDS)
|
||||
try:
|
||||
async with aiohttp.ClientSession(timeout=timeout) as http:
|
||||
async with http.post(url, json=payload) as resp:
|
||||
if resp.status >= 400:
|
||||
body = (await resp.text())[:200]
|
||||
logger.warning("Telemetry beacon rejected: HTTP %s %s", resp.status, body)
|
||||
else:
|
||||
logger.debug("Telemetry beacon delivered (HTTP %s)", resp.status)
|
||||
except Exception:
|
||||
# Never let telemetry surface as an error to operators.
|
||||
logger.debug("Telemetry beacon delivery failed", exc_info=True)
|
||||
Reference in New Issue
Block a user