nemo_gym.telemetry.observer

Event-loop, process, request, and rollout-driver metrics that show what is slowing a Gym process down.

A Gym server answers every request on one asyncio event loop. When synchronous work such as validation, serialization, or garbage collection holds that loop, every other request waits, even though no single request looks slow. These metrics make that pressure visible without inspecting private asyncio internals:

gym.event_loop.lag_ms (histogram, milliseconds): how late a sleeping coroutine wakes up. A coroutine asks to wake after a fixed interval, and the lag is how much later than that it actually runs. This is the scheduler-pressure signal.
gym.event_loop.tasks (gauge, opt-in): how many asyncio tasks exist on the loop. This is resident work, not a queue depth: an idle task waiting on the network counts as much as a runnable one.
gym.gc.pause_duration_ms (histogram, milliseconds), and the gym.gc.collections_total, gym.gc.collected_total, and gym.gc.uncollectable_total counters, all by nemo.gym.gc.generation. gym.gc.count (gauge, by nemo.gym.gc.generation) is the allocation count from gc.get_count().
process.cpu.time (counter, seconds, by cpu.mode), process.memory.usage, process.thread.count, and process.open_file_descriptor.count. These keep the OpenTelemetry process semantic-convention names, units, and attributes, so standard dashboards recognize them.
gym.server.requests.in_flight, gym.server.requests.unhandled_exceptions_total, and gym.server.responses_total by http.response.status_code. gym.server.requests.cancelled_total carries http.route and nemo.gym.request.phase. The outermost server middleware keeps these totals locally, and the exporter thread reads them.
gym.http.client.requests.in_flight: outbound calls through server_utils.request still waiting for their response headers.
gym.driver.rollouts.offered_total, gym.driver.rollouts.resident, and gym.driver.rollouts.active, plus the gym.driver.admission_duration_ms, gym.driver.output_write_duration_ms, gym.driver.output_write_bytes, and gym.driver.materialize_duration_ms histograms, from the rollout driver.
gym.driver.rollout.input_items, gym.driver.rollout.output_items, gym.driver.rollout.turns, and gym.driver.rollout.tool_calls: the size of each completed rollout, recorded once per rollout by the driver.

Each kind of work runs where it costs the least:

The event loop runs one sampler coroutine per interval, which measures lag, and counts tasks when that is on.
The exporter’s background thread reads process CPU, memory, threads, file descriptors, and GC totals when it exports, through observable instruments.
The garbage collector’s callback only stores timestamps and totals. It never calls OpenTelemetry, because it runs inside a collection. Pause durations wait in a bounded buffer until the next sample records them.

Process-wide state, which is the GC callback and the observable instruments, is set up once per process. The last event loop to stop removes it. These metrics are on whenever telemetry metrics are, because each costs at most a few microseconds per request or rollout. The one exception is the event-loop task count, which walks every task on the loop and is opt-in through telemetry.event_loop_task_count. Every failure is logged and swallowed, because telemetry must never stop a server.

Module Contents

Classes

Name	Description
`DriverCounts`	Rollout totals that the driver updates with plain integer arithmetic, read by the exporter thread.
`MeteredAdmission`	Wraps the driver’s concurrency limit to record each rollout’s wait and count the rollouts running inside it.
`RequestCounts`	Request totals that the server middleware and `server_utils.request` update with plain integer arithmetic.
`_GcRecorder`	Collects GC pause times and totals from `gc.callbacks` without calling OpenTelemetry.
`_Instruments`	The observer’s instruments for one meter, created once so a restarted observer never registers them twice.
`_ProcessObserver`	The per-process state of the observer: the GC recorder, the request totals, and this process’s handle.

Functions

