Vendor-neutral observability bootstrap.
The library never imports vendor SDKs (Sentry, Datadog, PostHog, App Insights).
Instead, it emits OpenTelemetry spans and metrics and lets operators wire any
OTLP-compatible backend by configuring environment variables (or by passing an
already-configured TracerProvider / MeterProvider).
Logs are NOT an OTel signal here: there is no LoggerProvider and no
LoggingHandler anywhere in the library. :func:setup_logging configures
structlog over the stdlib logging root logger, so log lines reach a
telemetry backend only if the operator has attached a handler to that root
logger themselves -- which is what configure_azure_monitor() does. That is
incidental wiring, not an emission path this library owns, and it is why
exception text has to be scrubbed inside the processor chain rather than at an
exporter (see _redact_exc).
Common deployment shapes:
- Datadog Agent — accepts OTLP on
localhost:4317. Set
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317.
- Sentry — Sentry Spotlight / Sentry OTel ingest, ditto.
- App Insights — set
OTEL_EXPORTER_OTLP_ENDPOINT to the Azure Monitor
connection string-derived OTLP URL (typically via the App Insights agent).
- PostHog — currently via PostHog Cloud OTLP endpoint, same env var.
For error reporting that doesn't fit OTel exception events (e.g., DLQ routing
to Sentry), users implement the ErrorReporter Protocol
as a DI provider — vendor-neutral and added with the observability surface.
The library depends only on opentelemetry-api at runtime. Operators
who want to configure providers programmatically (or use the in-process
testing utilities in taskq.testing) install the [otel] extra::
pip install "taskq-py[otel]"
which pulls in opentelemetry-sdk and opentelemetry-exporter-otlp.
For Prometheus scrapes, use the [prometheus] extra instead.
Semconv compliance: the library uses spec-compliant messaging semconv
attribute names (messaging.operation.type=publish,
messaging.operation.type=process, messaging.consumer.group.name, etc.)
so that operators who set OTEL_SEMCONV_STABILITY_OPT_IN=messaging get
consistent behavior. No runtime conditional branching on this env var is
needed — the attribute values are correct by construction.
INSTRUMENTATION_NAME
module-attribute
INSTRUMENTATION_NAME: str = 'taskq'
__all__
module-attribute
__all__ = [
"INSTRUMENTATION_NAME",
"ConsumedOutcome",
"ErrorReporter",
"ErrorReporterType",
"NullErrorReporter",
"bind_job_context",
"get_logger",
"get_meter",
"get_tracer",
"invoke_error_reporter",
"log_cancel_phase_change",
"log_state_change",
"record_archived_jobs",
"record_backpressure_error",
"record_cancel_requested",
"record_capacity_refresh_failure",
"record_consumed_message",
"record_cron_failure",
"record_cron_lock_contention",
"record_deadline_exceeded_swept",
"record_dispatch_duration",
"record_election_attempt",
"record_error_reporter_failure",
"record_exception_safe",
"record_expired_archive_jobs",
"record_heartbeat_miss",
"record_lock_expires_in_seconds",
"record_process_duration",
"record_progress_publish_failure",
"record_pruned_jobs",
"record_published_message",
"record_ratelimit_refund_failure",
"redact_payload",
"safe_exception_message",
"safe_start_span",
"set_exception_message_max_chars",
"set_exception_redaction_enabled",
"set_otel_enabled",
"setup_logging",
"update_disabled_schedules_count",
"update_heartbeat_consecutive_failures",
"update_queue_depth_cache",
"update_reservation_slots_cache",
"update_stranded_jobs_cache",
]
ConsumedOutcome
ConsumedOutcome = Literal[
"succeeded", "failed", "cancelled", "abandoned"
]
ErrorReporterType
ErrorReporterType = ErrorReporter
Structural type alias for DI registration and parameter annotations.
Use :class:ErrorReporter directly in most contexts; this alias is provided
for register_value(ErrorReporterType, ...) calls where a distinct type
object is needed for the registry key.
ErrorReporter
Bases: Protocol
Vendor-neutral hook for routing terminal job failures to external systems.
Implementations capture the error and job row, then forward to a
vendor-specific backend (Sentry, Datadog, a DLQ, etc.). The library
calls :meth:report when a job reaches a terminal failure state —
either because retries were exhausted or because the error was
non-retryable.
The call is wrapped in a try/except by :func:invoke_error_reporter;
a failing reporter never crashes the worker. Reporter failures are
counted on the taskq.error_reporter.failures counter with a
reporter_type attribute.
Register an :class:ErrorReporter instance as a DI provider
(registry.register_value(ErrorReporter, Scope.PROCESS, my_reporter))
or pass it directly to the worker bootstrap.
report
async
report(job: JobRow, exception: BaseException) -> None
Source code in src/taskq/obs/error_reporter.py
| async def report(self, job: JobRow, exception: BaseException) -> None: ...
|
NullErrorReporter
Default no-op :class:ErrorReporter — silently drops all reports.
Used when no vendor-specific error routing is configured. Instances
are stateless and safe to share.
report
async
report(job: JobRow, exception: BaseException) -> None
Source code in src/taskq/obs/error_reporter.py
| async def report(self, job: JobRow, exception: BaseException) -> None:
return None
|
get_meter
Return the library's meter. Honors any globally-configured provider.
Source code in src/taskq/obs/_otel.py
| def get_meter() -> Meter:
"""Return the library's meter. Honors any globally-configured provider."""
return metrics.get_meter(INSTRUMENTATION_NAME, _version())
|
get_tracer
Return the library's tracer. Honors any globally-configured provider.
Source code in src/taskq/obs/_otel.py
| def get_tracer() -> Tracer:
"""Return the library's tracer. Honors any globally-configured provider."""
return trace.get_tracer(INSTRUMENTATION_NAME, _version())
|
record_archived_jobs
record_archived_jobs(status: str, count: int = 1) -> None
Bump the archived.jobs counter.
Called alongside record_pruned_jobs at the prune sweep call site in leader.py.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_archived_jobs(status: str, count: int = 1) -> None:
"""Bump the archived.jobs counter.
Called alongside record_pruned_jobs at the prune sweep call site in leader.py.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_archived_jobs.add(count, {"status": status})
|
record_backpressure_error
record_backpressure_error(
actor: str, *, kind: str = "max_pending"
) -> None
Bump the backpressure.errors counter.
Unconditional (not gated by _otel_enabled): backpressure errors are
safety-critical signals that must be counted even when OTel is disabled,
so operators always have visibility into enqueue rejections.
Source code in src/taskq/obs/_otel.py
| def record_backpressure_error(actor: str, *, kind: str = "max_pending") -> None:
"""Bump the backpressure.errors counter.
Unconditional (not gated by ``_otel_enabled``): backpressure errors are
safety-critical signals that must be counted even when OTel is disabled,
so operators always have visibility into enqueue rejections.
"""
try:
_backpressure_errors.add(1, {"actor": actor, "kind": kind})
except Exception:
_log.warning("otel-metric-record-failed", instrument_name="taskq.backpressure.errors")
|
record_cancel_requested
record_cancel_requested() -> None
Bump the cancel-requested counter.
This counter is unconditional:
incremented once per JobsClient.cancel() call regardless of
cancellation_initiated outcome.
Source code in src/taskq/obs/_otel.py
| def record_cancel_requested() -> None:
"""Bump the cancel-requested counter.
This counter is unconditional:
incremented once per ``JobsClient.cancel()`` call regardless of
``cancellation_initiated`` outcome.
"""
try:
_cancellation_requested.add(1)
except Exception:
_log.warning("otel-metric-record-failed", instrument_name="taskq.cancellation.requested")
|
record_capacity_refresh_failure
record_capacity_refresh_failure(
*, has_snapshot: bool
) -> None
Count a failed capacity-cache refresh.
The cache fails OPEN by design -- it keeps the last snapshot, or falls back
to the @actor literal, and stamps refreshed_at so a sick backend is
not re-queried on every enqueue. That reasoning is sound; the problem was
that a load-shedding gate which relaxes precisely when the backend is
degraded had no signal an operator could alert on, only a warning log.
has_snapshot=False is the materially worse case: the first refresh at
process start failed, so there is no stored data at all and every enqueue
enforces the code literal rather than the operator's tightened
max_pending -- for a full TTL at a time, indefinitely while the backend
stays sick.
Unconditional (not gated by _otel_enabled) for the same reason as
record_backpressure_error: this is a safety-critical signal.
Source code in src/taskq/obs/_otel.py
| def record_capacity_refresh_failure(*, has_snapshot: bool) -> None:
"""Count a failed capacity-cache refresh.
The cache fails OPEN by design -- it keeps the last snapshot, or falls back
to the ``@actor`` literal, and stamps ``refreshed_at`` so a sick backend is
not re-queried on every enqueue. That reasoning is sound; the problem was
that a load-shedding gate which relaxes precisely when the backend is
degraded had no signal an operator could alert on, only a warning log.
``has_snapshot=False`` is the materially worse case: the first refresh at
process start failed, so there is no stored data at all and every enqueue
enforces the code literal rather than the operator's tightened
``max_pending`` -- for a full TTL at a time, indefinitely while the backend
stays sick.
Unconditional (not gated by ``_otel_enabled``) for the same reason as
``record_backpressure_error``: this is a safety-critical signal.
"""
try:
_capacity_refresh_failures.add(
1, {"degraded": "stale_snapshot" if has_snapshot else "no_snapshot"}
)
except Exception:
_log.warning(
"otel-metric-record-failed",
instrument_name="taskq.backpressure.capacity_refresh_failures",
)
|
record_consumed_message
record_consumed_message(
actor: str, queue: str, *, outcome: ConsumedOutcome
) -> None
Bump the consumed-messages counter.
Called after job completion, outside the CONSUMER span body,
to ensure sampling independence.
Respects _otel_enabled — no-op when False.
outcome is constrained to the semconv-specified valid set
{succeeded, failed, cancelled, abandoned}.
The consumer-path AttemptOutcome includes "scheduled" for
snooze/retry/reservation-denial; callers must map that to
"abandoned" before calling (the consumer released the job back
to the queue without completing it).
Source code in src/taskq/obs/_otel.py
| def record_consumed_message(actor: str, queue: str, *, outcome: ConsumedOutcome) -> None:
"""Bump the consumed-messages counter.
Called after job completion, outside the CONSUMER span body,
to ensure sampling independence.
Respects ``_otel_enabled`` — no-op when False.
``outcome`` is constrained to the semconv-specified valid set
``{succeeded, failed, cancelled, abandoned}``.
The consumer-path ``AttemptOutcome`` includes ``"scheduled"`` for
snooze/retry/reservation-denial; callers must map that to
``"abandoned"`` before calling (the consumer released the job back
to the queue without completing it).
"""
if not _otel_enabled:
return
_consumed_messages.add(1, {"actor": actor, "queue": queue, "outcome": outcome})
|
record_cron_failure
record_cron_failure(schedule_id: str, delta: int) -> None
Record a cron failure delta on the UpDownCounter.
On failure, callers add +1 per failure. On success, callers add
-current_count for that schedule to reset the counter to zero —
a simple add(-1) would leave a non-zero cumulative value if
there were multiple consecutive failures.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_cron_failure(schedule_id: str, delta: int) -> None:
"""Record a cron failure delta on the UpDownCounter.
On failure, callers add ``+1`` per failure. On success, callers add
``-current_count`` for that schedule to reset the counter to zero —
a simple ``add(-1)`` would leave a non-zero cumulative value if
there were multiple consecutive failures.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_cron_consecutive_failures.add(delta, {"schedule_id": schedule_id})
|
record_cron_lock_contention
record_cron_lock_contention(worker_id: str) -> None
Count a cron tick skipped because the advisory lock was held.
A steady low rate is the benign leader-handover overlap. A rate equal to
the tick rate, sustained, means cron is not running anywhere: the lock is
transaction-scoped and releases on COMMIT/ROLLBACK, which never happens if
the holding session was partitioned without a FIN. Before this counter the
two were indistinguishable, because the contended branch returned in
silence.
Respects _otel_enabled -- no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_cron_lock_contention(worker_id: str) -> None:
"""Count a cron tick skipped because the advisory lock was held.
A steady low rate is the benign leader-handover overlap. A rate equal to
the tick rate, sustained, means cron is not running anywhere: the lock is
transaction-scoped and releases on COMMIT/ROLLBACK, which never happens if
the holding session was partitioned without a FIN. Before this counter the
two were indistinguishable, because the contended branch returned in
silence.
Respects ``_otel_enabled`` -- no-op when False.
"""
if not _otel_enabled:
return
del worker_id # Why: not a dimension -- see the cardinality note above.
_cron_lock_contention.add(1)
|
record_deadline_exceeded_swept
record_deadline_exceeded_swept(
actor: str, count: int = 1
) -> None
Bump the deadline-exceeded sweep counter.
Unconditional (not gated by _otel_enabled): deadline-exceeded sweeps
indicate jobs that violated their execution budget — a correctness signal
that must be counted even when OTel is disabled, so operators always have
visibility into sweep activity.
Source code in src/taskq/obs/_otel.py
| def record_deadline_exceeded_swept(actor: str, count: int = 1) -> None:
"""Bump the deadline-exceeded sweep counter.
Unconditional (not gated by ``_otel_enabled``): deadline-exceeded sweeps
indicate jobs that violated their execution budget — a correctness signal
that must be counted even when OTel is disabled, so operators always have
visibility into sweep activity.
"""
try:
_deadline_exceeded_sweep_jobs_failed.add(count, {"actor": actor})
except Exception:
_log.warning(
"otel-metric-record-failed", instrument_name="taskq.deadline_exceeded_sweep.jobs_failed"
)
|
record_dispatch_duration
record_dispatch_duration(
queue: str, elapsed: float
) -> None
Record dispatch query latency on the histogram.
Called outside the dispatch span body for sampling independence.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_dispatch_duration(queue: str, elapsed: float) -> None:
"""Record dispatch query latency on the histogram.
Called outside the ``dispatch`` span body for sampling independence.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_dispatch_duration.record(elapsed, {"queue": queue})
|
record_election_attempt
record_election_attempt(
worker_id: str, *, won: bool
) -> None
Record a leader election attempt.
Always increments election_attempts; increments election_failures
only when the attempt did not win the lock.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_election_attempt(worker_id: str, *, won: bool) -> None:
"""Record a leader election attempt.
Always increments ``election_attempts``; increments ``election_failures``
only when the attempt did not win the lock.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
del worker_id # Why: not a dimension -- see the cardinality note above.
_leader_election_attempts.add(1)
if not won:
_leader_election_failures.add(1)
|
record_error_reporter_failure
record_error_reporter_failure(reporter_type: str) -> None
Bump the error_reporter.failures counter.
Called at the error-reporter catch site when report() raises.
reporter_type is the exception-safe class name of the reporter
instance (bounded cardinality — one per registered implementation).
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_error_reporter_failure(reporter_type: str) -> None:
"""Bump the error_reporter.failures counter.
Called at the error-reporter catch site when ``report()`` raises.
``reporter_type`` is the exception-safe class name of the reporter
instance (bounded cardinality — one per registered implementation).
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_error_reporter_failures.add(1, {"reporter_type": reporter_type})
|
record_expired_archive_jobs
record_expired_archive_jobs(
status: str, count: int = 1
) -> None
Bump the expired_archive.jobs counter.
Called at the archive expiry sweep call site in leader.py.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_expired_archive_jobs(status: str, count: int = 1) -> None:
"""Bump the expired_archive.jobs counter.
Called at the archive expiry sweep call site in leader.py.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_expired_archive_jobs.add(count, {"status": status})
|
record_heartbeat_miss
record_heartbeat_miss(worker_id: str) -> None
Bump the heartbeat.misses counter.
Called in heartbeat.py on each heartbeat renewal failure.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_heartbeat_miss(worker_id: str) -> None:
"""Bump the heartbeat.misses counter.
Called in heartbeat.py on each heartbeat renewal failure.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
del worker_id # Why: not a dimension -- see the cardinality note above.
_heartbeat_misses.add(1)
|
record_lock_expires_in_seconds
record_lock_expires_in_seconds(
worker_id: str, remaining_ttl: float
) -> None
Record remaining lock TTL on the histogram.
Called in heartbeat.py at each successful heartbeat renewal.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_lock_expires_in_seconds(worker_id: str, remaining_ttl: float) -> None:
"""Record remaining lock TTL on the histogram.
Called in heartbeat.py at each successful heartbeat renewal.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
del worker_id # Why: not a dimension -- see the cardinality note above.
_lock_expires_in_seconds.record(remaining_ttl)
|
record_process_duration
record_process_duration(
actor: str, queue: str, elapsed: float
) -> None
Record job execution duration on the histogram.
Called outside the CONSUMER span body for sampling independence.
Respects _otel_enabled — no-op when False.
Custom buckets are the operator's responsibility via SDK Views.
Source code in src/taskq/obs/_otel.py
| def record_process_duration(actor: str, queue: str, elapsed: float) -> None:
"""Record job execution duration on the histogram.
Called outside the CONSUMER span body for sampling independence.
Respects ``_otel_enabled`` — no-op when False.
Custom buckets are the operator's responsibility via SDK Views.
"""
if not _otel_enabled:
return
_process_duration.record(elapsed, {"actor": actor, "queue": queue})
|
record_progress_publish_failure
record_progress_publish_failure(
channel: str, error_type: str
) -> None
Bump the progress.publish_failures counter.
channel must be 'per_job' or 'global' — bounded cardinality.
error_type is the exception class name (e.g. 'ResponseError').
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_progress_publish_failure(channel: str, error_type: str) -> None:
"""Bump the progress.publish_failures counter.
``channel`` must be ``'per_job'`` or ``'global'`` — bounded cardinality.
``error_type`` is the exception class name (e.g. ``'ResponseError'``).
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_progress_publish_failures.add(1, {"channel": channel, "error_type": error_type})
|
record_pruned_jobs
record_pruned_jobs(
actor: str, status: str, count: int = 1
) -> None
Bump the pruned.jobs counter.
Called at the prune sweep call site in leader.py.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_pruned_jobs(actor: str, status: str, count: int = 1) -> None:
"""Bump the pruned.jobs counter.
Called at the prune sweep call site in leader.py.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_pruned_jobs.add(count, {"actor": actor, "status": status})
|
record_published_message
record_published_message(actor: str, queue: str) -> None
Bump the published-messages counter.
Called after successful enqueue, outside the PRODUCER span body,
to ensure sampling independence.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_published_message(actor: str, queue: str) -> None:
"""Bump the published-messages counter.
Called after successful enqueue, outside the PRODUCER span body,
to ensure sampling independence.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_published_messages.add(1, {"actor": actor, "queue": queue})
|
record_ratelimit_refund_failure
record_ratelimit_refund_failure(
bucket: str, backend: str
) -> None
Bump the ratelimit.refund_failures counter.
Called at the rate-limit refund failure catch site.
Respects _otel_enabled — no-op when False.
Source code in src/taskq/obs/_otel.py
| def record_ratelimit_refund_failure(bucket: str, backend: str) -> None:
"""Bump the ratelimit.refund_failures counter.
Called at the rate-limit refund failure catch site.
Respects ``_otel_enabled`` — no-op when False.
"""
if not _otel_enabled:
return
_ratelimit_refund_failures.add(1, {"bucket": bucket, "backend": backend})
|
safe_start_span
safe_start_span(
name: str,
*,
kind: SpanKind | None = None,
attributes: Attributes = None,
links: Sequence[Link] | None = None,
new_root: bool = False,
) -> Generator[Span, None, None]
Start a span safely — never propagates exceptions from OTel API calls.
Checks _otel_enabled first; when False, yields a no-op
NonRecordingSpan. When True, delegates to
get_tracer().start_as_current_span with a try/except around
span creation only. Exceptions from code inside the with block
propagate normally — only OTel API failures (misconfiguration,
exporter unavailability) are suppressed.
When new_root=True, passes an empty Context() so the span
has no parent — it is a root span linked (not parented) to the
ambient trace. This satisfies the "linked, not parented"
requirement for PRODUCER spans in the cron loop.
The SDK's own exception handling is switched OFF and replaced by
:func:_record_scrubbed_error — see the comment at the call below.
Source code in src/taskq/obs/_otel.py
| @contextlib.contextmanager
def safe_start_span(
name: str,
*,
kind: trace.SpanKind | None = None,
attributes: Attributes = None,
links: Sequence[trace.Link] | None = None,
new_root: bool = False,
) -> Generator[Span, None, None]:
"""Start a span safely — never propagates exceptions from OTel API calls.
Checks ``_otel_enabled`` first; when ``False``, yields a no-op
``NonRecordingSpan``. When ``True``, delegates to
``get_tracer().start_as_current_span`` with a ``try/except`` around
span *creation* only. Exceptions from code inside the ``with`` block
propagate normally — only OTel API failures (misconfiguration,
exporter unavailability) are suppressed.
When ``new_root=True``, passes an empty ``Context()`` so the span
has no parent — it is a root span linked (not parented) to the
ambient trace. This satisfies the "linked, not parented"
requirement for PRODUCER spans in the cron loop.
The SDK's own exception handling is switched OFF and replaced by
:func:`_record_scrubbed_error` — see the comment at the call below.
"""
if not _otel_enabled:
yield trace.NonRecordingSpan(_NOOP_SPAN_CONTEXT)
return
ctx: Context | None = Context() if new_root else None
try:
span_cm = get_tracer().start_as_current_span(
name,
context=ctx,
kind=kind if kind is not None else trace.SpanKind.INTERNAL,
attributes=attributes,
links=links,
# Why: both default to True, and the SDK derives its event and its
# status description from the RAW ``str(exc)`` with no hook to
# override it. At every call site that scrubs and re-raises, that
# emitted a SECOND, unscrubbed ``exception`` event and overwrote
# the scrubbed status description -- shipping the Postgres DETAIL
# row values (idempotency_key / identity_key / fairness_key, all
# caller-supplied) straight to the telemetry backend, undoing
# ``_redact_exc`` entirely. No source-level guard can see this:
# the leaking call is made by the SDK, not by TaskQ.
record_exception=False,
set_status_on_exception=False,
)
except Exception:
_log.warning("otel-span-creation-failed", span_name=name)
yield trace.NonRecordingSpan(_NOOP_SPAN_CONTEXT)
return
with span_cm as span:
try:
yield span
except Exception as exc:
# ``Exception``, not ``BaseException``: mirrors what the SDK's own
# ``use_span`` catches, so cancellation semantics are unchanged.
_record_scrubbed_error(span, exc)
raise
|
set_otel_enabled
set_otel_enabled(enabled: bool) -> None
Set the module-level OTel enabled flag.
Called by worker startup code after loading WorkerSettings so that
all safe helpers check the flag without requiring a WorkerSettings
import at every call site. Avoids circular imports (modules like
dispatch.py import from obs and should not import from
settings.py in a circular path).
Source code in src/taskq/obs/_otel.py
| def set_otel_enabled(enabled: bool) -> None:
"""Set the module-level OTel enabled flag.
Called by worker startup code after loading ``WorkerSettings`` so that
all safe helpers check the flag without requiring a ``WorkerSettings``
import at every call site. Avoids circular imports (modules like
``dispatch.py`` import from ``obs`` and should not import from
``settings.py`` in a circular path).
"""
global _otel_enabled
_otel_enabled = enabled
|
update_disabled_schedules_count
update_disabled_schedules_count(count: int) -> None
Update the module-level disabled-schedules count.
Called by the leader's schedule management code when schedules are
disabled or re-enabled.
Source code in src/taskq/obs/_otel.py
| def update_disabled_schedules_count(count: int) -> None:
"""Update the module-level disabled-schedules count.
Called by the leader's schedule management code when schedules are
disabled or re-enabled.
"""
global _disabled_schedules_count
_disabled_schedules_count = count
|
update_heartbeat_consecutive_failures
update_heartbeat_consecutive_failures(
worker_id: str, count: int
) -> None
Update the module-level heartbeat consecutive-failures value.
Called by the heartbeat loop after each tick so the synchronous
gauge callback can read the latest value on scrape.
Source code in src/taskq/obs/_otel.py
| def update_heartbeat_consecutive_failures(worker_id: str, count: int) -> None:
"""Update the module-level heartbeat consecutive-failures value.
Called by the heartbeat loop after each tick so the synchronous
gauge callback can read the latest value on scrape.
"""
global _heartbeat_consecutive_failures_count
del worker_id # Why: not a dimension -- see the cardinality note above.
_heartbeat_consecutive_failures_count = count
|
update_queue_depth_cache
update_queue_depth_cache(data: dict[str, int]) -> None
Replace the queue-depth cache with fresh data from the leader's PG query.
Called by the background async task in the leader loop every 15s.
The synchronous gauge callback reads from this cache.
Source code in src/taskq/obs/_otel.py
| def update_queue_depth_cache(data: dict[str, int]) -> None:
"""Replace the queue-depth cache with fresh data from the leader's PG query.
Called by the background async task in the leader loop every 15s.
The synchronous gauge callback reads from this cache.
"""
global _queue_depth_cache
_queue_depth_cache = dict(data)
|
update_reservation_slots_cache
update_reservation_slots_cache(
data: dict[str, int],
) -> None
Replace the reservation-slots cache with fresh data from the leader's PG query.
Called by the background async task in the leader loop every 15s.
The synchronous gauge callback reads from this cache.
Source code in src/taskq/obs/_otel.py
| def update_reservation_slots_cache(data: dict[str, int]) -> None:
"""Replace the reservation-slots cache with fresh data from the leader's PG query.
Called by the background async task in the leader loop every 15s.
The synchronous gauge callback reads from this cache.
"""
global _reservation_slots_cache
_reservation_slots_cache = dict(data)
|
update_stranded_jobs_cache
update_stranded_jobs_cache(data: dict[str, int]) -> None
Replace the stranded-jobs cache with fresh data from the leader's query.
Stranded jobs are pending/scheduled jobs whose actor has no actor_config
row, which makes them permanently undispatchable: the dispatch CTE derives
its candidates from per_actor_capacity, which is FROM actor_config.
This gauge exists because the detector previously emitted a log line and
nothing else, exactly once per actor per process lifetime -- so the
condition was invisible in metrics and its only trace was a single WARN at
onset, which is the moment nobody is looking. An empty dict clears the
gauge, so recovery is visible too.
Source code in src/taskq/obs/_otel.py
| def update_stranded_jobs_cache(data: dict[str, int]) -> None:
"""Replace the stranded-jobs cache with fresh data from the leader's query.
Stranded jobs are pending/scheduled jobs whose actor has no `actor_config`
row, which makes them permanently undispatchable: the dispatch CTE derives
its candidates from `per_actor_capacity`, which is `FROM actor_config`.
This gauge exists because the detector previously emitted a log line and
nothing else, exactly once per actor per process lifetime -- so the
condition was invisible in metrics and its only trace was a single WARN at
onset, which is the moment nobody is looking. An empty dict clears the
gauge, so recovery is visible too.
"""
global _stranded_jobs_cache
_stranded_jobs_cache = dict(data)
|
record_exception_safe
record_exception_safe(
span: Span, exc: BaseException
) -> None
Record exc on span without leaking row values or credentials.
Emits the same exception event shape the OTel semantic conventions
define, so backends that special-case it still render an exception.
Source code in src/taskq/obs/_redact_exc.py
| def record_exception_safe(span: "Span", exc: BaseException) -> None:
"""Record *exc* on *span* without leaking row values or credentials.
Emits the same ``exception`` event shape the OTel semantic conventions
define, so backends that special-case it still render an exception.
"""
span.add_event(
"exception",
attributes={
"exception.type": type(exc).__qualname__,
"exception.message": safe_exception_message(exc),
"exception.stacktrace": _safe_stacktrace(exc),
},
)
|
safe_exception_message
safe_exception_message(exc: BaseException) -> str
Exception text with the Postgres DETAIL dropped and URI creds masked.
The primary Postgres message is kept: it is a static template naming the
constraint or relation, which is the part that is actually diagnostic.
HINT and CONTEXT are kept for the same reason -- neither carries
row values, and both are what an operator reads next.
Source code in src/taskq/obs/_redact_exc.py
| def safe_exception_message(exc: BaseException) -> str:
"""Exception text with the Postgres DETAIL dropped and URI creds masked.
The primary Postgres message is kept: it is a static template naming the
constraint or relation, which is the part that is actually diagnostic.
``HINT`` and ``CONTEXT`` are kept for the same reason -- neither carries
row values, and both are what an operator reads next.
"""
return _bound_message(_scrub_text(str(exc)))
|
set_exception_message_max_chars
set_exception_message_max_chars(limit: int) -> None
Set the module-level bound on scrubbed message text.
Mirrors :func:set_exception_redaction_enabled: a module global set once
at worker startup, so the obs layer needs no import of settings.
Source code in src/taskq/obs/_redact_exc.py
| def set_exception_message_max_chars(limit: int) -> None:
"""Set the module-level bound on scrubbed message text.
Mirrors :func:`set_exception_redaction_enabled`: a module global set once
at worker startup, so the obs layer needs no import of settings.
"""
global _max_message_chars
_max_message_chars = limit
|
set_exception_redaction_enabled
set_exception_redaction_enabled(enabled: bool) -> None
Set the module-level exception-redaction flag.
Mirrors :func:taskq.obs.set_otel_enabled: worker startup calls this once
after loading WorkerSettings so every scrub site reads a module global
instead of importing settings (which would be a circular import from
the modules that depend on obs).
Passing False disables the DETAIL drop on BOTH the span and the log
channel -- they share :func:_scrub_text, so the toggle cannot be applied
to one and not the other. It does NOT disable the URI credential mask.
Source code in src/taskq/obs/_redact_exc.py
| def set_exception_redaction_enabled(enabled: bool) -> None:
"""Set the module-level exception-redaction flag.
Mirrors :func:`taskq.obs.set_otel_enabled`: worker startup calls this once
after loading ``WorkerSettings`` so every scrub site reads a module global
instead of importing ``settings`` (which would be a circular import from
the modules that depend on ``obs``).
Passing ``False`` disables the DETAIL drop on BOTH the span and the log
channel -- they share :func:`_scrub_text`, so the toggle cannot be applied
to one and not the other. It does NOT disable the URI credential mask.
"""
global _redaction_enabled
_redaction_enabled = enabled
|
bind_job_context
bind_job_context(
log: BoundLogger,
*,
job_id: UUID,
actor: str,
queue: str,
attempt: int,
identity_key: str | None,
trace_id: str,
span_id: str | None = None,
batch_id: str | None = None,
) -> structlog.stdlib.BoundLogger
Bind job-scope fields to a logger, returning a new immutable BoundLogger.
identity_key, span_id, and batch_id are omitted from the bound
dict when None — not set to null or empty string . trace_id
is always bound (defaults to "" when no active OTel span per spec).
Returns a new BoundLogger; does not mutate the input.
Source code in src/taskq/obs/_structlog.py
| def bind_job_context(
log: structlog.stdlib.BoundLogger,
*,
job_id: UUID,
actor: str,
queue: str,
attempt: int,
identity_key: str | None,
trace_id: str,
span_id: str | None = None,
batch_id: str | None = None,
) -> structlog.stdlib.BoundLogger:
"""Bind job-scope fields to a logger, returning a new immutable BoundLogger.
``identity_key``, ``span_id``, and ``batch_id`` are omitted from the bound
dict when ``None`` — not set to null or empty string . ``trace_id``
is always bound (defaults to ``""`` when no active OTel span per spec).
Returns a new ``BoundLogger``; does not mutate the input.
"""
fields: dict[str, str | int] = {
"job_id": str(job_id),
"actor": actor,
"queue": queue,
"attempt": attempt,
"trace_id": trace_id,
}
if identity_key is not None:
fields["identity_key"] = identity_key
if span_id is not None:
fields["span_id"] = span_id
if batch_id is not None:
fields["batch_id"] = batch_id
return log.bind(**fields)
|
get_logger
get_logger(name: str) -> structlog.stdlib.BoundLogger
Return a structlog.stdlib.BoundLogger for the given dotted name.
Replaces direct structlog.get_logger() calls in library code so that
pyright strict mode gets an explicit return type (structlog.get_logger
returns Any).
Source code in src/taskq/obs/_structlog.py
| def get_logger(name: str) -> structlog.stdlib.BoundLogger:
"""Return a ``structlog.stdlib.BoundLogger`` for the given dotted name.
Replaces direct ``structlog.get_logger()`` calls in library code so that
pyright strict mode gets an explicit return type (``structlog.get_logger``
returns ``Any``).
"""
return structlog.get_logger(name)
|
log_cancel_phase_change
log_cancel_phase_change(
log: BoundLogger,
*,
from_phase: int,
to_phase: int,
**extra: object,
) -> None
Emit an INFO log line with kind="cancel_phase_change".
from_phase and to_phase are the cancel-phase integers before
and after the escalation. cancel_observed_at is NOT included — it
is per-handler context, not part of the canonical schema.
Source code in src/taskq/obs/_structlog.py
| def log_cancel_phase_change(
log: structlog.stdlib.BoundLogger,
*,
from_phase: int,
to_phase: int,
**extra: object,
) -> None:
"""Emit an INFO log line with ``kind="cancel_phase_change"``.
``from_phase`` and ``to_phase`` are the cancel-phase integers before
and after the escalation. ``cancel_observed_at`` is NOT included — it
is per-handler context, not part of the canonical schema.
"""
log.info(
"cancel_phase_change",
kind="cancel_phase_change",
from_phase=from_phase,
to_phase=to_phase,
**extra,
)
|
log_state_change
log_state_change(
log: BoundLogger,
*,
from_state: str,
to_state: str,
**extra: object,
) -> None
Emit an INFO log line with kind="state_change".
from_state and to_state are the job-status values before and
after the transition. All bound fields from the pre-bound log
(which carries job context from :func:bind_job_context) are included
automatically. The event name is "state-change" so the log is
queryable by both event and kind.
Source code in src/taskq/obs/_structlog.py
| def log_state_change(
log: structlog.stdlib.BoundLogger,
*,
from_state: str,
to_state: str,
**extra: object,
) -> None:
"""Emit an INFO log line with ``kind="state_change"``.
``from_state`` and ``to_state`` are the job-status values before and
after the transition. All bound fields from the pre-bound ``log``
(which carries job context from :func:`bind_job_context`) are included
automatically. The event name is ``"state-change"`` so the log is
queryable by both event and kind.
"""
log.info("state-change", kind="state_change", from_state=from_state, to_state=to_state, **extra)
|
redact_payload
redact_payload(payload: object) -> str
Return the first 16 characters of the SHA-256 hex digest of the JSON-serialized payload.
Raw payload content does not appear in the return value. Deterministic
for the same input.
Source code in src/taskq/obs/_structlog.py
| def redact_payload(payload: object) -> str:
"""Return the first 16 characters of the SHA-256 hex digest of the JSON-serialized payload.
Raw payload content does not appear in the return value. Deterministic
for the same input.
"""
serialized = dumps_str(payload).encode()
return hashlib.sha256(serialized).hexdigest()[:16]
|
setup_logging
setup_logging(
*, level: str = "INFO", log_format: str = "json"
) -> None
Configure structlog with the canonical processor chain.
Production (log_format="json"): JSONRenderer via
ProcessorFormatter stdlib bridge. Development (log_format="console"):
ConsoleRenderer via ProcessorFormatter. Idempotent — guarded
by _logging_configured flag. Not called at import time .
Source code in src/taskq/obs/_structlog.py
| def setup_logging(
*,
level: str = "INFO",
log_format: str = "json",
) -> None:
"""Configure structlog with the canonical processor chain.
Production (``log_format="json"``): ``JSONRenderer`` via
``ProcessorFormatter`` stdlib bridge. Development (``log_format="console"``):
``ConsoleRenderer`` via ``ProcessorFormatter``. Idempotent — guarded
by ``_logging_configured`` flag. Not called at import time .
"""
global _logging_configured
if _logging_configured:
return
shared_processors: list[structlog.types.Processor] = [
_safe_processor_wrapper(structlog.contextvars.merge_contextvars),
_safe_processor_wrapper(structlog.stdlib.add_log_level),
_safe_processor_wrapper(structlog.stdlib.add_logger_name),
_safe_processor_wrapper(structlog.processors.StackInfoRenderer()),
_safe_processor_wrapper(structlog.processors.TimeStamper(fmt="iso", utc=True)),
_safe_processor_wrapper(_otel_span_processor),
_safe_processor_wrapper(structlog.processors.EventRenamer("event")),
# Last before the formatter handoff: final scrub of exception-bearing
# fields so both renderers (and any future one) see scrubbed values.
_safe_processor_wrapper(_scrub_exception_fields),
# SHARED, not formatter-local: whatever survives this chain becomes
# ``record.msg`` and is read by every root handler, not just TaskQ's.
# A raw exception object or ``sys.exc_info()`` triple left on the event
# dict is therefore an export surface for any vendor handler that
# stringifies values. Console pays for this with a plain scrubbed
# ``exception.stacktrace`` field instead of ConsoleRenderer's pretty
# traceback — the same record reaches the same vendor handlers whichever
# renderer the operator picked, so the dev view does not get an
# unredacted exemption.
_safe_processor_wrapper(_render_exc_info_safe),
]
formatter_processors: list[structlog.types.Processor]
if log_format == "console":
renderer: structlog.types.Processor = structlog.dev.ConsoleRenderer()
formatter_processors = [
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
renderer,
]
else:
from taskq._json import structlog_serializer
renderer = structlog.processors.JSONRenderer(serializer=structlog_serializer)
formatter_processors = [
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
# Still needed for FOREIGN records: ``ProcessorFormatter`` lifts
# their ``record.exc_info`` onto the event dict here, after the
# shared chain has run, and orjson drops the whole line on a raw
# tuple. Idempotent for TaskQ's own records — ``exc_info`` is
# already gone by then.
_safe_processor_wrapper(_render_exc_info_safe),
renderer,
]
structlog.configure(
processors=[
*shared_processors,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
wrapper_class=_ExcInfoSafeBoundLogger,
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
formatter = structlog.stdlib.ProcessorFormatter(
processors=formatter_processors,
foreign_pre_chain=[
_safe_processor_wrapper(structlog.processors.TimeStamper(fmt="iso", utc=True)),
_safe_processor_wrapper(structlog.stdlib.add_log_level),
_safe_processor_wrapper(structlog.stdlib.ExtraAdder()),
# After ExtraAdder so foreign records' extras are scrubbed too.
_safe_processor_wrapper(_scrub_exception_fields),
],
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
if not any(
isinstance(h, logging.StreamHandler)
and isinstance(h.formatter, structlog.stdlib.ProcessorFormatter)
for h in logging.root.handlers
):
logging.root.addHandler(handler)
logging.root.setLevel(level)
_logging_configured = True
|
invoke_error_reporter
async
invoke_error_reporter(
reporter: ErrorReporter | None,
job: JobRow,
exception: BaseException,
timeout: float = 3.0,
*,
log: BoundLogger | None = None,
) -> None
Invoke reporter.report(), swallowing and counting failures.
A None reporter is treated as a no-op (equivalent to
:class:NullErrorReporter). Exceptions from report() are caught,
logged at WARNING, and counted on the taskq.error_reporter.failures
counter — they never propagate to the caller.
This mirrors the defensive pattern of
:func:~taskq.retry.invoke_on_retry_exhausted: a user-supplied hook
must never crash the worker's terminal-write path.
Source code in src/taskq/obs/error_reporter.py
| async def invoke_error_reporter(
reporter: ErrorReporter | None,
job: JobRow,
exception: BaseException,
timeout: float = 3.0, # noqa: ASYNC109 Why: parameter name matches the on_retry_exhausted/invoke_on_success convention; asyncio.wait_for requires a timeout value, not asyncio.timeout() context manager
*,
log: structlog.stdlib.BoundLogger | None = None,
) -> None:
"""Invoke *reporter*.report(), swallowing and counting failures.
A ``None`` reporter is treated as a no-op (equivalent to
:class:`NullErrorReporter`). Exceptions from ``report()`` are caught,
logged at WARNING, and counted on the ``taskq.error_reporter.failures``
counter — they never propagate to the caller.
This mirrors the defensive pattern of
:func:`~taskq.retry.invoke_on_retry_exhausted`: a user-supplied hook
must never crash the worker's terminal-write path.
"""
if reporter is None:
return
logger: structlog.stdlib.BoundLogger = log if log is not None else _log
reporter_type = type(reporter).__name__
try:
await asyncio.wait_for(reporter.report(job, exception), timeout=timeout)
except TimeoutError:
logger.warning(
"error-reporter-timeout",
job_id=str(job.id),
actor=job.actor,
reporter_type=reporter_type,
timeout_seconds=timeout,
)
except Exception as exc:
logger.warning(
"error-reporter-failed",
job_id=str(job.id),
actor=job.actor,
reporter_type=reporter_type,
error=repr(exc),
)
record_error_reporter_failure(reporter_type)
|