faust.sensors.otel

Monitor reporting metrics via OpenTelemetry.

This sensor mirrors the metric set that StatsdMonitor reports, but records it through the OpenTelemetry metrics API instead of a statsd socket. Faust depends only on opentelemetry-api: every instrument is a cheap no-op until the application configures a global MeterProvider with the exporter of its choice (OTLP, Prometheus, console, …), so importing this module never hard-fails Faust core.

Unlike statsd – which encodes dimensions such as topic and partition into the metric name – the OpenTelemetry conventions favour a single instrument dimensioned by attributes. This sensor follows that convention: e.g. a single faust.messages.received counter carries topic/partition attributes rather than one metric per topic.

class faust.sensors.otel.OTelMetrics(meter: Meter)

Container for the OpenTelemetry instruments used by the monitor.

Instruments are created once, up front, from a single Meter; recording happens by supplying attributes at call time. Latency histograms use the ms unit because Faust observes every latency in milliseconds (see ms_since()).

messages_received: Counter

Monotonic counters (.add).

events_total: Counter

messages_sent: Counter

messages_send_errors: Counter

table_operations: Counter

assignments: Counter

http_requests: Counter

custom_counts: Counter

messages_active: UpDownCounter

Up/down counters for values that rise and fall (.add +/-).

events_active: UpDownCounter

rebalances_active: UpDownCounter

rebalances_recovering: UpDownCounter

events_runtime: Histogram

Latency histograms in milliseconds (.record).

commit_latency: Histogram

send_latency: Histogram

send_error_latency: Histogram

assignment_latency: Histogram

rebalance_return_latency: Histogram

rebalance_end_latency: Histogram

http_latency: Histogram

offset_read: Gauge

Synchronous gauges for last-known values (.set).

offset_committed: Gauge

offset_end: Gauge

producer_buffer: Gauge

class faust.sensors.otel.OpenTelemetryMonitor(*, meter: Optional[Meter] = None, service_name: str = 'faust', metrics: Optional[OTelMetrics] = None, **kwargs: Any)

OpenTelemetry Faust Sensor.

Records the same metrics as StatsdMonitor through the OpenTelemetry metrics API, dimensioned by attributes.

The application is responsible for configuring a global MeterProvider (and an exporter) before starting the app; until then the instruments are cheap no-ops. A meter may be supplied explicitly, otherwise one is resolved from the globally-registered provider.

logger: logging.Logger = <Logger faust.sensors.otel (WARNING)>

meter

Return the OpenTelemetry meter for this monitor.

metrics

Return the container of OpenTelemetry instruments.

on_message_in(tp: TP, offset: int, message: Message) → None

Call before message is delegated to streams.

Return type:

None

on_message_out(tp: TP, offset: int, message: Message) → None

Call when message is fully acknowledged and can be committed.

Return type:

None

on_stream_event_in(tp: TP, offset: int, stream: StreamT, event: EventT) → Optional[Dict]

Call when stream starts processing an event.

Return type:

_UnionGenericAlias[_SpecialGenericAlias, None]

on_stream_event_out(tp: TP, offset: int, stream: StreamT, event: EventT, state: Optional[Dict] = None) → None

Call when stream is done processing an event.

Return type:

None

on_table_get(table: CollectionT, key: Any) → None

Call when value in table is retrieved.

Return type:

None

on_table_set(table: CollectionT, key: Any, value: Any) → None

Call when new value for key in table is set.

Return type:

None

on_table_del(table: CollectionT, key: Any) → None

Call when key in a table is deleted.

Return type:

None

on_send_completed(producer: ProducerT, state: Any, metadata: RecordMetadata) → None

Call when producer finished sending message.

Return type:

None

on_send_error(producer: ProducerT, exc: BaseException, state: Any) → None

Call when producer was unable to publish message.

Return type:

None

on_assignment_error(assignor: PartitionAssignorT, state: Dict, exc: BaseException) → None

Partition assignor did not complete assignor due to error.

Return type:

None

on_assignment_completed(assignor: PartitionAssignorT, state: Dict) → None

Partition assignor completed assignment.

Return type:

None

on_rebalance_start(app: AppT) → Dict

Cluster rebalance in progress.

Return type:

_SpecialGenericAlias

on_rebalance_return(app: AppT, state: Dict) → None

Consumer replied assignment is done to broker.

Return type:

None

on_rebalance_end(app: AppT, state: Dict) → None

Cluster rebalance fully completed (including recovery).

Return type:

None

on_commit_completed(consumer: ConsumerT, state: Any) → None

Call when consumer commit offset operation completed.

Return type:

None

on_tp_commit(tp_offsets: MutableMapping[TP, int]) → None

Call when offset in topic partition is committed.

Return type:

None

track_tp_end_offset(tp: TP, offset: int) → None

Track new topic partition end offset for monitoring lags.

Return type:

None

on_web_request_end(app: AppT, request: Request, response: Optional[Response], state: Dict, *, view: Optional[View] = None) → None

Web server finished working on request.

Return type:

None

on_threaded_producer_buffer_processed(app: AppT, size: int) → None

Call when the threaded producer flushed its buffer.

Return type:

None

count(metric_name: str, count: int = 1) → None

Count metric by name.

Return type:

None