Name	Description
`_acquire_process_observer`	-
`_by_generation`	-
`_cpu_times`	-
`_driver_total`	Report a driver total only in a process that has offered rollouts, so servers export no constant-zero series.
`_flush_metrics`	-
`_observation`	-
`_observe`	Wrap a reading of the running observer as an observable-instrument callback.
`_release_process_observer`	-
`_sample`	-
`count_resident`	Await one rollout, counting it as resident until it finishes.
`driver_metrics`	Return the process’s observer state while an observer is running, for the rollout driver, else None.
`observe_app_event_loop`	Run `observe_event_loop` for the lifetime of a FastAPI app, keeping any lifespan the app already has.
`observe_event_loop`	Sample the running event loop, and observe this process, for as long as the context is open.
`observer_settings`	Return the sample interval and whether to count tasks when metrics are exporting in this process, else None.
`request_counts`	Return the request totals while an observer is running, else None.

Data

CPU_MODE_ATTRIBUTE

DRIVER_ACTIVE_INSTRUMENT

DRIVER_ADMISSION_INSTRUMENT

DRIVER_DURATION_BOUNDARIES_MS

DRIVER_MATERIALIZE_INSTRUMENT

DRIVER_OFFERED_INSTRUMENT

DRIVER_RESIDENT_INSTRUMENT

DRIVER_WRITE_BYTES_BOUNDARIES

DRIVER_WRITE_BYTES_INSTRUMENT

DRIVER_WRITE_DURATION_INSTRUMENT

EVENT_LOOP_ATTRIBUTE

EVENT_LOOP_LAG_BOUNDARIES_MS

EVENT_LOOP_LAG_INSTRUMENT

EVENT_LOOP_TASKS_INSTRUMENT

GC_COLLECTED_INSTRUMENT

GC_COLLECTIONS_INSTRUMENT

GC_COUNT_INSTRUMENT

GC_GENERATION_ATTRIBUTE

GC_PAUSE_BOUNDARIES_MS

GC_PAUSE_INSTRUMENT

GC_UNCOLLECTABLE_INSTRUMENT

HTTP_ROUTE_ATTRIBUTE

HTTP_STATUS_CODE_ATTRIBUTE

OUTBOUND_IN_FLIGHT_INSTRUMENT

PROCESS_CPU_TIME_INSTRUMENT

PROCESS_MEMORY_USAGE_INSTRUMENT

PROCESS_OPEN_FDS_INSTRUMENT

PROCESS_THREAD_COUNT_INSTRUMENT

REQUESTS_CANCELLED_INSTRUMENT

REQUESTS_IN_FLIGHT_INSTRUMENT

REQUESTS_UNHANDLED_EXCEPTIONS_INSTRUMENT

REQUEST_PHASE_ATTRIBUTE

RESPONSES_INSTRUMENT

ROLLOUT_COUNT_BOUNDARIES

ROLLOUT_INPUT_ITEMS_INSTRUMENT

ROLLOUT_OUTPUT_ITEMS_INSTRUMENT

ROLLOUT_TOOL_CALLS_INSTRUMENT

ROLLOUT_TURNS_INSTRUMENT

_ACTIVE_LOOPS

_INSTRUMENTS

_MAX_BUFFERED_GC_PAUSES

_PROCESS_LOCK

_PROCESS_OBSERVER

logger

API

class nemo_gym.telemetry.observer.DriverCounts()

Rollout totals that the driver updates with plain integer arithmetic, read by the exporter thread.

active

= 0

Rollouts admitted past the concurrency limit and running.

offered

= 0

Rollouts handed to dispatch.

resident

= 0

Rollout tasks that exist and have not finished, whether waiting for admission or running.

class nemo_gym.telemetry.observer.MeteredAdmission(
    semaphore: typing.Any,
    observer: nemo_gym.telemetry.observer._ProcessObserver
)

Wraps the driver’s concurrency limit to record each rollout’s wait and count the rollouts running inside it.

nemo_gym.telemetry.observer.MeteredAdmission.__aenter__() -> None

async

nemo_gym.telemetry.observer.MeteredAdmission.__aexit__(
    exc_info: typing.Any = ()
) -> typing.Any

async

