Add metrics for users per community and user activity recency

app/backend/src/couchers/metrics.py

@@ -15,13 +15,25 @@
     multiprocess,
 )
 from prometheus_client.registry import CollectorRegistry
-from sqlalchemy import select
+from sqlalchemy import and_, case, select
 from sqlalchemy.sql import distinct, func
 from sqlalchemy.sql.selectable import Select
 
 from couchers.db import session_scope
 from couchers.helpers.completed_profile import has_completed_profile_expression
-from couchers.models import BackgroundJob, EventOccurrenceAttendee, HostingStatus, HostRequest, Message, Reference, User
+from couchers.materialized_views import ClusterSubscriptionCount
+from couchers.models import (
+    BackgroundJob,
+    Cluster,
+    EventOccurrenceAttendee,
+    HostingStatus,
+    HostRequest,
+    Message,
+    Node,
+    NodeType,
+    Reference,
+    User,
+)
 from couchers.models.moderation import (
     ModerationAction,
     ModerationObjectType,
@@ -94,6 +106,41 @@ def f() -> Any:
     return gauge
 
 
+# list of labeled gauges and the function to populate their label values just before collection
+_set_hacky_labeled_gauges_funcs: list[tuple[Gauge, Callable[[Gauge], None]]] = []
+
+
+def _make_labeled_gauge_from_query(
+    name: str,
+    description: str,
+    labelname: str,
+    statement: Select[Any],
+    default_label_values: list[str] | None = None,
+) -> Gauge:
+    """
+    Given a name, description, label name and statement, creates a gauge with one label set from the statement.
+
+    statement should be a sqlalchemy SELECT statement that returns rows of (label_value, count).
+
+    default_label_values, if given, are seeded to zero before the query results are applied, so that label
+    values with no matching rows are still emitted.
+    """
+
+    gauge = Gauge(name, description, labelnames=[labelname], multiprocess_mode="mostrecent")
+
+    def f(g: Gauge) -> None:
+        with tracer.start_as_current_span(f"metric.{name}"):
+            with session_scope() as session:
+                rows = session.execute(statement).all()
+        for label_value in default_label_values or []:
+            g.labels(label_value).set(0)
+        for label_value, count in rows:
+            g.labels(str(label_value)).set(count)
+
+    _set_hacky_labeled_gauges_funcs.append((gauge, f))
+    return gauge
+
+
 active_users_gauges: list[Gauge] = [
     _make_gauge_from_query(
         f"couchers_active_users_{name}",
@@ -114,6 +161,51 @@ def f() -> Any:
     "couchers_users", "Total number of users", select(func.count()).select_from(User).where(User.is_visible)
 )
 
+# Number of users per community, labeled by community name. Only includes communities at the region level or
+# broader (world, macroregion, region).
+users_per_community_gauge: Gauge = _make_labeled_gauge_from_query(
+    "couchers_users_per_community",
+    "Number of users per community, for regions and broader",
+    "community",
+    (
+        select(Cluster.name, func.coalesce(ClusterSubscriptionCount.count, 0))
+        .select_from(Node)
+        .join(Cluster, and_(Cluster.parent_node_id == Node.id, Cluster.is_official_cluster))
+        .outerjoin(ClusterSubscriptionCount, ClusterSubscriptionCount.cluster_id == Cluster.id)
+        .where(Node.node_type <= NodeType.region)
+    ),
+)
+
+# Number of users bucketed by how recently they were last active.
+_active_users_buckets: list[tuple[str, timedelta | None]] = [
+    ("<1d", timedelta(days=1)),
+    ("1d-1w", timedelta(days=7)),
+    ("1w-1m", timedelta(days=31)),
+    ("1m-6m", timedelta(days=183)),
+    ("6m-12m", timedelta(days=365)),
+    ("12m-24m", timedelta(days=730)),
+    ("24m+", None),
+]
+_active_users_age = func.now() - User.last_active
+active_users_by_recency_gauge: Gauge = _make_labeled_gauge_from_query(
+    "couchers_active_users_by_recency",
+    "Number of users bucketed by how recently they were last active",
+    "period",
+    (
+        select(
+            case(
+                *[(_active_users_age < interval, label) for label, interval in _active_users_buckets if interval],
+                else_=_active_users_buckets[-1][0],
+            ).label("period"),
+            func.count(),
+        )
+        .select_from(User)
+        .where(User.is_visible)
+        .group_by("period")
+    ),
+    default_label_values=[label for label, _ in _active_users_buckets],
+)
+
 man_gauge: Gauge = _make_gauge_from_query(
     "couchers_users_man",
     "Total number of users with gender 'Man'",
@@ -548,6 +640,8 @@ def app(environ: Any, start_response: Any) -> Any:
         # set hacky gauges
         for gauge, f in _set_hacky_gauges_funcs:
             gauge.set(f())
+        for gauge, labeled_f in _set_hacky_labeled_gauges_funcs:
+            labeled_f(gauge)
 
         data = generate_latest(registry)
         start_response("200 OK", [("Content-type", CONTENT_TYPE_LATEST), ("Content-Length", str(len(data)))])

app/backend/src/tests/test_metrics.py

@@ -0,0 +1,86 @@
+from datetime import timedelta
+
+import pytest
+from google.protobuf import empty_pb2
+from sqlalchemy import update
+
+from couchers.db import session_scope
+from couchers.materialized_views import refresh_materialized_views
+from couchers.metrics import (
+    _set_hacky_labeled_gauges_funcs,
+    active_users_by_recency_gauge,
+    users_per_community_gauge,
+)
+from couchers.models import User
+from couchers.utils import now
+from tests.fixtures.db import generate_user
+from tests.test_communities import create_community
+
+
+@pytest.fixture(autouse=True)
+def _(testconfig):
+    pass
+
+
+def _populate(gauge):
+    for registered_gauge, f in _set_hacky_labeled_gauges_funcs:
+        if registered_gauge is gauge:
+            f(registered_gauge)
+            return
+    raise AssertionError("gauge is not a registered labeled gauge")
+
+
+def _sample_values(gauge):
+    return {
+        sample.labels[gauge._labelnames[0]]: sample.value for metric in gauge.collect() for sample in metric.samples
+    }
+
+
+def test_users_per_community_gauge(db):
+    user1, _ = generate_user()
+    user2, _ = generate_user()
+    user3, _ = generate_user()
+
+    with session_scope() as session:
+        world = create_community(session, 0, 100, "World", [user1], [], None)
+        macroregion = create_community(session, 0, 50, "Macroregion", [user2], [], world)
+        region = create_community(session, 0, 25, "Region", [user3], [user1], macroregion)
+        # subregion is below the region level and must be excluded
+        create_community(session, 0, 10, "Subregion", [user2], [user3], region)
+
+    # the gauge reads from the cluster_subscription_counts materialized view
+    refresh_materialized_views(empty_pb2.Empty())
+
+    _populate(users_per_community_gauge)
+    values = _sample_values(users_per_community_gauge)
+
+    assert values["World"] == 1
+    assert values["Macroregion"] == 1
+    assert values["Region"] == 2
+    assert "Subregion" not in values
+
+
+def test_active_users_by_recency_gauge(db):
+    ages = {
+        "<1d": timedelta(hours=2),
+        "1d-1w": timedelta(days=3),
+        "1w-1m": timedelta(days=14),
+        "1m-6m": timedelta(days=60),
+        "6m-12m": timedelta(days=250),
+        "12m-24m": timedelta(days=500),
+        "24m+": timedelta(days=800),
+    }
+    user_ids_by_bucket = {}
+    for bucket in ages:
+        user, _ = generate_user()
+        user_ids_by_bucket[bucket] = user.id
+
+    with session_scope() as session:
+        for bucket, age in ages.items():
+            session.execute(update(User).where(User.id == user_ids_by_bucket[bucket]).values(last_active=now() - age))
+
+    _populate(active_users_by_recency_gauge)
+    values = _sample_values(active_users_by_recency_gauge)
+
+    for bucket in ages:
+        assert values[bucket] == 1