Coverage for app/backend/src/couchers/metrics.py: 96%
266 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1import threading
2import time
3from collections.abc import Callable, Sequence
4from concurrent.futures import ThreadPoolExecutor
5from datetime import datetime, timedelta
6from typing import Any, cast
8from opentelemetry import trace
9from prometheus_client import (
10 CONTENT_TYPE_LATEST,
11 CollectorRegistry,
12 Counter,
13 Gauge,
14 Histogram,
15 exposition,
16 generate_latest,
17 multiprocess,
18)
19from prometheus_client.registry import CollectorRegistry
20from sqlalchemy import Engine, and_, select
21from sqlalchemy.pool import QueuePool
22from sqlalchemy.sql import distinct, func
23from sqlalchemy.sql.elements import ColumnElement
24from sqlalchemy.sql.selectable import Select
26from couchers import experimentation
27from couchers.config import config
28from couchers.db import session_scope
29from couchers.helpers.completed_profile import has_completed_profile_expression
30from couchers.materialized_views import ClusterSubscriptionCount
31from couchers.models import (
32 BackgroundJob,
33 ClientPlatform,
34 Cluster,
35 EventOccurrenceAttendee,
36 HostingStatus,
37 HostRequest,
38 Message,
39 Node,
40 NodeType,
41 NonvisibleUserAccessType,
42 NonvisibleUserState,
43 Reference,
44 User,
45 UserActivity,
46)
47from couchers.models.moderation import (
48 ModerationAction,
49 ModerationObjectType,
50 ModerationQueueItem,
51 ModerationState,
52 ModerationTrigger,
53 ModerationVisibility,
54)
55from couchers.models.uploads import PhotoGalleryItem
56from couchers.perf import PerfResult
58tracer = trace.get_tracer(__name__)
60registry: CollectorRegistry = CollectorRegistry()
61multiprocess.MultiProcessCollector(registry) # type: ignore[no-untyped-call]
63_INF: float = float("inf")
65# Dense from 1ms to ~300ms where most calls land, sparse out to 10min for long background jobs.
66MACHINE_DURATION_SECONDS: tuple[float, ...] = (
67 0.001,
68 0.0025,
69 0.005,
70 0.0075,
71 0.01,
72 0.015,
73 0.02,
74 0.03,
75 0.04,
76 0.05,
77 0.06,
78 0.075,
79 0.1,
80 0.125,
81 0.15,
82 0.2,
83 0.25,
84 0.3,
85 0.4,
86 0.5,
87 0.75,
88 1.0,
89 1.5,
90 2.0,
91 3.0,
92 5.0,
93 7.5,
94 10.0,
95 15.0,
96 30.0,
97 60,
98 120,
99 300,
100 600,
101 _INF,
102)
104start_time_gauge: Gauge = Gauge(
105 "couchers_start_time_seconds",
106 "Unix timestamp of when the process started",
107 multiprocess_mode="max",
108)
109start_time_gauge.set(time.time())
111commit_timestamp_gauge: Gauge = Gauge(
112 "couchers_commit_timestamp_seconds",
113 "Unix timestamp of the deployed commit, 0 if not a CI build",
114 multiprocess_mode="max",
115)
116# left at its default of 0 when COMMIT_TIMESTAMP is empty (i.e. not a CI build)
117if config.COMMIT_TIMESTAMP: 117 ↛ 120line 117 didn't jump to line 120 because the condition on line 117 was always true
118 commit_timestamp_gauge.set(datetime.fromisoformat(config.COMMIT_TIMESTAMP).timestamp())
120jobs_duration_histogram: Histogram = Histogram(
121 "couchers_background_jobs_seconds",
122 "Durations of background jobs",
123 labelnames=["job", "status", "attempt", "exception"],
124 buckets=MACHINE_DURATION_SECONDS,
125)
128def observe_in_jobs_duration_histogram(
129 job_type: str, job_state: str, try_count: int, exception_name: str, duration_s: float
130) -> None:
131 jobs_duration_histogram.labels(job_type, job_state, str(try_count), exception_name).observe(duration_s)
134jobs_queued_histogram: Histogram = Histogram(
135 "couchers_background_jobs_queued_seconds",
136 "Time background job spent queued before being picked up",
137 labelnames=["priority"],
138 buckets=(
139 0.01,
140 0.05,
141 0.1,
142 0.5,
143 1.0,
144 2.5,
145 5.0,
146 10,
147 20,
148 30,
149 40,
150 50,
151 60,
152 90,
153 120,
154 180,
155 240,
156 300,
157 360,
158 420,
159 480,
160 540,
161 600,
162 720,
163 900,
164 1800,
165 3600,
166 _INF,
167 ),
168)
171servicer_duration_histogram: Histogram = Histogram(
172 "couchers_servicer_duration_seconds",
173 "Durations of processing gRPC calls",
174 labelnames=["method", "logged_in", "code", "exception"],
175 buckets=MACHINE_DURATION_SECONDS,
176)
179def observe_in_servicer_duration_histogram(
180 method: str, user_id: Any, status_code: str, exception_type: str, duration_s: float
181) -> None:
182 servicer_duration_histogram.labels(method, user_id is not None, status_code, exception_type).observe(duration_s)
185servicer_setup_errors_counter: Counter = Counter(
186 "couchers_servicer_setup_errors_total",
187 "Number of unexpected errors raised during gRPC interceptor setup, before the handler is invoked",
188 labelnames=["method", "exception"],
189)
192def observe_in_servicer_setup_errors_counter(method: str, exception_type: str) -> None:
193 servicer_setup_errors_counter.labels(method, exception_type).inc()
196# Per-request resource accounting (see couchers/perf.py), labelled by method only to keep cardinality modest. The
197# histogram _sum gives the cost rate per endpoint via rate() (DB-seconds/sec, CPU-seconds/sec); the buckets give the
198# per-call distribution.
199servicer_db_time_histogram: Histogram = Histogram(
200 "couchers_servicer_db_time_seconds",
201 "Time spent in DB cursor execution per gRPC call",
202 labelnames=["method"],
203 buckets=MACHINE_DURATION_SECONDS,
204)
205servicer_cpu_time_histogram: Histogram = Histogram(
206 "couchers_servicer_cpu_seconds",
207 "Backend thread CPU time per gRPC call",
208 labelnames=["method"],
209 buckets=MACHINE_DURATION_SECONDS,
210)
211# Fibonacci bucket boundaries: roughly exponential, good resolution for an unbounded value
212servicer_db_query_count_histogram: Histogram = Histogram(
213 "couchers_servicer_db_query_count",
214 "Number of SQL statements executed per gRPC call",
215 labelnames=["method"],
216 buckets=(1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, _INF),
217)
218servicer_db_write_query_count_histogram: Histogram = Histogram(
219 "couchers_servicer_db_write_query_count",
220 "Number of INSERT/UPDATE/DELETE statements executed per gRPC call",
221 labelnames=["method"],
222 buckets=(1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, _INF),
223)
226def observe_in_servicer_perf_histograms(method: str, perf: PerfResult | None) -> None:
227 if perf is None: 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true
228 return
229 servicer_db_time_histogram.labels(method).observe(perf.db_time_ms / 1000)
230 servicer_cpu_time_histogram.labels(method).observe(perf.cpu_ms / 1000)
231 servicer_db_query_count_histogram.labels(method).observe(perf.db_query_count)
232 servicer_db_write_query_count_histogram.labels(method).observe(perf.db_write_query_count)
235# Auth/setup phase (everything before the handler body), same db-vs-cpu split as the handler-body histograms above.
236servicer_setup_db_time_histogram: Histogram = Histogram(
237 "couchers_servicer_setup_db_time_seconds",
238 "Time spent in DB cursor execution during the auth/setup phase per gRPC call",
239 labelnames=["method"],
240 buckets=MACHINE_DURATION_SECONDS,
241)
242servicer_setup_cpu_time_histogram: Histogram = Histogram(
243 "couchers_servicer_setup_cpu_seconds",
244 "Backend thread CPU time during the auth/setup phase per gRPC call",
245 labelnames=["method"],
246 buckets=MACHINE_DURATION_SECONDS,
247)
250def observe_in_servicer_setup_histogram(method: str, perf: PerfResult | None) -> None:
251 if perf is None: 251 ↛ 252line 251 didn't jump to line 252 because the condition on line 251 was never true
252 return
253 servicer_setup_db_time_histogram.labels(method).observe(perf.db_time_ms / 1000)
254 servicer_setup_cpu_time_histogram.labels(method).observe(perf.cpu_ms / 1000)
257servicer_pool_wait_histogram: Histogram = Histogram(
258 "couchers_servicer_pool_wait_seconds",
259 "Time spent waiting to check out a DB connection from the pool per gRPC call",
260 labelnames=["method"],
261 buckets=MACHINE_DURATION_SECONDS,
262)
265def observe_in_servicer_pool_wait_histogram(method: str, pool_wait_s: float) -> None:
266 servicer_pool_wait_histogram.labels(method).observe(pool_wait_s)
269# Separate diagnostic, not part of the additive duration pie: "serialize" runs after the duration window closes.
270servicer_serde_histogram: Histogram = Histogram(
271 "couchers_servicer_serde_seconds",
272 "Protobuf request deserialization / response serialization time per gRPC call",
273 labelnames=["method", "direction"],
274 buckets=MACHINE_DURATION_SECONDS,
275)
278def observe_in_servicer_serde_histogram(method: str, direction: str, serde_s: float) -> None:
279 servicer_serde_histogram.labels(method, direction).observe(serde_s)
282# liveall keeps one series per worker pid (and drops dead workers), so these also show load balance across workers.
283# Updated from inside each worker since the /metrics scrape runs in the parent, which has neither pool.
284grpc_in_flight_gauge: Gauge = Gauge(
285 "couchers_grpc_in_flight",
286 "Outstanding gRPC calls (running plus queued for a server thread), per worker process",
287 multiprocess_mode="liveall",
288)
289grpc_threadpool_queue_depth_gauge: Gauge = Gauge(
290 "couchers_grpc_threadpool_queue_depth",
291 "gRPC calls queued waiting for a free server thread, per worker process",
292 multiprocess_mode="liveall",
293)
294db_pool_checked_out_gauge: Gauge = Gauge(
295 "couchers_db_pool_checked_out",
296 "Checked-out DB connections, per worker process",
297 multiprocess_mode="liveall",
298)
301def start_worker_resource_sampler(executor: ThreadPoolExecutor, engine: Engine, interval: float = 1.0) -> None:
302 def sample() -> None:
303 while True:
304 # _work_queue is private but stable: tasks gRPC has submitted that no thread has picked up yet
305 grpc_threadpool_queue_depth_gauge.set(executor._work_queue.qsize())
306 db_pool_checked_out_gauge.set(cast(QueuePool, engine.pool).checkedout())
307 time.sleep(interval)
309 threading.Thread(target=sample, daemon=True, name="resource-sampler").start()
312supervised_children_alive_gauge: Gauge = Gauge(
313 "couchers_supervised_children_alive",
314 "Child processes (API workers, background workers, scheduler) the supervisor currently sees alive",
315 multiprocess_mode="mostrecent",
316)
319# Simple count of API calls, broken down by method and the client platform header. Cheap (a counter, no buckets) and
320# answers "how much traffic comes from each platform".
321api_calls_counter: Counter = Counter(
322 "couchers_api_calls_total",
323 "Number of gRPC API calls",
324 labelnames=["method", "platform"],
325)
328def observe_api_call(method: str, client_platform: ClientPlatform | None) -> None:
329 api_calls_counter.labels(method, client_platform.name if client_platform is not None else "unknown").inc()
332# list of gauge names and function to execute to set value to
333# the python prometheus client does not support Gauge.set_function, so instead we hack around it and set each gauge just
334# before collection with this
335_set_hacky_gauges_funcs: list[tuple[Gauge, Callable[[], Any]]] = []
338def _make_gauge_from_query(name: str, description: str, statement: Select[Any]) -> Gauge:
339 """
340 Given a name, description and statement that is a sqlalchemy statement, creates a gauge from it
342 statement should be a sqlalchemy SELECT statement that returns a single number
343 """
345 def f() -> Any:
346 with tracer.start_as_current_span(f"metric.{name}"):
347 with session_scope() as session:
348 return session.execute(statement).scalar_one()
350 gauge = Gauge(name, description, multiprocess_mode="mostrecent")
351 _set_hacky_gauges_funcs.append((gauge, f))
352 return gauge
355# list of labeled gauges and the function to populate their label values just before collection
356_set_hacky_labeled_gauges_funcs: list[tuple[Gauge, Callable[[Gauge], None]]] = []
359def _make_labeled_gauge_from_query(
360 name: str,
361 description: str,
362 labelname: str,
363 statement: Select[Any],
364) -> Gauge:
365 """
366 Given a name, description, label name and statement, creates a gauge with one label set from the statement.
368 statement should be a sqlalchemy SELECT statement that returns rows of (label_value, count).
369 """
371 gauge = Gauge(name, description, labelnames=[labelname], multiprocess_mode="mostrecent")
373 def f(g: Gauge) -> None:
374 with tracer.start_as_current_span(f"metric.{name}"):
375 with session_scope() as session:
376 rows = session.execute(statement).all()
377 for label_value, count in rows:
378 g.labels(str(label_value)).set(count)
380 _set_hacky_labeled_gauges_funcs.append((gauge, f))
381 return gauge
384# list of functions that each run one query and populate several gauges from the single result row
385_set_hacky_multi_gauges_funcs: list[Callable[[], None]] = []
388def _make_gauges_from_single_pass(
389 span_name: str,
390 make_statement: Callable[[list[Any]], Select[Any]],
391 specs: Sequence[tuple[str, str, ColumnElement[bool] | None]],
392 labeled_specs: Sequence[tuple[Gauge, str, ColumnElement[bool]]] = (),
393) -> list[Gauge]:
394 """
395 Creates a gauge per spec, plus label values on already-created labeled gauges, all from one pass over the table.
397 Each spec is (name, description, condition) and counts the rows matching that condition, or every row if it is
398 None; each labeled spec is (gauge, label value, condition). make_statement is handed the count columns and
399 returns the statement to run them in, so the caller owns the FROM and any joins.
401 Counting with count(*) FILTER in a single statement rather than one statement per gauge is the whole point: as
402 separate queries these were a full sequential scan of the table each, on every scrape.
403 """
404 conditions = [condition for _, _, condition in specs] + [condition for _, _, condition in labeled_specs]
405 keys = [f"c{i}" for i in range(len(conditions))]
406 statement = make_statement(
407 [
408 (func.count() if condition is None else func.count().filter(condition)).label(key)
409 for key, condition in zip(keys, conditions)
410 ]
411 )
412 gauges = [Gauge(name, description, multiprocess_mode="mostrecent") for name, description, _ in specs]
414 def f() -> None:
415 with tracer.start_as_current_span(f"metric.{span_name}"):
416 with session_scope() as session:
417 row = session.execute(statement).one()._mapping
418 for gauge, key in zip(gauges, keys):
419 gauge.set(row[key])
420 for (labeled_gauge, label_value, _), labeled_key in zip(labeled_specs, keys[len(specs) :]):
421 labeled_gauge.labels(label_value).set(row[labeled_key])
423 _set_hacky_multi_gauges_funcs.append(f)
424 return gauges
427_active_user_periods: list[tuple[str, str, timedelta]] = [
428 ("5m", "5 min", timedelta(minutes=5)),
429 ("24h", "24 hours", timedelta(hours=24)),
430 ("1month", "1 month", timedelta(weeks=4)),
431 ("3month", "3 months", timedelta(weeks=13)),
432 ("6month", "6 months", timedelta(weeks=26)),
433 ("12month", "12 months", timedelta(days=365)),
434]
436# Number of users bucketed by how recently they were last active. Ordered upper bounds, made mutually exclusive by
437# _active_users_bucket_specs so every bucket can be counted in the same pass.
438_active_users_buckets: list[tuple[str, timedelta | None]] = [
439 ("<1d", timedelta(days=1)),
440 ("1d-1w", timedelta(days=7)),
441 ("1w-1m", timedelta(weeks=4)),
442 ("1m-6m", timedelta(weeks=26)),
443 ("6m-12m", timedelta(days=365)),
444 ("12m-24m", timedelta(days=730)),
445 ("24m+", None),
446]
447_active_users_age = func.now() - User.last_active
449active_users_by_recency_gauge: Gauge = Gauge(
450 "couchers_active_users_by_recency",
451 "Number of users bucketed by how recently they were last active",
452 labelnames=["period"],
453 multiprocess_mode="mostrecent",
454)
457def _active_users_bucket_specs() -> list[tuple[Gauge, str, ColumnElement[bool]]]:
458 specs = []
459 lower: timedelta | None = None
460 for label, upper in _active_users_buckets:
461 bounds = []
462 if lower is not None:
463 bounds.append(_active_users_age >= lower)
464 if upper is not None:
465 bounds.append(_active_users_age < upper)
466 specs.append((active_users_by_recency_gauge, label, and_(*bounds)))
467 lower = upper
468 return specs
471def _make_users_gauges() -> list[Gauge]:
472 # Galleries holding at least one photo, outer joined below so the avatar check in the completed profile spec is a
473 # hash join rather than a subplan run once per user inside the FILTER clause.
474 galleries_with_photos = select(PhotoGalleryItem.gallery_id).distinct().subquery("galleries_with_photos")
476 specs: list[tuple[str, str, ColumnElement[bool] | None]] = [
477 ("couchers_users", "Total number of users", None),
478 *[
479 (
480 f"couchers_active_users_{name}",
481 f"Number of active users in the last {description}",
482 _active_users_age < interval,
483 )
484 for name, description, interval in _active_user_periods
485 ],
486 ("couchers_users_man", "Total number of users with gender 'Man'", User.gender == "Man"),
487 ("couchers_users_woman", "Total number of users with gender 'Woman'", User.gender == "Woman"),
488 ("couchers_users_nonbinary", "Total number of users with gender 'Non-binary'", User.gender == "Non-binary"),
489 (
490 "couchers_users_can_host",
491 "Total number of users with hosting status 'can_host'",
492 User.hosting_status == HostingStatus.can_host,
493 ),
494 (
495 "couchers_users_cant_host",
496 "Total number of users with hosting status 'cant_host'",
497 User.hosting_status == HostingStatus.cant_host,
498 ),
499 (
500 "couchers_users_maybe",
501 "Total number of users with hosting status 'maybe'",
502 User.hosting_status == HostingStatus.maybe,
503 ),
504 (
505 "couchers_users_completed_profile",
506 "Total number of users with a completed profile",
507 has_completed_profile_expression(galleries_with_photos),
508 ),
509 (
510 "couchers_users_completed_my_home",
511 "Total number of users with a completed my home section",
512 cast(ColumnElement[bool], User.has_completed_my_home),
513 ),
514 ]
516 return _make_gauges_from_single_pass(
517 "couchers_users",
518 lambda columns: (
519 select(*columns)
520 .select_from(User)
521 .outerjoin(galleries_with_photos, galleries_with_photos.c.gallery_id == User.profile_gallery_id)
522 .where(User.is_visible)
523 ),
524 specs,
525 _active_users_bucket_specs(),
526 )
529# Each of these was its own SELECT count(*) FROM users, which added up to roughly sixteen full scans of the table on
530# every scrape and dominated all sequential tuple reads in the database.
531users_gauges: list[Gauge] = _make_users_gauges()
533# Number of users per community, labeled by community name. Only includes communities at the region level or
534# broader (world, macroregion, region).
535users_per_community_gauge: Gauge = _make_labeled_gauge_from_query(
536 "couchers_users_per_community",
537 "Number of users per community, for regions and broader",
538 "community",
539 (
540 select(Cluster.name, func.coalesce(ClusterSubscriptionCount.count, 0))
541 .select_from(Node)
542 .join(Cluster, and_(Cluster.parent_node_id == Node.id, Cluster.is_official_cluster))
543 .outerjoin(ClusterSubscriptionCount, ClusterSubscriptionCount.cluster_id == Cluster.id)
544 .where(Node.node_type <= NodeType.region)
545 ),
546)
548# Window for the per-platform daily-active-user metrics. Kept to 24h so the user_activity scan stays cheap (an index
549# scan of just the last day's rows), letting these gauges be computed inline on every scrape.
550_ACTIVE_USERS_BY_PLATFORM_WINDOW = timedelta(hours=24)
551# Platforms counted as "mobile" for the mobile-share fraction (native apps plus the mobile web viewport).
552_MOBILE_PLATFORMS = [ClientPlatform.web_mobile, ClientPlatform.app_ios, ClientPlatform.app_android]
555def active_users_by_platform_statement() -> Select[Any]:
556 # one scan of the last 24h of user_activity: distinct active users in total, the mobile subset (for the share
557 # fraction), and a breakdown per platform. client_platform is set from a header the client explicitly sends; it's
558 # null for some other client (e.g. an API key script) or activity from before the header existed, so the
559 # per-platform counts don't sum to the total and "mobile" needs its own union count rather than summing labels.
560 distinct_users = func.count(distinct(UserActivity.user_id))
561 return (
562 select(
563 distinct_users.label("total"),
564 distinct_users.filter(UserActivity.client_platform.in_(_MOBILE_PLATFORMS)).label("mobile"),
565 *[
566 distinct_users.filter(UserActivity.client_platform == platform).label(platform.name)
567 for platform in ClientPlatform
568 ],
569 )
570 .select_from(UserActivity)
571 .join(User, User.id == UserActivity.user_id)
572 .where(User.is_visible)
573 .where(UserActivity.period > func.now() - _ACTIVE_USERS_BY_PLATFORM_WINDOW)
574 )
577# Distinct active users in the last 24h, split by client platform.
578active_users_by_platform_gauge: Gauge = Gauge(
579 "couchers_active_users_by_platform",
580 "Distinct active users in the last 24h, split by client platform (web_desktop, web_mobile, app_ios, app_android)",
581 labelnames=["platform"],
582 multiprocess_mode="mostrecent",
583)
585# Fraction of the last 24h's distinct active users who had any mobile activity. The headline "mobile is key" number.
586active_users_mobile_fraction_gauge: Gauge = Gauge(
587 "couchers_active_users_mobile_fraction",
588 "Fraction of distinct active users in the last 24h with any mobile activity (web_mobile, app_ios, app_android)",
589 multiprocess_mode="mostrecent",
590)
593def _set_active_users_by_platform(gauge: Gauge) -> None:
594 with tracer.start_as_current_span("metric.couchers_active_users_by_platform"):
595 with session_scope() as session:
596 row = session.execute(active_users_by_platform_statement()).one()._mapping
597 for platform in ClientPlatform:
598 gauge.labels(platform.name).set(row[platform.name])
599 total = row["total"]
600 active_users_mobile_fraction_gauge.set(row["mobile"] / total if total else 0.0)
603_set_hacky_labeled_gauges_funcs.append((active_users_by_platform_gauge, _set_active_users_by_platform))
605sent_message_gauge: Gauge = _make_gauge_from_query(
606 "couchers_users_sent_message",
607 "Total number of users who have sent a message",
608 (select(func.count(distinct(Message.author_id))).join(User, User.id == Message.author_id).where(User.is_visible)),
609)
611sent_request_gauge: Gauge = _make_gauge_from_query(
612 "couchers_users_sent_request",
613 "Total number of users who have sent a host request",
614 (
615 select(func.count(distinct(HostRequest.initiator_user_id)))
616 .join(User, User.id == HostRequest.initiator_user_id)
617 .where(User.is_visible)
618 ),
619)
621has_reference_gauge: Gauge = _make_gauge_from_query(
622 "couchers_users_has_reference",
623 "Total number of users who have a reference",
624 (
625 select(func.count(distinct(Reference.to_user_id)))
626 .join(User, User.id == Reference.to_user_id)
627 .where(User.is_visible)
628 ),
629)
631rsvpd_to_event_gauge: Gauge = _make_gauge_from_query(
632 "couchers_users_rsvpd_to_event",
633 "Total number of users who have RSVPd to an event",
634 (
635 select(func.count(distinct(EventOccurrenceAttendee.user_id)))
636 .join(User, User.id == EventOccurrenceAttendee.user_id)
637 .where(User.is_visible)
638 ),
639)
641background_jobs_ready_to_execute_gauge: Gauge = _make_gauge_from_query(
642 "couchers_background_jobs_ready_to_execute",
643 "Total number of background jobs ready to execute",
644 select(func.count()).select_from(BackgroundJob).where(BackgroundJob.ready_for_retry),
645)
647background_jobs_no_jobs_counter: Counter = Counter(
648 "couchers_background_jobs_no_jobs_total",
649 "Number of times a bg worker tries to grab a job but there is none",
650)
652background_jobs_got_job_counter: Counter = Counter(
653 "couchers_background_jobs_got_job_total",
654 "Number of times a bg worker grabbed a job",
655)
658signup_initiations_counter: Counter = Counter(
659 "couchers_signup_initiations_total",
660 "Number of initiated signups",
661)
662signup_completions_counter: Counter = Counter(
663 "couchers_signup_completions_total",
664 "Number of completed signups",
665 labelnames=["gender"],
666)
667# Per-step signup funnel counters. Each fires once, the first time a signup flow satisfies the given gate, so
668# that step_total/initiations_total gives the fraction of signups that reached that step. Unlabeled to match
669# signup_initiations_counter for clean ratios.
670signup_account_filled_counter: Counter = Counter(
671 "couchers_signup_account_filled_total",
672 "Number of signup flows that filled in their account details",
673)
674signup_email_verified_counter: Counter = Counter(
675 "couchers_signup_email_verified_total",
676 "Number of signup flows that verified their email address",
677)
678signup_guidelines_accepted_counter: Counter = Counter(
679 "couchers_signup_guidelines_accepted_total",
680 "Number of signup flows that accepted the community guidelines",
681)
682signup_motivations_filled_counter: Counter = Counter(
683 "couchers_signup_motivations_filled_total",
684 "Number of signup flows that filled in their motivations",
685)
686signup_time_histogram: Histogram = Histogram(
687 "couchers_signup_time_seconds",
688 "Time taken for a user to sign up",
689 labelnames=["gender"],
690 buckets=(30, 60, 90, 120, 180, 240, 300, 360, 420, 480, 540, 600, 900, 1200, 1800, 3600, 7200, _INF),
691)
693logins_counter: Counter = Counter(
694 "couchers_logins_total",
695 "Number of logins",
696 labelnames=["gender"],
697)
699password_reset_initiations_counter: Counter = Counter(
700 "couchers_password_reset_initiations_total",
701 "Number of password reset initiations",
702)
703password_reset_completions_counter: Counter = Counter(
704 "couchers_password_reset_completions_total",
705 "Number of password reset completions",
706)
708account_deletion_initiations_counter: Counter = Counter(
709 "couchers_account_deletion_initiations_total",
710 "Number of account deletion initiations",
711 labelnames=["gender"],
712)
713account_deletion_completions_counter: Counter = Counter(
714 "couchers_account_deletion_completions_total",
715 "Number of account deletion completions",
716 labelnames=["gender"],
717)
718account_recoveries_counter: Counter = Counter(
719 "couchers_account_recoveries_total",
720 "Number of account recoveries",
721 labelnames=["gender"],
722)
724strong_verification_initiations_counter: Counter = Counter(
725 "couchers_strong_verification_initiations_total",
726 "Number of strong verification initiations",
727 labelnames=["gender"],
728)
729strong_verification_completions_counter: Counter = Counter(
730 "couchers_strong_verification_completions_total",
731 "Number of strong verification completions",
732)
733strong_verification_data_deletions_counter: Counter = Counter(
734 "couchers_strong_verification_data_deletions_total",
735 "Number of strong verification data deletions",
736 labelnames=["gender"],
737)
739host_requests_sent_counter: Counter = Counter(
740 "couchers_host_requests_total",
741 "Number of host requests sent",
742 labelnames=["from_gender", "to_gender"],
743)
744host_request_responses_counter: Counter = Counter(
745 "couchers_host_requests_responses_total",
746 "Number of responses to host requests",
747 labelnames=["responder_gender", "other_gender", "response_type"],
748)
750sent_messages_counter: Counter = Counter(
751 "couchers_sent_messages_total",
752 "Number of messages sent",
753 labelnames=["gender", "message_type"],
754)
757push_notification_counter: Counter = Counter(
758 "couchers_push_notification_total",
759 "Number of push notification delivery attempts",
760 labelnames=["platform", "outcome"],
761)
762emails_counter: Counter = Counter(
763 "couchers_emails_total",
764 "Number of emails sent",
765)
768# Revenue from successful Stripe charges, in cents, split by source (donation vs merch).
769revenue_cents_counter: Counter = Counter(
770 "couchers_revenue_cents_total",
771 "Revenue from successful Stripe charges, in cents",
772 labelnames=["type"],
773)
776def observe_revenue(revenue_type: str, amount_cents: int) -> None:
777 revenue_cents_counter.labels(revenue_type).inc(amount_cents)
780antibots_assessed_counter: Counter = Counter(
781 "couchers_antibots_assessed_total",
782 "Number of times an antibot assessment is created",
783 labelnames=["action"],
784)
786antibot_score_histogram: Histogram = Histogram(
787 "couchers_antibot_score",
788 "Score of antibot assessments",
789 labelnames=["action"],
790 buckets=tuple(x / 20 for x in range(0, 21)),
791)
793host_request_first_response_histogram: Histogram = Histogram(
794 "couchers_host_request_first_response_seconds",
795 "Response time to host requests",
796 labelnames=["host_gender", "surfer_gender", "response_type"],
797 buckets=(
798 1 * 60, # 1m
799 2 * 60, # 2m
800 5 * 60, # 5m
801 10 * 60, # 10m
802 15 * 60, # 15m
803 30 * 60, # 30m
804 45 * 60, # 45m
805 3_600, # 1h
806 2 * 3_600, # 2h
807 3 * 3_600, # 3h
808 6 * 3_600, # 6h
809 12 * 3_600, # 12h
810 86_400, # 24h
811 2 * 86_400, # 2d
812 5 * 86_400, # 4d
813 602_000, # 1w
814 2 * 602_000, # 2w
815 3 * 602_000, # 3w
816 4 * 602_000, # 4w
817 _INF,
818 ),
819)
820account_age_on_host_request_create_histogram: Histogram = Histogram(
821 "couchers_account_age_on_host_request_create_histogram_seconds",
822 "Age of account sending a host request",
823 labelnames=["surfer_gender", "host_gender"],
824 buckets=(
825 5 * 60, # 5m
826 10 * 60, # 10m
827 15 * 60, # 15m
828 30 * 60, # 30m
829 45 * 60, # 45m
830 3_600, # 1h
831 2 * 3_600, # 2h
832 3 * 3_600, # 3h
833 6 * 3_600, # 6h
834 12 * 3_600, # 12h
835 86_400, # 24h
836 2 * 86_400, # 2d
837 3 * 86_400, # 3d
838 4 * 86_400, # 4d
839 5 * 86_400, # 5d
840 6 * 86_400, # 6d
841 602_000, # 1w
842 2 * 602_000, # 2w
843 3 * 602_000, # 3w
844 4 * 602_000, # 4w
845 5 * 602_000, # 5w
846 10 * 602_000, # 10w
847 25 * 602_000, # 25w
848 52 * 602_000, # 52w
849 104 * 602_000, # 104w
850 _INF,
851 ),
852)
855# =============================================================================
856# Moderation metrics
857# =============================================================================
859# Gauges: Queue lengths
860moderation_queue_length_gauge: Gauge = _make_gauge_from_query(
861 "couchers_moderation_queue_length",
862 "Total number of unresolved items in the moderation queue",
863 select(func.count()).select_from(ModerationQueueItem).where(ModerationQueueItem.resolved_by_log_id.is_(None)),
864)
866moderation_queue_length_by_trigger_gauges: list[Gauge] = [
867 _make_gauge_from_query(
868 f"couchers_moderation_queue_length_{trigger.name.lower()}",
869 f"Number of unresolved items in the moderation queue with trigger {trigger.name}",
870 select(func.count())
871 .select_from(ModerationQueueItem)
872 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
873 .where(ModerationQueueItem.trigger == trigger),
874 )
875 for trigger in ModerationTrigger
876]
878moderation_queue_length_by_object_type_gauges: list[Gauge] = [
879 _make_gauge_from_query(
880 f"couchers_moderation_queue_length_{object_type.name.lower()}",
881 f"Number of unresolved items in the moderation queue for {object_type.name}",
882 select(func.count())
883 .select_from(ModerationQueueItem)
884 .join(ModerationState, ModerationQueueItem.moderation_state_id == ModerationState.id)
885 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
886 .where(ModerationState.object_type == object_type),
887 )
888 for object_type in ModerationObjectType
889]
891# Gauges: Items in each visibility state by object type
892moderation_visibility_gauges: list[Gauge] = [
893 _make_gauge_from_query(
894 f"couchers_moderation_items_{object_type.name.lower()}_{visibility.name.lower()}",
895 f"Number of {object_type.name} items with visibility {visibility.name}",
896 select(func.count())
897 .select_from(ModerationState)
898 .where(ModerationState.object_type == object_type)
899 .where(ModerationState.visibility == visibility),
900 )
901 for object_type in ModerationObjectType
902 for visibility in ModerationVisibility
903]
905# Counters: Moderation actions taken
906moderation_actions_counter: Counter = Counter(
907 "couchers_moderation_actions_total",
908 "Number of moderation actions taken",
909 labelnames=["action", "object_type"],
910)
913def observe_moderation_action(action: ModerationAction, object_type: ModerationObjectType) -> None:
914 moderation_actions_counter.labels(action.name, object_type.name).inc()
917# Counters: Visibility state transitions
918moderation_visibility_transitions_counter: Counter = Counter(
919 "couchers_moderation_visibility_transitions_total",
920 "Number of visibility state transitions",
921 labelnames=["from_visibility", "to_visibility", "object_type"],
922)
925def observe_moderation_visibility_transition(
926 from_visibility: ModerationVisibility, to_visibility: ModerationVisibility, object_type: ModerationObjectType
927) -> None:
928 moderation_visibility_transitions_counter.labels(from_visibility.name, to_visibility.name, object_type.name).inc()
931# Counters: Auto-approved items
932moderation_auto_approved_counter: Counter = Counter(
933 "couchers_moderation_auto_approved_total",
934 "Number of items that were auto-approved",
935)
938# Counters: Queue items created
939moderation_queue_items_created_counter: Counter = Counter(
940 "couchers_moderation_queue_items_created_total",
941 "Number of moderation queue items created",
942 labelnames=["trigger", "object_type"],
943)
946def observe_moderation_queue_item_created(trigger: ModerationTrigger, object_type: ModerationObjectType) -> None:
947 moderation_queue_items_created_counter.labels(trigger.name, object_type.name).inc()
950# Counters: Queue items resolved
951moderation_queue_items_resolved_counter: Counter = Counter(
952 "couchers_moderation_queue_items_resolved_total",
953 "Number of moderation queue items resolved",
954 labelnames=["trigger", "action", "object_type"],
955)
958def observe_moderation_queue_item_resolved(
959 trigger: ModerationTrigger, action: ModerationAction, object_type: ModerationObjectType
960) -> None:
961 moderation_queue_items_resolved_counter.labels(trigger.name, action.name, object_type.name).inc()
964# Histogram: Time to resolve queue items
965moderation_queue_resolution_time_histogram: Histogram = Histogram(
966 "couchers_moderation_queue_resolution_seconds",
967 "Time taken to resolve moderation queue items",
968 labelnames=["trigger", "action", "object_type"],
969 buckets=(
970 0.1,
971 0.25,
972 0.5,
973 1,
974 2.5,
975 5,
976 10,
977 30,
978 60,
979 5 * 60,
980 15 * 60,
981 30 * 60,
982 3_600,
983 2 * 3_600,
984 6 * 3_600,
985 12 * 3_600,
986 86_400,
987 2 * 86_400,
988 3 * 86_400,
989 7 * 86_400,
990 14 * 86_400,
991 30 * 86_400,
992 _INF,
993 ),
994)
997def observe_moderation_queue_resolution_time(
998 trigger: ModerationTrigger, action: ModerationAction, object_type: ModerationObjectType, duration_s: float
999) -> None:
1000 moderation_queue_resolution_time_histogram.labels(trigger.name, action.name, object_type.name).observe(duration_s)
1003nonvisible_user_access_counter: Counter = Counter(
1004 "couchers_nonvisible_user_access_total",
1005 "Number of access events involving nonvisible (banned/shadowed/deleted) users",
1006 labelnames=["access_type", "target_state"],
1007)
1010def observe_nonvisible_user_access(access_type: NonvisibleUserAccessType, target_state: NonvisibleUserState) -> None:
1011 nonvisible_user_access_counter.labels(access_type.name, target_state.name).inc()
1014postcards_sent_counter: Counter = Counter(
1015 "couchers_postcards_sent_total",
1016 "Number of postcards sent via MyPostcard",
1017 labelnames=["country_code"],
1018)
1021# Native app / OTA update metrics. Bucket layout is minute-resolution at the low end (watch an OTA
1022# rolling out), dense around the OTA (~28d) and store (~91d) windows, and sparse past it for stragglers.
1023_NATIVE_AGE_BUCKETS: tuple[float, ...] = (
1024 60,
1025 5 * 60,
1026 15 * 60,
1027 30 * 60,
1028 3_600,
1029 2 * 3_600,
1030 6 * 3_600,
1031 12 * 3_600,
1032 86_400,
1033 2 * 86_400,
1034 3 * 86_400,
1035 5 * 86_400,
1036 7 * 86_400,
1037 10 * 86_400,
1038 14 * 86_400,
1039 21 * 86_400,
1040 28 * 86_400,
1041 35 * 86_400,
1042 45 * 86_400,
1043 60 * 86_400,
1044 75 * 86_400,
1045 91 * 86_400,
1046 120 * 86_400,
1047 150 * 86_400,
1048 180 * 86_400,
1049 270 * 86_400,
1050 365 * 86_400,
1051 730 * 86_400,
1052 _INF,
1053)
1055native_bundle_age_histogram: Histogram = Histogram(
1056 "couchers_native_bundle_age_seconds",
1057 "Age of the OTA bundle reported by the client at CheckNativeStatus, by platform and launch source",
1058 labelnames=["platform", "is_ota_launch"],
1059 buckets=_NATIVE_AGE_BUCKETS,
1060)
1063def observe_native_bundle_age(platform: str, is_ota_launch: bool, age_s: float) -> None:
1064 native_bundle_age_histogram.labels(platform or "unknown", "true" if is_ota_launch else "false").observe(age_s)
1067native_binary_age_histogram: Histogram = Histogram(
1068 "couchers_native_binary_age_seconds",
1069 "Age of the embedded native binary reported by the client at CheckNativeStatus, by platform",
1070 labelnames=["platform"],
1071 buckets=_NATIVE_AGE_BUCKETS,
1072)
1075def observe_native_binary_age(platform: str, age_s: float) -> None:
1076 native_binary_age_histogram.labels(platform or "unknown").observe(age_s)
1079native_update_decisions_counter: Counter = Counter(
1080 "couchers_native_update_decisions_total",
1081 "CheckNativeStatus decisions, by platform / action / severity",
1082 labelnames=["platform", "action", "severity"],
1083)
1086def observe_native_update_decision(platform: str, action: str, severity: str) -> None:
1087 native_update_decisions_counter.labels(platform or "unknown", action, severity).inc()
1090native_banned_bundle_hits_counter: Counter = Counter(
1091 "couchers_native_banned_bundle_hits_total",
1092 "CheckNativeStatus calls from a device running a banned OTA bundle, by platform",
1093 labelnames=["platform"],
1094)
1097def observe_native_banned_bundle_hit(platform: str) -> None:
1098 native_banned_bundle_hits_counter.labels(platform or "unknown").inc()
1101native_ota_manifest_requests_counter: Counter = Counter(
1102 "couchers_native_ota_manifest_requests_total",
1103 "GetNativeUpdateManifest requests, by platform and result (served, no_update, no_match)",
1104 labelnames=["platform", "result"],
1105)
1108def observe_native_ota_manifest_request(platform: str, result: str) -> None:
1109 native_ota_manifest_requests_counter.labels(platform or "unknown", result).inc()
1112# One increment per CheckNativeStatus, labeled by build/bundle identity, to see the live mix of
1113# versions and bundles running in the fleet.
1114native_client_checkins_counter: Counter = Counter(
1115 "couchers_native_client_checkins_total",
1116 "CheckNativeStatus calls, labeled by build/bundle identity",
1117 labelnames=[
1118 "platform",
1119 "is_ota_launch",
1120 "embedded_display_version",
1121 "embedded_runtime_version",
1122 "ota_display_version",
1123 "ota_update_id",
1124 ],
1125)
1128def observe_native_client_checkin(
1129 platform: str,
1130 is_ota_launch: bool,
1131 embedded_display_version: str,
1132 embedded_runtime_version: str,
1133 ota_display_version: str,
1134 ota_update_id: str,
1135) -> None:
1136 native_client_checkins_counter.labels(
1137 platform or "unknown",
1138 "true" if is_ota_launch else "false",
1139 embedded_display_version or "unknown",
1140 embedded_runtime_version or "unknown",
1141 ota_display_version or "none",
1142 ota_update_id or "none",
1143 ).inc()
1146# Recomputed at scrape time via the hacky-gauge mechanism, so it reflects live age. 0 when disabled
1147# or never pulled.
1148def _feature_flags_staleness_seconds() -> float:
1149 return experimentation.seconds_since_last_fetch() or 0.0
1152feature_flags_staleness_gauge: Gauge = Gauge(
1153 "couchers_feature_flags_staleness_seconds",
1154 "Seconds since feature flags were last successfully fetched from GrowthBook",
1155 multiprocess_mode="mostrecent",
1156)
1157_set_hacky_gauges_funcs.append((feature_flags_staleness_gauge, _feature_flags_staleness_seconds))
1160feature_flag_evaluations_counter: Counter = Counter(
1161 "couchers_feature_flag_evaluations_total",
1162 "Number of feature flag evaluations, by flag key, evaluation source, and resolved value",
1163 labelnames=["flag_key", "source", "value"],
1164)
1166_MAX_FLAG_VALUE_LABEL_LEN = 32
1169def _stringify_flag_value(value: Any) -> str:
1170 if isinstance(value, bool):
1171 return "true" if value else "false"
1172 if isinstance(value, (int, float, str)):
1173 s = str(value)
1174 return s if len(s) <= _MAX_FLAG_VALUE_LABEL_LEN else f"<{type(value).__name__}>"
1175 if value is None: 1175 ↛ 1177line 1175 didn't jump to line 1177 because the condition on line 1175 was always true
1176 return "None"
1177 return f"<{type(value).__name__}>"
1180def observe_feature_flag_evaluation(flag_key: str, source: str, value: Any) -> None:
1181 feature_flag_evaluations_counter.labels(flag_key, source, _stringify_flag_value(value)).inc()
1184def create_prometheus_server(port: int) -> Any:
1185 """custom start method to fix problem descrbied in https://github.com/prometheus/client_python/issues/155"""
1187 def app(environ: Any, start_response: Any) -> Any:
1188 # set hacky gauges
1189 for gauge, f in _set_hacky_gauges_funcs:
1190 gauge.set(f())
1191 for gauge, labeled_f in _set_hacky_labeled_gauges_funcs:
1192 labeled_f(gauge)
1193 for multi_f in _set_hacky_multi_gauges_funcs:
1194 multi_f()
1196 data = generate_latest(registry)
1197 start_response("200 OK", [("Content-type", CONTENT_TYPE_LATEST), ("Content-Length", str(len(data)))])
1198 return [data]
1200 httpd = exposition.make_server( # type: ignore[attr-defined]
1201 "", port, app, exposition.ThreadingWSGIServer, handler_class=exposition._SilentHandler
1202 )
1203 t = threading.Thread(target=httpd.serve_forever)
1204 t.daemon = True
1205 t.start()
1206 return httpd