feat(observability): RPC + workflow + DB-pool metrics for the agent JSON-RPC endpoint

[stephen-wang24]

Summary

Brings the agent JSON-RPC endpoint (POST /agents/{id}/rpc — task/create / message/send / event/send) up to the metrics observability contract, and adds high-signal Postgres connection-pool saturation metrics. Everything here is OTel-only (no StatsD dual-emit — these are new series with no existing consumer), and a cheap no-op when no OTLP endpoint is configured.

RPC RED metrics (agentex.rpc.*)

• agentex.rpc.request.duration (timed dispatch → final byte, so streaming responses are covered end-to-end), agentex.rpc.requests, agentex.rpc.errors.
• Attributes follow the OTel RPC semantic conventions: rpc.system, rpc.method, rpc.jsonrpc.error_code, error.type, streaming. No high-cardinality identifiers (no task/agent/request ids). JSON-RPC failures return HTTP 200, so the error code rides on rpc.jsonrpc.error_code (set only on failure), not an HTTP status.
• Never-raise emission; a single terminal emit per request covering the whole stream.

Workflow-level GenAI metric

• gen_ai.workflow.duration for task/create, labeled gen_ai.operation.name=invoke_workflow, so workflow-level duration is directly queryable instead of only inferable from RPC duration. invoke_agent stays reserved for the agent loop's own duration, so the two operation names never mix gateway and in-pod durations.

DB connection-pool metrics (OTel DB semconv)

• db.client.connection.wait_time, db.client.connection.pending_requests, db.client.connection.timeouts — the pre-outage early-warning signals for pool exhaustion.
• Sourced from an InstrumentedAsyncAdaptedQueuePool that wraps Pool.connect() (the single, non-recursive acquisition entry point, run inside the acquiring greenlet). The existing checkout/checkin listeners can't produce these: by the time checkout fires the wait is over, there's no waiter count, and a pool timeout raises before any connection is handed out.
• wait_time uses explicit second-scale histogram buckets so quantiles stay usable (the SDK default boundaries assume a millisecond unit).

Testing

• Unit tests in tests/unit/utils/test_rpc_metrics.py and test_db_metrics.py: unconfigured no-op, never-raise on backend faults, metric name/attribute assertions, and the pool success / timeout / non-timeout-error paths. All green; ruff clean.

Notes / follow-ups

• Requires an OTLP endpoint configured on the pods to export.
• HTTP auto-instrumentation (route-templated http_server_request_duration_seconds) lands on a separate branch.
• Dashboard/Mimir verification and matrix backfill to follow once the target environment is available.

🤖 Generated with Claude Code

Greptile Summary

This PR adds OTel-only RED metrics for the agent JSON-RPC endpoint (agentex.rpc.*), a GenAI workflow-duration histogram for task/create, and Postgres connection-pool acquisition metrics (db.client.connection.wait_time/pending_requests/timeouts/failures) sourced from a new InstrumentedAsyncAdaptedQueuePool subclass. It also removes the namespace: agentex from the local OTel collector's Prometheus exporter to fix a double-prefix bug that would have appeared for new agentex.*-prefixed instrument names.

• RPC metrics: A synchronous @contextmanager (rpc_request_timing) wraps both sync and streaming handlers; a RpcCallOutcome dataclass lets handlers classify JSON-RPC error codes, with BaseException catching client disconnects (e.g., GeneratorExit) separately from structured RPC errors. Never-raise contract is enforced throughout.
• DB pool metrics: InstrumentedAsyncAdaptedQueuePool.connect() is the correct interception seam (checkout/checkin events can\u2019t see the wait or a timeout), instruments are attached post-construction and carried across recreate(), and the wait_time histogram uses explicit second-scale buckets to keep quantiles usable.
• Collector config: Removing namespace: agentex aligns local PromQL with the production Mimir pipeline but renames existing agentex_db_* local series to db_*; teams with local Grafana dashboards querying the old prefixed names will need to update their queries.

Confidence Score: 5/5

Safe to merge; all changes are additive instrumentation with a never-raise emission contract and no mutations to the core RPC execution path.

The RPC and DB pool metric additions are purely observational. The rpc_request_timing contextmanager correctly handles BaseException including GeneratorExit, the pool override degrades to a passthrough when OTel is unconfigured, and recreate() carries instrumentation forward. Unit tests cover the no-op path, never-raise contract, double-count regression, and the streaming GeneratorExit scenario.

Files Needing Attention: No files require special attention. The otel-collector-config.yaml namespace removal warrants a local dashboard audit if Prometheus queries exist against the old agentex_db_*-prefixed series.

Important Files Changed