class nemo_gym.telemetry.observer.RequestCounts()

Request totals that the server middleware and server_utils.request update with plain integer arithmetic.

The exporter thread reads them through observable instruments, so a request costs no OpenTelemetry call. The serving event loop updates the inbound totals. outbound_in_flight can also change on another thread’s event loop, such as the one the synchronous sandbox API runs on. Each update is a single += on an attribute with no call in between, which CPython with the GIL never interrupts with another thread, so no update is lost.

cancelled

= 0

cancelled_by_route_and_phase

dict[tuple[str, str], int] = {}

Cumulative cancellations by registered route template and response-start phase.

in_flight

= 0

outbound_in_flight

= 0

responses

dict[int, int] = {}

Responses by HTTP status code, updated only by the serving event loop. HTTP bounds the set of codes.

unhandled_exceptions

= 0

nemo_gym.telemetry.observer.RequestCounts.count_cancellation(
    route: str,
    response_started: bool
) -> None

Count one disconnect; route must be a registered template or the unmatched sentinel.

nemo_gym.telemetry.observer.RequestCounts.count_response(
    status: int
) -> None

class nemo_gym.telemetry.observer._GcRecorder()

Collects GC pause times and totals from gc.callbacks without calling OpenTelemetry.

It takes no lock. A lock would deadlock when a collection starts on a thread that already holds it, for example while drain_pauses allocates. Python runs one collection at a time, so the callback never races itself, and the buffer swap in drain_pauses is safe under the GIL.

_pauses

list[tuple[int, float]] = []

_started

dict[int, float] = {}

collected

= [0, 0, 0]

collections

= [0, 0, 0]

uncollectable

= [0, 0, 0]

nemo_gym.telemetry.observer._GcRecorder.__call__(
    phase: str,
    info: dict[str, int]
) -> None

nemo_gym.telemetry.observer._GcRecorder.drain_pauses() -> list[tuple[int, float]]

class nemo_gym.telemetry.observer._Instruments(
    meter: typing.Any
)

The observer’s instruments for one meter, created once so a restarted observer never registers them twice.

admission

gc_pause

lag

materialize

tasks

write_bytes

write_duration

class nemo_gym.telemetry.observer._ProcessObserver(
    meter: typing.Any
)

The per-process state of the observer: the GC recorder, the request totals, and this process’s handle.

admission

= instruments.admission

driver

= DriverCounts()

= _GcRecorder()

gc_pause

= instruments.gc_pause

input_items

= instruments.input_items

lag

= instruments.lag

materialize

= instruments.materialize

output_items

= instruments.output_items

process

= psutil.Process()

requests

= RequestCounts()

tasks

= instruments.tasks

tool_calls

= instruments.tool_calls

turns

= instruments.turns

write_bytes

= instruments.write_bytes

write_duration

= instruments.write_duration

nemo_gym.telemetry.observer._ProcessObserver.record_gc_pauses() -> None

nemo_gym.telemetry.observer._acquire_process_observer(
    meter: typing.Any
) -> nemo_gym.telemetry.observer._ProcessObserver

nemo_gym.telemetry.observer._by_generation(
    values: typing.Callable[[_ProcessObserver], list[int]]
) -> typing.Callable[[Any], list]

nemo_gym.telemetry.observer._cpu_times(
    state: nemo_gym.telemetry.observer._ProcessObserver
) -> list

nemo_gym.telemetry.observer._driver_total(
    read: typing.Callable[[DriverCounts], int]
) -> typing.Callable[[Any], list]

Report a driver total only in a process that has offered rollouts, so servers export no constant-zero series.

nemo_gym.telemetry.observer._flush_metrics() -> None

nemo_gym.telemetry.observer._observation(
    value: float,
    attributes: typing.Optional[dict[str, typing.Any]] = None
) -> typing.Any

nemo_gym.telemetry.observer._observe(
    read: typing.Callable[[_ProcessObserver], list]
) -> typing.Callable[[Any], list]

