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