Movses AslanyanandClaude Opus 5.5
920712f32b
- ledgerline.transactions.total (counter: kind, status, currency) and ledgerline.transaction.duration (histogram, seconds: kind) in gateway, the only place that sees the final outcome, failures included. currency is client input, so values outside the seeded set are recorded as "other". - ledgerline.fraud.score (histogram, buckets in 0.1 steps to 1.0) in fraud. - ledgerline.stream.lag (observable gauge: consumer_group) in audit and notifier, read from XINFO GROUPS through a separate sync Redis client, because callbacks run on the scrape thread; the call is untraced so scrapes don't create traces. - ledgerline.db.pool.in_use (observable up-down counter: pool) in every service with a DB engine. The SQLAlchemy instrumentation's own pool metric names a pool by DSN, which merges the normal and starved engines. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
59 lines
2.3 KiB
Python
59 lines
2.3 KiB
Python
"""Observable (polled) instruments shared by more than one service.
|
|
|
|
Their callbacks run on prometheus_client's scrape thread, not on the event
|
|
loop, so they only do thread-safe reads: a pool's checked-out count, and a
|
|
query through a separate synchronous Redis client (the shared asyncio one
|
|
belongs to the loop)."""
|
|
import redis
|
|
from opentelemetry import metrics
|
|
from opentelemetry.instrumentation.utils import suppress_instrumentation
|
|
from opentelemetry.metrics import CallbackOptions, Observation
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
from .settings import settings
|
|
|
|
meter = metrics.get_meter(__name__)
|
|
|
|
|
|
def observe_db_pools(pools: dict[str, AsyncEngine]) -> None:
|
|
"""ledgerline.db.pool.in_use, one series per named pool. The SQLAlchemy
|
|
instrumentation's own db.client.connections.usage names a pool by its
|
|
connection string, which a service's normal and starved engines share, so
|
|
it sums the two into one series."""
|
|
|
|
def callback(_options: CallbackOptions):
|
|
return [
|
|
Observation(engine.sync_engine.pool.checkedout(), {"pool": name})
|
|
for name, engine in pools.items()
|
|
]
|
|
|
|
meter.create_observable_up_down_counter(
|
|
"ledgerline.db.pool.in_use",
|
|
callbacks=[callback],
|
|
unit="{connection}",
|
|
description="Connections currently checked out of the pool",
|
|
)
|
|
|
|
|
|
def observe_stream_lag(consumer_group: str) -> None:
|
|
"""ledgerline.stream.lag for this process's consumer group: stream entries
|
|
not yet delivered to the group (XINFO GROUPS `lag`, Redis 7+)."""
|
|
client = redis.Redis.from_url(settings.redis_url, decode_responses=True, socket_timeout=1)
|
|
|
|
def callback(_options: CallbackOptions):
|
|
# Untraced, or every scrape would be a one-span root trace in Tempo.
|
|
with suppress_instrumentation():
|
|
groups = client.xinfo_groups(settings.stream_name)
|
|
return [
|
|
Observation(group["lag"], {"consumer_group": consumer_group})
|
|
for group in groups
|
|
if group["name"] == consumer_group and group["lag"] is not None
|
|
]
|
|
|
|
meter.create_observable_gauge(
|
|
"ledgerline.stream.lag",
|
|
callbacks=[callback],
|
|
unit="{message}",
|
|
description="Stream entries not yet delivered to the consumer group",
|
|
)
|