feat: track json-rpc metrics

yarn-project/aztec-node/src/bin/index.ts

@@ -5,7 +5,11 @@ import {
   startHttpRpcServer,
 } from '@aztec/foundation/json-rpc/server';
 import { createLogger } from '@aztec/foundation/log';
-import { getOtelJsonRpcPropagationMiddleware } from '@aztec/telemetry-client';
+import {
+  getOtelJsonRpcDiagnosticsMiddleware,
+  getOtelJsonRpcPropagationMiddleware,
+  getOtelJsonRpcServerMetricsMiddleware,
+} from '@aztec/telemetry-client';
 
 import {
   type AztecNodeConfig,
@@ -49,7 +53,8 @@ async function main() {
   const services: NamespacedApiHandlers = {};
   registerAztecNodeRpcHandlers(aztecNode, services);
   const rpcServer = createNamespacedSafeJsonRpcServer(services, {
-    middlewares: [getOtelJsonRpcPropagationMiddleware()],
+    diagnostic: getOtelJsonRpcDiagnosticsMiddleware(),
+    middlewares: [getOtelJsonRpcServerMetricsMiddleware(), getOtelJsonRpcPropagationMiddleware()],
   });
   await startHttpRpcServer(rpcServer, { port: +AZTEC_NODE_PORT, apiPrefix: API_PREFIX });
   logger.info(`Aztec Node JSON-RPC Server listening on port ${AZTEC_NODE_PORT}`);

yarn-project/aztec/src/cli/aztec_start_action.ts

@@ -10,7 +10,11 @@ import type { LogFn, Logger } from '@aztec/foundation/log';
 import type { ChainConfig } from '@aztec/stdlib/config';
 import { getPackageVersion } from '@aztec/stdlib/update-checker';
 import { getVersioningMiddleware } from '@aztec/stdlib/versioning';
-import { getOtelJsonRpcDiagnosticsMiddleware, getOtelJsonRpcPropagationMiddleware } from '@aztec/telemetry-client';
+import {
+  getOtelJsonRpcDiagnosticsMiddleware,
+  getOtelJsonRpcPropagationMiddleware,
+  getOtelJsonRpcServerMetricsMiddleware,
+} from '@aztec/telemetry-client';
 
 import { createLocalNetwork } from '../local-network/index.js';
 import { github, splash } from '../splash.js';
@@ -93,7 +97,11 @@ export async function aztecStart(options: any, userLog: LogFn, debugLogger: Logg
       diagnostic: getOtelJsonRpcDiagnosticsMiddleware(),
       http200OnError: false,
       log: debugLogger,
-      middlewares: [getOtelJsonRpcPropagationMiddleware(), getVersioningMiddleware(versions, versioningOpts)],
+      middlewares: [
+        getOtelJsonRpcServerMetricsMiddleware(),
+        getOtelJsonRpcPropagationMiddleware(),
+        getVersioningMiddleware(versions, versioningOpts),
+      ],
       maxBatchSize: options.rpcMaxBatchSize,
       maxBodySizeBytes: options.rpcMaxBodySize,
     });
@@ -103,7 +111,11 @@ export async function aztecStart(options: any, userLog: LogFn, debugLogger: Logg
 
   // If there are any admin services, start a separate JSON-RPC server for them
   if (Object.entries(adminServices).length > 0) {
-    const adminMiddlewares = [getOtelJsonRpcPropagationMiddleware(), getVersioningMiddleware(versions, versioningOpts)];
+    const adminMiddlewares = [
+      getOtelJsonRpcServerMetricsMiddleware(),
+      getOtelJsonRpcPropagationMiddleware(),
+      getVersioningMiddleware(versions, versioningOpts),
+    ];
 
     // Resolve the admin API key (auto-generated and persisted, or opt-out)
     const apiKeyResolution = await resolveAdminApiKey(
@@ -116,7 +128,7 @@ export async function aztecStart(options: any, userLog: LogFn, debugLogger: Logg
       debugLogger,
     );
     if (apiKeyResolution) {
-      adminMiddlewares.unshift(getApiKeyAuthMiddleware(apiKeyResolution.apiKeyHash));
+      adminMiddlewares.splice(1, 0, getApiKeyAuthMiddleware(apiKeyResolution.apiKeyHash));
     } else {
       debugLogger.warn('No admin API key set — admin endpoint is unauthenticated');
     }

yarn-project/foundation/src/json-rpc/server/safe_json_rpc_server.test.ts

@@ -118,6 +118,31 @@ describe('SafeJsonRpcServer', () => {
       expect(calls).toEqual(['start:count:42:test-value', 'end:count']);
     });
 
+    it('reports request validation duration and outcome to diagnostics', async () => {
+      const validations: Array<{ durationMs: number | undefined; succeeded: boolean | undefined }> = [];
+      server = createSafeJsonRpcServer<TestStateApi>(testState, TestStateSchema, {
+        diagnostic: async (ctx, next) => {
+          try {
+            await next();
+          } finally {
+            validations.push({
+              durationMs: ctx.requestValidationDurationMs,
+              succeeded: ctx.requestValidationSucceeded,
+            });
+          }
+        },
+      });
+
+      await send({ method: 'count', params: [] });
+      await send({ method: 'getNote', params: ['invalid'] });
+
+      expect(validations).toHaveLength(2);
+      expect(validations[0]?.durationMs).toBeGreaterThanOrEqual(0);
+      expect(validations[0]?.succeeded).toBe(true);
+      expect(validations[1]?.durationMs).toBeGreaterThanOrEqual(0);
+      expect(validations[1]?.succeeded).toBe(false);
+    });
+
     it('runs diagnostics for each request in a batch', async () => {
       const methods: string[] = [];
       server = createSafeJsonRpcServer<TestStateApi>(testState, TestStateSchema, {

yarn-project/foundation/src/json-rpc/server/safe_json_rpc_server.ts

@@ -17,6 +17,7 @@ import {
   parseWithOptionals,
   schemaHasMethod,
 } from '../../schemas/index.js';
+import { Timer } from '../../timer/index.js';
 import { jsonStringify } from '../convert.js';
 import { assert } from '../js_utils.js';
 
@@ -25,6 +26,8 @@ export type DiagnosticsData = {
   method: string;
   params: any[];
   headers: http.IncomingHttpHeaders;
+  requestValidationDurationMs?: number;
+  requestValidationSucceeded?: boolean;
 };
 
 export type DiagnosticsMiddleware = (ctx: DiagnosticsData, next: () => Promise<void>) => Promise<void>;
@@ -217,8 +220,12 @@ export class SafeJsonRpcServer {
         let result: any;
 
         if (this.diagnosticsMiddleware) {
-          await this.diagnosticsMiddleware({ id: id ?? null, method, params, headers }, async () => {
-            result = await this.proxy.call(method, params);
+          const diagnosticsData: DiagnosticsData = { id: id ?? null, method, params, headers };
+          await this.diagnosticsMiddleware(diagnosticsData, async () => {
+            result = await this.proxy.call(method, params, (durationMs, succeeded) => {
+              diagnosticsData.requestValidationDurationMs = durationMs;
+              diagnosticsData.requestValidationSucceeded = succeeded;
+            });
           });
         } else {
           result = await this.proxy.call(method, params);
@@ -299,7 +306,11 @@ export type StatusCheckFn = () => boolean | Promise<boolean>;
 
 interface Proxy {
   hasMethod(methodName: string): boolean;
-  call(methodName: string, jsonParams?: any[]): Promise<any>;
+  call(
+    methodName: string,
+    jsonParams?: any[],
+    onRequestValidated?: (durationMs: number, succeeded: boolean) => void,
+  ): Promise<any>;
 }
 
 /**
@@ -323,14 +334,26 @@ export class SafeJsonProxy<T extends object = any> implements Proxy {
    * @param jsonParams - The RPC parameters.
    * @returns The remote result.
    */
-  public async call(methodName: string, jsonParams: any[] = []) {
+  public async call(
+    methodName: string,
+    jsonParams: any[] = [],
+    onRequestValidated?: (durationMs: number, succeeded: boolean) => void,
+  ) {
     this.log.debug(format(`request`, methodName, jsonParams));
 
     assert(Array.isArray(jsonParams), `Params to ${methodName} is not an array: ${jsonParams}`);
     assert(schemaHasMethod(this.schema, methodName), `Method ${methodName} not found in schema`);
     const method = this.handler[methodName as keyof T];
     assert(typeof method === 'function', `Method ${methodName} is not a function`);
-    const args = await parseWithOptionals(jsonParams, getSchemaParameters(this.schema[methodName]));
+    const validationTimer = new Timer();
+    let args: any[];
+    try {
+      args = await parseWithOptionals(jsonParams, getSchemaParameters(this.schema[methodName]));
+      onRequestValidated?.(validationTimer.ms(), true);
+    } catch (error) {
+      onRequestValidated?.(validationTimer.ms(), false);
+      throw error;
+    }
     const ret = await method.apply(this.handler, args);
     this.log.debug(format('response', methodName, ret));
     return ret;
@@ -350,12 +373,16 @@ class NamespacedSafeJsonProxy implements Proxy {
     }
   }
 
-  public call(namespacedMethodName: string, jsonParams: any[] = []) {
+  public call(
+    namespacedMethodName: string,
+    jsonParams: any[] = [],
+    onRequestValidated?: (durationMs: number, succeeded: boolean) => void,
+  ) {
     const [namespace, methodName] = namespacedMethodName.split('_', 2);
     assert(namespace && methodName, `Invalid namespaced method name: ${namespacedMethodName}`);
     const handler = this.proxies[namespace];
     assert(handler, `Namespace not found: ${namespace}`);
-    return handler.call(methodName, jsonParams);
+    return handler.call(methodName, jsonParams, onRequestValidated);
   }
 
   public hasMethod(namespacedMethodName: string): boolean {

yarn-project/telemetry-client/src/attributes.ts

@@ -18,6 +18,8 @@
 export const HTTP_REQUEST_HOST = 'http.header.request.host';
 export const HTTP_RESPONSE_STATUS_CODE = 'http.response.status_code';
 
+export const JSON_RPC_REJECTION_REASON = 'aztec.json_rpc.rejection_reason';
+
 /** The Aztec network identifier */
 export const NETWORK_NAME = 'aztec.network_name';
 

yarn-project/telemetry-client/src/index.ts

@@ -8,3 +8,4 @@ export * from './l1_metrics.js';
 export * from './wrappers/index.js';
 export * from './start.js';
 export * from './otel_propagation.js';
+export * from './json_rpc_server_metrics.js';

yarn-project/telemetry-client/src/json_rpc_server_metrics.ts

@@ -0,0 +1,148 @@
+import { Timer } from '@aztec/foundation/timer';
+
+import type Koa from 'koa';
+
+import * as Attributes from './attributes.js';
+import * as Metrics from './metrics.js';
+import { getTelemetryClient } from './start.js';
+import type { Histogram, TelemetryClient, UpDownCounter } from './telemetry.js';
+import { ATTR_JSONRPC_METHOD, ATTR_JSONRPC_SERVICE } from './vendor/attributes.js';
+
+const BATCH_SIZE_BUCKETS = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000];
+
+/** Fixed reasons for rejecting an RPC request before dispatching it to a registered handler. */
+export type JsonRpcRejectionReason =
+  | 'unauthorized'
+  | 'parse_error'
+  | 'invalid_request'
+  | 'method_not_found'
+  | 'bad_request'
+  | 'internal_error';
+
+/** Records bounded-cardinality metrics for registered JSON-RPC calls, rejected requests, and batches. */
+export class JsonRpcServerMetrics {
+  private readonly requestCount: UpDownCounter;
+  private readonly requestDuration: Histogram;
+  private readonly requestValidationDuration: Histogram;
+  private readonly rejectedRequestCount: UpDownCounter;
+  private readonly batchCount: UpDownCounter;
+  private readonly batchDuration: Histogram;
+  private readonly batchSize: Histogram;
+
+  constructor(telemetry: TelemetryClient) {
+    const meter = telemetry.getMeter('JsonRpcServer');
+    this.requestCount = meter.createUpDownCounter(Metrics.JSON_RPC_SERVER_REQUEST_COUNT);
+    this.requestDuration = meter.createHistogram(Metrics.JSON_RPC_SERVER_REQUEST_DURATION);
+    this.requestValidationDuration = meter.createHistogram(Metrics.JSON_RPC_SERVER_REQUEST_VALIDATION_DURATION);
+    this.rejectedRequestCount = meter.createUpDownCounter(Metrics.JSON_RPC_SERVER_REJECTED_REQUEST_COUNT);
+    this.batchCount = meter.createUpDownCounter(Metrics.JSON_RPC_SERVER_BATCH_COUNT);
+    this.batchDuration = meter.createHistogram(Metrics.JSON_RPC_SERVER_BATCH_DURATION);
+    this.batchSize = meter.createHistogram(Metrics.JSON_RPC_SERVER_BATCH_SIZE, {
+      advice: { explicitBucketBoundaries: BATCH_SIZE_BUCKETS },
+    });
+  }
+
+  /** Records the outcome and handler duration of a registered RPC method. */
+  public recordRequest(fullMethod: string, durationMs: number, ok: boolean): void {
+    const [service, method] = splitJsonRpcMethod(fullMethod);
+    const attributes = {
+      ...(service === undefined ? {} : { [ATTR_JSONRPC_SERVICE]: service }),
+      [ATTR_JSONRPC_METHOD]: method,
+      [Attributes.OK]: ok,
+    };
+    this.requestCount.add(1, attributes);
+    this.requestDuration.record(durationMs, attributes);
+  }
+
+  /** Records a pre-dispatch rejection using a fixed reason. */
+  public recordRejectedRequest(reason: JsonRpcRejectionReason): void {
+    this.rejectedRequestCount.add(1, { [Attributes.JSON_RPC_REJECTION_REASON]: reason });
+  }
+
+  /** Records the duration and outcome of validating a registered RPC method's parameters. */
+  public recordRequestValidation(fullMethod: string, durationMs: number, ok: boolean): void {
+    const [service, method] = splitJsonRpcMethod(fullMethod);
+    this.requestValidationDuration.record(durationMs, {
+      ...(service === undefined ? {} : { [ATTR_JSONRPC_SERVICE]: service }),
+      [ATTR_JSONRPC_METHOD]: method,
+      [Attributes.OK]: ok,
+    });
+  }
+
+  /** Records the outcome, processing duration, and number of calls in a batch envelope. */
+  public recordBatch(size: number, durationMs: number, ok: boolean): void {
+    const attributes = { [Attributes.OK]: ok };
+    this.batchCount.add(1, attributes);
+    this.batchDuration.record(durationMs, attributes);
+    this.batchSize.record(size, attributes);
+  }
+}
+
+let metricsOwner: TelemetryClient | undefined;
+let metrics: JsonRpcServerMetrics | undefined;
+
+export function getJsonRpcServerMetrics(): JsonRpcServerMetrics {
+  const telemetry = getTelemetryClient();
+  if (metricsOwner !== telemetry) {
+    metricsOwner = telemetry;
+    metrics = new JsonRpcServerMetrics(telemetry);
+  }
+  return metrics!;
+}
+
+export function getOtelJsonRpcServerMetricsMiddleware(
+  metricsProvider: () => JsonRpcServerMetrics = getJsonRpcServerMetrics,
+): (ctx: Koa.Context, next: () => Promise<void>) => Promise<void> {
+  return async function otelJsonRpcServerMetrics(ctx, next) {
+    const timer = new Timer();
+    await next();
+
+    const requestBody = (ctx.request as { body?: unknown }).body;
+    if (Array.isArray(requestBody)) {
+      metricsProvider().recordBatch(requestBody.length, timer.ms(), Array.isArray(ctx.body));
+    }
+
+    for (const reason of getRejectionReasons(ctx.status, ctx.body)) {
+      metricsProvider().recordRejectedRequest(reason);
+    }
+  };
+}
+
+export function splitJsonRpcMethod(fullMethod: string): [service: string | undefined, method: string] {
+  const separator = fullMethod.indexOf('_');
+  return separator === -1 ? [undefined, fullMethod] : [fullMethod.slice(0, separator), fullMethod.slice(separator + 1)];
+}
+
+function getRejectionReasons(status: number, response: unknown): JsonRpcRejectionReason[] {
+  if (status === 401) {
+    return ['unauthorized'];
+  }
+
+  const responses = Array.isArray(response) ? response : [response];
+  return responses.flatMap(item => {
+    const code = getErrorCode(item);
+    if (code === -32700) {
+      return ['parse_error'];
+    }
+    if (code === -32601) {
+      return ['method_not_found'];
+    }
+    if (code === -32600) {
+      return [status >= 500 ? 'internal_error' : 'invalid_request'];
+    }
+    if (code === -32000) {
+      return ['bad_request'];
+    }
+    return [];
+  });
+}
+
+function getErrorCode(response: unknown): number | undefined {
+  if (!response || typeof response !== 'object' || !('error' in response)) {
+    return undefined;
+  }
+  const error = response.error;
+  return error && typeof error === 'object' && 'code' in error && typeof error.code === 'number'
+    ? error.code
+    : undefined;
+}