Wrap a reading of the running observer as an observable-instrument callback.

It reports nothing while no observer runs, and one unavailable measurement never stops the others from exporting.

nemo_gym.telemetry.observer._release_process_observer() -> None

nemo_gym.telemetry.observer._sample(
    state: nemo_gym.telemetry.observer._ProcessObserver,
    interval_seconds: float,
    loop_name: str,
    count_tasks: bool
) -> None

async

nemo_gym.telemetry.observer.count_resident(
    awaitable: typing.Any,
    observer: nemo_gym.telemetry.observer._ProcessObserver
) -> typing.Any

async

Await one rollout, counting it as resident until it finishes.

nemo_gym.telemetry.observer.driver_metrics() -> typing.Optional[nemo_gym.telemetry.observer._ProcessObserver]

Return the process’s observer state while an observer is running, for the rollout driver, else None.

nemo_gym.telemetry.observer.observe_app_event_loop(
    app: typing.Any,
    loop_name: str
) -> None

Run observe_event_loop for the lifetime of a FastAPI app, keeping any lifespan the app already has.

nemo_gym.telemetry.observer.observe_event_loop(
    loop_name: str
) -> typing.AsyncIterator[None]

async

Sample the running event loop, and observe this process, for as long as the context is open.

loop_name is a bounded label for the loop, such as server, head_server, or driver. Does nothing unless telemetry metrics are exporting.

nemo_gym.telemetry.observer.observer_settings() -> typing.Optional[tuple[float, bool]]

Return the sample interval and whether to count tasks when metrics are exporting in this process, else None.

nemo_gym.telemetry.observer.request_counts() -> typing.Optional[nemo_gym.telemetry.observer.RequestCounts]

Return the request totals while an observer is running, else None.

Server middleware calls this once per request, so it is a plain global read.

nemo_gym.telemetry.observer.CPU_MODE_ATTRIBUTE = 'cpu.mode'

nemo_gym.telemetry.observer.DRIVER_ACTIVE_INSTRUMENT = 'gym.driver.rollouts.active'

nemo_gym.telemetry.observer.DRIVER_ADMISSION_INSTRUMENT = 'gym.driver.admission_duration_ms'

nemo_gym.telemetry.observer.DRIVER_DURATION_BOUNDARIES_MS: tuple[float, ...] = (0.1, 1, 10, 100, 500, 1000, 5000, 10000, 30000, 60000, 300000, 1800000)

nemo_gym.telemetry.observer.DRIVER_MATERIALIZE_INSTRUMENT = 'gym.driver.materialize_duration_ms'

nemo_gym.telemetry.observer.DRIVER_OFFERED_INSTRUMENT = 'gym.driver.rollouts.offered_total'

nemo_gym.telemetry.observer.DRIVER_RESIDENT_INSTRUMENT = 'gym.driver.rollouts.resident'

nemo_gym.telemetry.observer.DRIVER_WRITE_BYTES_BOUNDARIES: tuple[float, ...] = (1000, 10000, 100000, 1000000, 10000000, 100000000)

nemo_gym.telemetry.observer.DRIVER_WRITE_BYTES_INSTRUMENT = 'gym.driver.output_write_bytes'

nemo_gym.telemetry.observer.DRIVER_WRITE_DURATION_INSTRUMENT = 'gym.driver.output_write_duration_ms'

nemo_gym.telemetry.observer.EVENT_LOOP_ATTRIBUTE = 'nemo.gym.event_loop'

nemo_gym.telemetry.observer.EVENT_LOOP_LAG_BOUNDARIES_MS: tuple[float, ...] = (0.5, 1, 2.5, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000)

nemo_gym.telemetry.observer.EVENT_LOOP_LAG_INSTRUMENT = 'gym.event_loop.lag_ms'

nemo_gym.telemetry.observer.EVENT_LOOP_TASKS_INSTRUMENT = 'gym.event_loop.tasks'