Filename	Overview
agentex/src/utils/rpc_metrics.py	New module: RPC RED metrics and workflow-duration histogram with correct OTel semconv, never-raise emission, and single-record-per-request guarantee via contextmanager.
agentex/src/utils/db_metrics.py	Adds InstrumentedAsyncAdaptedQueuePool and three new OTel connection-acquisition metrics; pending_requests increment occurs outside the try block, so an OTel SDK raise there leaves the counter unbalanced without a compensating decrement.
agentex/src/api/routes/agents.py	Wraps sync and streaming RPC handlers with rpc_request_timing; streaming handler correctly carries instrumentation across yields and GeneratorExit.
agentex/src/config/dependencies.py	Adds poolclass=InstrumentedAsyncAdaptedQueuePool to all three engine constructors; straightforward wiring change.
agentex/otel/otel-collector-config.yaml	Removes namespace: agentex from Prometheus exporter to prevent double-prefixing for agentex.* metrics; renames existing local db.* metric series as a side effect.
agentex/tests/unit/utils/test_rpc_metrics.py	Comprehensive unit tests covering no-op, never-raise, attribute correctness, double-count regression, GeneratorExit, and handler-classification precedence.
agentex/tests/unit/utils/test_db_metrics.py	Unit tests for the pool instrumentation covering passthrough, success, timeout, non-timeout error, recreate propagation, and end-to-end wiring.

Sequence Diagram

sequenceDiagram
    participant Client
    participant agents_py as agents.py
    participant rpc_timing as rpc_request_timing
    participant use_case as agents_acp_use_case
    participant otel as OTel SDK

    Client->>agents_py: "POST /agents/{id}/rpc"
    agents_py->>rpc_timing: __enter__ (start timer)
    rpc_timing-->>agents_py: RpcCallOutcome

    alt Sync request
        agents_py->>use_case: handle_rpc_request()
        use_case-->>agents_py: result_entity
        agents_py->>rpc_timing: __exit__ (normal)
        rpc_timing->>otel: record duration/requests
        agents_py-->>Client: AgentRPCResponse (200)
    else Streaming request
        agents_py->>use_case: handle_rpc_request()
        use_case-->>agents_py: AsyncIterator
        loop Each chunk
            agents_py-->>Client: NDJSON chunk (yield)
        end
        alt Stream error
            agents_py->>rpc_timing: rpc_call.fail(-32603, e)
            agents_py-->>Client: error frame (yield)
        end
        agents_py->>rpc_timing: __exit__ (GeneratorExit or normal)
        rpc_timing->>otel: record duration/requests/errors
    end

    Note over agents_py,otel: task/create also emits gen_ai.workflow.duration

Loading

_{Reviews (6): Last reviewed commit: "Merge branch 'main' into stephen-wang/rp..." | Re-trigger Greptile}

[stephen-wang24]

@greptileai

[harvhan]

record_rpc_request fires before AgentRPCResponse.model_validate on the success path. If that validation raises, the except ValidationError branch records the same request a second time as a -32602 error, so agentex.rpc.requests and the duration histogram count it twice. Suggest validating first, then recording:

response = AgentRPCResponse.model_validate(
    {"id": request.id, "result": serialized_result, "error": None}
)
record_rpc_request(
    method=request.method.value,
    streaming=False,
    duration_s=time.perf_counter() - start,
)
return response

[stephen-wang24]

good point, will record it after the validation to avoid double metricing

[stephen-wang24]

Fixed in 15ba465. Recording now goes through a rpc_request_timing context manager that records exactly once when the handler scope exits, and model_validate runs inside that scope, if it raises, the except ValidationError branch classifies the call as -32602 and that's the single record. Double counting is structurally impossible now.

[harvhan]

nit: a client disconnect mid-stream raises GeneratorExit (or CancelledError) at the yield, which skips the except Exception branch, so this finally records the request as a success with a truncated duration. That deflates streaming duration percentiles and inflates the success rate. Suggest tracking error_code and error_type separately so disconnects carry error.type without counting as JSON-RPC errors (record_rpc_request already supports that combination, since the error counter is driven only by error_code):

error_type: str | None = None
error_code: int | None = None
...
except (GeneratorExit, asyncio.CancelledError) as e:
    # Client went away mid-stream: tag it, but not a JSON-RPC error.
    error_type = type(e).__name__
    raise
except Exception as e:
    error_type = type(e).__name__
    error_code = -32603
    ...
finally:
    record_rpc_request(
        ...,
        error_code=error_code,
        error_type=error_type,
    )

That also lets dashboards exclude disconnects from percentiles via error.type, and drops the None if error_type is None else -32603 derivation.

[NiteshDhanpal]

I guess we should measure how long a connection is held.. So, here we are measuring the wait to acquire a connection but not how long it's been held. Pool exhaustion is a function of both i.e how long requests queue and how long connections stay checked out once acquired

[stephen-wang24]

Yes, hold time is covered above in db.client.connection.use_time

[NiteshDhanpal]

So we're only capturing timeouts? Timeout is good but connect can fail for numerous reasons right?. Say, what if Postgres is rejecting a connection due to max connections already reached on the DB? Or there's a network drop? It's neither a timeout nor a successful wait, so it just goes dark.. worth adding a catch-all failure path for that.

[stephen-wang24]

Good point, i was adding for metrics in semconv and failure is not one of it, added in 15ba465 for emitting generic failures

[NiteshDhanpal]

Nice!

[NiteshDhanpal]

Could we pull this into a @contextmanager that handles start/end timing and records the metric (including error classification) in one place, rather than each handler managing start = time.perf_counter() and calling record_rpc_request at every exit point? That would let us reuse it across the library instead of repeating the same start/duration/record pattern in every function that needs rpc timing

[stephen-wang24]

Sounds good. Done in 15ba465 to use a context manager