Coverage for app/backend/src/couchers/metrics.py: 96%
279 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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.middleware.perf import PerfResult
32from couchers.models import (
33 BackgroundJob,
34 ClientPlatform,
35 Cluster,
36 EventOccurrenceAttendee,
37 HostingStatus,
38 HostRequest,
39 Message,
40 Node,
41 NodeType,
42 NonvisibleUserAccessType,
43 NonvisibleUserState,
44 Reference,
45 User,
46 UserActivity,
47)
48from couchers.models.moderation import (
49 ModerationAction,
50 ModerationObjectType,
51 ModerationQueueItem,
52 ModerationState,
53 ModerationTrigger,
54 ModerationVisibility,
55)
56from couchers.models.uploads import PhotoGalleryItem
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/middleware/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)
693signup_email_changes_counter: Counter = Counter(
694 "couchers_signup_email_changes_total",
695 "Number of times email is changed during signup",
696)
698logins_counter: Counter = Counter(
699 "couchers_logins_total",
700 "Number of logins",
701 labelnames=["gender"],
702)
704password_reset_initiations_counter: Counter = Counter(
705 "couchers_password_reset_initiations_total",
706 "Number of password reset initiations",
707)
708password_reset_completions_counter: Counter = Counter(
709 "couchers_password_reset_completions_total",
710 "Number of password reset completions",
711)
713account_deletion_initiations_counter: Counter = Counter(
714 "couchers_account_deletion_initiations_total",
715 "Number of account deletion initiations",
716 labelnames=["gender"],
717)
718account_deletion_completions_counter: Counter = Counter(
719 "couchers_account_deletion_completions_total",
720 "Number of account deletion completions",
721 labelnames=["gender"],
722)
723account_recoveries_counter: Counter = Counter(
724 "couchers_account_recoveries_total",
725 "Number of account recoveries",
726 labelnames=["gender"],
727)
729strong_verification_initiations_counter: Counter = Counter(
730 "couchers_strong_verification_initiations_total",
731 "Number of strong verification initiations",
732 labelnames=["gender"],
733)
734strong_verification_completions_counter: Counter = Counter(
735 "couchers_strong_verification_completions_total",
736 "Number of strong verification completions",
737)
738strong_verification_data_deletions_counter: Counter = Counter(
739 "couchers_strong_verification_data_deletions_total",
740 "Number of strong verification data deletions",
741 labelnames=["gender"],
742)
744host_requests_sent_counter: Counter = Counter(
745 "couchers_host_requests_total",
746 "Number of host requests sent",
747 labelnames=["from_gender", "to_gender"],
748)
749host_request_responses_counter: Counter = Counter(
750 "couchers_host_requests_responses_total",
751 "Number of responses to host requests",
752 labelnames=["responder_gender", "other_gender", "response_type"],
753)
755sent_messages_counter: Counter = Counter(
756 "couchers_sent_messages_total",
757 "Number of messages sent",
758 labelnames=["gender", "message_type"],
759)
762push_notification_counter: Counter = Counter(
763 "couchers_push_notification_total",
764 "Number of push notification delivery attempts",
765 labelnames=["platform", "outcome"],
766)
767emails_counter: Counter = Counter(
768 "couchers_emails_total",
769 "Number of emails sent",
770)
773# Revenue from successful Stripe charges, in cents, split by source (donation vs merch).
774revenue_cents_counter: Counter = Counter(
775 "couchers_revenue_cents_total",
776 "Revenue from successful Stripe charges, in cents",
777 labelnames=["type"],
778)
781def observe_revenue(revenue_type: str, amount_cents: int) -> None:
782 revenue_cents_counter.labels(revenue_type).inc(amount_cents)
785antibots_assessed_counter: Counter = Counter(
786 "couchers_antibots_assessed_total",
787 "Number of times an antibot assessment is created",
788 labelnames=["action"],
789)
791antibot_score_histogram: Histogram = Histogram(
792 "couchers_antibot_score",
793 "Score of antibot assessments",
794 labelnames=["action"],
795 buckets=tuple(x / 20 for x in range(0, 21)),
796)
798host_request_first_response_histogram: Histogram = Histogram(
799 "couchers_host_request_first_response_seconds",
800 "Response time to host requests",
801 labelnames=["host_gender", "surfer_gender", "response_type"],
802 buckets=(
803 1 * 60, # 1m
804 2 * 60, # 2m
805 5 * 60, # 5m
806 10 * 60, # 10m
807 15 * 60, # 15m
808 30 * 60, # 30m
809 45 * 60, # 45m
810 3_600, # 1h
811 2 * 3_600, # 2h
812 3 * 3_600, # 3h
813 6 * 3_600, # 6h
814 12 * 3_600, # 12h
815 86_400, # 24h
816 2 * 86_400, # 2d
817 5 * 86_400, # 4d
818 602_000, # 1w
819 2 * 602_000, # 2w
820 3 * 602_000, # 3w
821 4 * 602_000, # 4w
822 _INF,
823 ),
824)
825account_age_on_host_request_create_histogram: Histogram = Histogram(
826 "couchers_account_age_on_host_request_create_histogram_seconds",
827 "Age of account sending a host request",
828 labelnames=["surfer_gender", "host_gender"],
829 buckets=(
830 5 * 60, # 5m
831 10 * 60, # 10m
832 15 * 60, # 15m
833 30 * 60, # 30m
834 45 * 60, # 45m
835 3_600, # 1h
836 2 * 3_600, # 2h
837 3 * 3_600, # 3h
838 6 * 3_600, # 6h
839 12 * 3_600, # 12h
840 86_400, # 24h
841 2 * 86_400, # 2d
842 3 * 86_400, # 3d
843 4 * 86_400, # 4d
844 5 * 86_400, # 5d
845 6 * 86_400, # 6d
846 602_000, # 1w
847 2 * 602_000, # 2w
848 3 * 602_000, # 3w
849 4 * 602_000, # 4w
850 5 * 602_000, # 5w
851 10 * 602_000, # 10w
852 25 * 602_000, # 25w
853 52 * 602_000, # 52w
854 104 * 602_000, # 104w
855 _INF,
856 ),
857)
860# =============================================================================
861# Moderation metrics
862# =============================================================================
864# Gauges: Queue lengths
865moderation_queue_length_gauge: Gauge = _make_gauge_from_query(
866 "couchers_moderation_queue_length",
867 "Total number of unresolved items in the moderation queue",
868 select(func.count()).select_from(ModerationQueueItem).where(ModerationQueueItem.resolved_by_log_id.is_(None)),
869)
871moderation_queue_length_by_trigger_gauges: list[Gauge] = [
872 _make_gauge_from_query(
873 f"couchers_moderation_queue_length_{trigger.name.lower()}",
874 f"Number of unresolved items in the moderation queue with trigger {trigger.name}",
875 select(func.count())
876 .select_from(ModerationQueueItem)
877 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
878 .where(ModerationQueueItem.trigger == trigger),
879 )
880 for trigger in ModerationTrigger
881]
883moderation_queue_length_by_object_type_gauges: list[Gauge] = [
884 _make_gauge_from_query(
885 f"couchers_moderation_queue_length_{object_type.name.lower()}",
886 f"Number of unresolved items in the moderation queue for {object_type.name}",
887 select(func.count())
888 .select_from(ModerationQueueItem)
889 .join(ModerationState, ModerationQueueItem.moderation_state_id == ModerationState.id)
890 .where(ModerationQueueItem.resolved_by_log_id.is_(None))
891 .where(ModerationState.object_type == object_type),
892 )
893 for object_type in ModerationObjectType
894]
896# Gauges: Items in each visibility state by object type
897moderation_visibility_gauges: list[Gauge] = [
898 _make_gauge_from_query(
899 f"couchers_moderation_items_{object_type.name.lower()}_{visibility.name.lower()}",
900 f"Number of {object_type.name} items with visibility {visibility.name}",
901 select(func.count())
902 .select_from(ModerationState)
903 .where(ModerationState.object_type == object_type)
904 .where(ModerationState.visibility == visibility),
905 )
906 for object_type in ModerationObjectType
907 for visibility in ModerationVisibility
908]
910# Counters: Moderation actions taken
911moderation_actions_counter: Counter = Counter(
912 "couchers_moderation_actions_total",
913 "Number of moderation actions taken",
914 labelnames=["action", "object_type"],
915)
918def observe_moderation_action(action: ModerationAction, object_type: ModerationObjectType) -> None:
919 moderation_actions_counter.labels(action.name, object_type.name).inc()
922# Counters: Visibility state transitions
923moderation_visibility_transitions_counter: Counter = Counter(
924 "couchers_moderation_visibility_transitions_total",
925 "Number of visibility state transitions",
926 labelnames=["from_visibility", "to_visibility", "object_type"],
927)
930def observe_moderation_visibility_transition(
931 from_visibility: ModerationVisibility, to_visibility: ModerationVisibility, object_type: ModerationObjectType
932) -> None:
933 moderation_visibility_transitions_counter.labels(from_visibility.name, to_visibility.name, object_type.name).inc()
936# Counters: Auto-approved items
937moderation_auto_approved_counter: Counter = Counter(
938 "couchers_moderation_auto_approved_total",
939 "Number of items that were auto-approved",
940)
943# Counters: Queue items created
944moderation_queue_items_created_counter: Counter = Counter(
945 "couchers_moderation_queue_items_created_total",
946 "Number of moderation queue items created",
947 labelnames=["trigger", "object_type"],
948)
951def observe_moderation_queue_item_created(trigger: ModerationTrigger, object_type: ModerationObjectType) -> None:
952 moderation_queue_items_created_counter.labels(trigger.name, object_type.name).inc()
955# Counters: Queue items resolved
956moderation_queue_items_resolved_counter: Counter = Counter(
957 "couchers_moderation_queue_items_resolved_total",
958 "Number of moderation queue items resolved",
959 labelnames=["trigger", "action", "object_type"],
960)
963def observe_moderation_queue_item_resolved(
964 trigger: ModerationTrigger, action: ModerationAction, object_type: ModerationObjectType
965) -> None:
966 moderation_queue_items_resolved_counter.labels(trigger.name, action.name, object_type.name).inc()
969# Histogram: Time to resolve queue items
970moderation_queue_resolution_time_histogram: Histogram = Histogram(
971 "couchers_moderation_queue_resolution_seconds",
972 "Time taken to resolve moderation queue items",
973 labelnames=["trigger", "action", "object_type"],
974 buckets=(
975 0.1,
976 0.25,
977 0.5,
978 1,
979 2.5,
980 5,
981 10,
982 30,
983 60,
984 5 * 60,
985 15 * 60,
986 30 * 60,
987 3_600,
988 2 * 3_600,
989 6 * 3_600,
990 12 * 3_600,
991 86_400,
992 2 * 86_400,
993 3 * 86_400,
994 7 * 86_400,
995 14 * 86_400,
996 30 * 86_400,
997 _INF,
998 ),
999)
1002def observe_moderation_queue_resolution_time(
1003 trigger: ModerationTrigger, action: ModerationAction, object_type: ModerationObjectType, duration_s: float
1004) -> None:
1005 moderation_queue_resolution_time_histogram.labels(trigger.name, action.name, object_type.name).observe(duration_s)
1008nonvisible_user_access_counter: Counter = Counter(
1009 "couchers_nonvisible_user_access_total",
1010 "Number of access events involving nonvisible (banned/shadowed/deleted) users",
1011 labelnames=["access_type", "target_state"],
1012)
1015def observe_nonvisible_user_access(access_type: NonvisibleUserAccessType, target_state: NonvisibleUserState) -> None:
1016 nonvisible_user_access_counter.labels(access_type.name, target_state.name).inc()
1019postcards_sent_counter: Counter = Counter(
1020 "couchers_postcards_sent_total",
1021 "Number of postcards sent via MyPostcard",
1022 labelnames=["country_code"],
1023)
1026# Native app / OTA update metrics. Bucket layout is minute-resolution at the low end (watch an OTA
1027# rolling out), dense around the OTA (~28d) and store (~91d) windows, and sparse past it for stragglers.
1028_NATIVE_AGE_BUCKETS: tuple[float, ...] = (
1029 60,
1030 5 * 60,
1031 15 * 60,
1032 30 * 60,
1033 3_600,
1034 2 * 3_600,
1035 6 * 3_600,
1036 12 * 3_600,
1037 86_400,
1038 2 * 86_400,
1039 3 * 86_400,
1040 5 * 86_400,
1041 7 * 86_400,
1042 10 * 86_400,
1043 14 * 86_400,
1044 21 * 86_400,
1045 28 * 86_400,
1046 35 * 86_400,
1047 45 * 86_400,
1048 60 * 86_400,
1049 75 * 86_400,
1050 91 * 86_400,
1051 120 * 86_400,
1052 150 * 86_400,
1053 180 * 86_400,
1054 270 * 86_400,
1055 365 * 86_400,
1056 730 * 86_400,
1057 _INF,
1058)
1060native_bundle_age_histogram: Histogram = Histogram(
1061 "couchers_native_bundle_age_seconds",
1062 "Age of the OTA bundle reported by the client at CheckNativeStatus, by platform and launch source",
1063 labelnames=["platform", "is_ota_launch"],
1064 buckets=_NATIVE_AGE_BUCKETS,
1065)
1068def observe_native_bundle_age(platform: str, is_ota_launch: bool, age_s: float) -> None:
1069 native_bundle_age_histogram.labels(platform or "unknown", "true" if is_ota_launch else "false").observe(age_s)
1072native_binary_age_histogram: Histogram = Histogram(
1073 "couchers_native_binary_age_seconds",
1074 "Age of the embedded native binary reported by the client at CheckNativeStatus, by platform",
1075 labelnames=["platform"],
1076 buckets=_NATIVE_AGE_BUCKETS,
1077)
1080def observe_native_binary_age(platform: str, age_s: float) -> None:
1081 native_binary_age_histogram.labels(platform or "unknown").observe(age_s)
1084native_update_decisions_counter: Counter = Counter(
1085 "couchers_native_update_decisions_total",
1086 "CheckNativeStatus decisions, by platform / action / severity",
1087 labelnames=["platform", "action", "severity"],
1088)
1091def observe_native_update_decision(platform: str, action: str, severity: str) -> None:
1092 native_update_decisions_counter.labels(platform or "unknown", action, severity).inc()
1095native_banned_bundle_hits_counter: Counter = Counter(
1096 "couchers_native_banned_bundle_hits_total",
1097 "CheckNativeStatus calls from a device running a banned OTA bundle, by platform",
1098 labelnames=["platform"],
1099)
1102def observe_native_banned_bundle_hit(platform: str) -> None:
1103 native_banned_bundle_hits_counter.labels(platform or "unknown").inc()
1106native_ota_manifest_requests_counter: Counter = Counter(
1107 "couchers_native_ota_manifest_requests_total",
1108 "GetNativeUpdateManifest requests, by platform and result (served, no_update, no_match)",
1109 labelnames=["platform", "result"],
1110)
1113def observe_native_ota_manifest_request(platform: str, result: str) -> None:
1114 native_ota_manifest_requests_counter.labels(platform or "unknown", result).inc()
1117# One increment per CheckNativeStatus, labeled by build/bundle identity, to see the live mix of
1118# versions and bundles running in the fleet.
1119native_client_checkins_counter: Counter = Counter(
1120 "couchers_native_client_checkins_total",
1121 "CheckNativeStatus calls, labeled by build/bundle identity",
1122 labelnames=[
1123 "platform",
1124 "is_ota_launch",
1125 "embedded_display_version",
1126 "embedded_runtime_version",
1127 "ota_display_version",
1128 "ota_update_id",
1129 ],
1130)
1133def observe_native_client_checkin(
1134 platform: str,
1135 is_ota_launch: bool,
1136 embedded_display_version: str,
1137 embedded_runtime_version: str,
1138 ota_display_version: str,
1139 ota_update_id: str,
1140) -> None:
1141 native_client_checkins_counter.labels(
1142 platform or "unknown",
1143 "true" if is_ota_launch else "false",
1144 embedded_display_version or "unknown",
1145 embedded_runtime_version or "unknown",
1146 ota_display_version or "none",
1147 ota_update_id or "none",
1148 ).inc()
1151# Recomputed at scrape time via the hacky-gauge mechanism, so it reflects live age. 0 when disabled
1152# or never pulled.
1153def _feature_flags_staleness_seconds() -> float:
1154 return experimentation.seconds_since_last_fetch() or 0.0
1157feature_flags_staleness_gauge: Gauge = Gauge(
1158 "couchers_feature_flags_staleness_seconds",
1159 "Seconds since feature flags were last successfully fetched from GrowthBook",
1160 multiprocess_mode="mostrecent",
1161)
1162_set_hacky_gauges_funcs.append((feature_flags_staleness_gauge, _feature_flags_staleness_seconds))
1165feature_flag_evaluations_counter: Counter = Counter(
1166 "couchers_feature_flag_evaluations_total",
1167 "Number of feature flag evaluations, by flag key, evaluation source, and resolved value",
1168 labelnames=["flag_key", "source", "value"],
1169)
1171_MAX_FLAG_VALUE_LABEL_LEN = 32
1174def _stringify_flag_value(value: Any) -> str:
1175 if isinstance(value, bool):
1176 return "true" if value else "false"
1177 if isinstance(value, (int, float, str)):
1178 s = str(value)
1179 return s if len(s) <= _MAX_FLAG_VALUE_LABEL_LEN else f"<{type(value).__name__}>"
1180 if value is None: 1180 ↛ 1182line 1180 didn't jump to line 1182 because the condition on line 1180 was always true
1181 return "None"
1182 return f"<{type(value).__name__}>"
1185def observe_feature_flag_evaluation(flag_key: str, source: str, value: Any) -> None:
1186 feature_flag_evaluations_counter.labels(flag_key, source, _stringify_flag_value(value)).inc()
1189# =============================================================================
1190# Rate limiting metrics (see couchers/middleware/ratelimit.py)
1191# =============================================================================
1193# "shadowed" = a limit tripped but enforcement is off; "failed_open" = the store was unreachable so nothing
1194# could be counted. Both let the request through.
1195rate_limit_checks_counter: Counter = Counter(
1196 "couchers_rate_limit_checks_total",
1197 "Rate limit checks, by method and decision (allowed, shadowed, blocked, failed_open)",
1198 labelnames=["method", "decision"],
1199)
1202def observe_rate_limit_check(method: str, decision: str) -> None:
1203 rate_limit_checks_counter.labels(method, decision).inc()
1206# one increment per counter that tripped, so a single request can trip several
1207rate_limit_trips_counter: Counter = Counter(
1208 "couchers_rate_limit_trips_total",
1209 "Rate limit counters that tripped, by method, scope, dimension, and whether enforced",
1210 labelnames=["method", "scope", "dimension", "enforced"],
1211)
1214def observe_rate_limit_trip(method: str, scope: str, dimension: str, enforced: bool) -> None:
1215 rate_limit_trips_counter.labels(method, scope, dimension, "true" if enforced else "false").inc()
1218# unlabelled by method: the work is the same whichever method is being counted, and a label would multiply
1219# the series by the whole API surface
1220rate_limit_duration_histogram: Histogram = Histogram(
1221 "couchers_rate_limit_duration_seconds",
1222 "Time spent in the rate limit check",
1223 buckets=MACHINE_DURATION_SECONDS,
1224)
1227def observe_rate_limit_duration(duration_s: float) -> None:
1228 rate_limit_duration_histogram.observe(duration_s)
1231rate_limit_store_errors_counter: Counter = Counter(
1232 "couchers_rate_limit_store_errors_total",
1233 "Rate limit counter-store errors, which fail open, by exception type",
1234 labelnames=["exception"],
1235)
1238def observe_rate_limit_store_error(exception_type: str) -> None:
1239 rate_limit_store_errors_counter.labels(exception_type).inc()
1242def create_prometheus_server(port: int) -> Any:
1243 """custom start method to fix problem descrbied in https://github.com/prometheus/client_python/issues/155"""
1245 def app(environ: Any, start_response: Any) -> Any:
1246 # set hacky gauges
1247 for gauge, f in _set_hacky_gauges_funcs:
1248 gauge.set(f())
1249 for gauge, labeled_f in _set_hacky_labeled_gauges_funcs:
1250 labeled_f(gauge)
1251 for multi_f in _set_hacky_multi_gauges_funcs:
1252 multi_f()
1254 data = generate_latest(registry)
1255 start_response("200 OK", [("Content-type", CONTENT_TYPE_LATEST), ("Content-Length", str(len(data)))])
1256 return [data]
1258 httpd = exposition.make_server( # type: ignore[attr-defined]
1259 "", port, app, exposition.ThreadingWSGIServer, handler_class=exposition._SilentHandler
1260 )
1261 t = threading.Thread(target=httpd.serve_forever)
1262 t.daemon = True
1263 t.start()
1264 return httpd