nemo_gym.telemetry.observer.GC_COLLECTED_INSTRUMENT = 'gym.gc.collected_total'

nemo_gym.telemetry.observer.GC_COLLECTIONS_INSTRUMENT = 'gym.gc.collections_total'

nemo_gym.telemetry.observer.GC_COUNT_INSTRUMENT = 'gym.gc.count'

nemo_gym.telemetry.observer.GC_GENERATION_ATTRIBUTE = 'nemo.gym.gc.generation'

nemo_gym.telemetry.observer.GC_PAUSE_BOUNDARIES_MS: tuple[float, ...] = (0.1, 0.25, 0.5, 1, 2.5, 5, 10, 25, 50, 100, 250, 500, 1000)

nemo_gym.telemetry.observer.GC_PAUSE_INSTRUMENT = 'gym.gc.pause_duration_ms'

nemo_gym.telemetry.observer.GC_UNCOLLECTABLE_INSTRUMENT = 'gym.gc.uncollectable_total'

nemo_gym.telemetry.observer.HTTP_ROUTE_ATTRIBUTE = 'http.route'

nemo_gym.telemetry.observer.HTTP_STATUS_CODE_ATTRIBUTE = 'http.response.status_code'

nemo_gym.telemetry.observer.OUTBOUND_IN_FLIGHT_INSTRUMENT = 'gym.http.client.requests.in_flight'

nemo_gym.telemetry.observer.PROCESS_CPU_TIME_INSTRUMENT = 'process.cpu.time'

nemo_gym.telemetry.observer.PROCESS_MEMORY_USAGE_INSTRUMENT = 'process.memory.usage'

nemo_gym.telemetry.observer.PROCESS_OPEN_FDS_INSTRUMENT = 'process.open_file_descriptor.count'

nemo_gym.telemetry.observer.PROCESS_THREAD_COUNT_INSTRUMENT = 'process.thread.count'

nemo_gym.telemetry.observer.REQUESTS_CANCELLED_INSTRUMENT = 'gym.server.requests.cancelled_total'

nemo_gym.telemetry.observer.REQUESTS_IN_FLIGHT_INSTRUMENT = 'gym.server.requests.in_flight'

nemo_gym.telemetry.observer.REQUESTS_UNHANDLED_EXCEPTIONS_INSTRUMENT = 'gym.server.requests.unhandled_exceptions_total'

nemo_gym.telemetry.observer.REQUEST_PHASE_ATTRIBUTE = 'nemo.gym.request.phase'

nemo_gym.telemetry.observer.RESPONSES_INSTRUMENT = 'gym.server.responses_total'

nemo_gym.telemetry.observer.ROLLOUT_COUNT_BOUNDARIES: tuple[float, ...] = (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 4096)

nemo_gym.telemetry.observer.ROLLOUT_INPUT_ITEMS_INSTRUMENT = 'gym.driver.rollout.input_items'

nemo_gym.telemetry.observer.ROLLOUT_OUTPUT_ITEMS_INSTRUMENT = 'gym.driver.rollout.output_items'

nemo_gym.telemetry.observer.ROLLOUT_TOOL_CALLS_INSTRUMENT = 'gym.driver.rollout.tool_calls'

nemo_gym.telemetry.observer.ROLLOUT_TURNS_INSTRUMENT = 'gym.driver.rollout.turns'

nemo_gym.telemetry.observer._ACTIVE_LOOPS = 0

nemo_gym.telemetry.observer._INSTRUMENTS: WeakKeyDictionary[Any, _Instruments] = weakref.WeakKeyDictionary()

nemo_gym.telemetry.observer._MAX_BUFFERED_GC_PAUSES = 10000

nemo_gym.telemetry.observer._PROCESS_LOCK = threading.Lock()

nemo_gym.telemetry.observer._PROCESS_OBSERVER: Optional[_ProcessObserver] = None

nemo_gym.telemetry.observer.logger = logging.getLogger(__name__)