feat(gax): add logNetworkAttempt to LoggerHelper

Author: shivanee-p

core/packages/gax/src/observability/LoggerHelper.ts

@@ -33,6 +33,7 @@ import {
   resolveExceptionMessage,
   resolveExceptionStacktrace,
   resolveExceptionType,
+  resolveResponsePayload,
   resolveResponseStatusCode,
 } from './errorUtil';
 import {getPromiseTarget, handlePromise, handleStream} from './TracerHelper';
@@ -57,6 +58,22 @@ export interface LogPayload {
   'exception.stacktrace'?: string;
 }
 
+/**
+ * Resolves a `LogSeverity` value across `google-logging-utils` versions where
+ * `LogSeverity` may be exported as a type only.
+ */
+function resolveLogSeverity(
+  level: 'DEBUG' | 'INFO' | 'WARNING' | 'ERROR',
+): loggingUtils.LogSeverity {
+  return (
+    (
+      loggingUtils as {
+        LogSeverity?: Record<string, loggingUtils.LogSeverity>;
+      }
+    ).LogSeverity?.[level] ?? (level as loggingUtils.LogSeverity)
+  );
+}
+
 /**
  * Returns the `google-logging-utils` logger for the given GCP client service,
  * defaulting to the `'google-gax'` namespace when no service name is provided.
@@ -125,6 +142,48 @@ export function buildLogAttributes(
   return attributes;
 }
 
+/**
+ * Executes `fn` and attaches stream, promise, callback, or synchronous error
+ * handlers to invoke `onError` when the operation fails.
+ */
+function executeLoggedOperation<T>(
+  fn: (loggedCallback?: APICallback) => T,
+  onError: (err: unknown) => void,
+  isStreamCall: boolean,
+  userCallback?: APICallback,
+): T {
+  // Wrap user callback to log RPC errors before forwarding.
+  let callbackInvoked = false;
+  const loggedCallback: APICallback | undefined = userCallback
+    ? function (this: unknown, ...args: Parameters<APICallback>) {
+        callbackInvoked = true;
+        const err = args[0];
+        if (err) {
+          onError(err);
+        }
+        userCallback.apply(this, args);
+      }
+    : undefined;
+
+  // Execute call and attach error handlers for streams, promises, or sync throws.
+  try {
+    const result = fn(loggedCallback);
+    const promiseTarget = !isStreamCall ? getPromiseTarget(result) : null;
+    if (isStreamCall && result instanceof EventEmitter) {
+      handleStream(result, onError, () => {}, Boolean(loggedCallback));
+    } else if (promiseTarget) {
+      handlePromise(promiseTarget, onError, () => {});
+    }
+    return result;
+  } catch (syncErr) {
+    // Avoid logging errors thrown inside userCallback after RPC completion.
+    if (!callbackInvoked) {
+      onError(syncErr);
+    }
+    throw syncErr;
+  }
+}
+
 /**
  * Executes an API call and emits a single `WARNING` log entry if the logical
  * request fails across promise, callback, stream, or synchronous execution boundaries.
@@ -179,17 +238,10 @@ export function logCall<T = GaxCallResult>(
       ...(stacktrace !== undefined ? {'exception.stacktrace': stacktrace} : {}),
     };
 
-    const warningSeverity =
-      (
-        loggingUtils as {
-          LogSeverity?: {WARNING: loggingUtils.LogSeverity};
-        }
-      ).LogSeverity?.WARNING ?? ('WARNING' as loggingUtils.LogSeverity);
-
     // Emit structured WARNING log.
     logger(
       {
-        severity: warningSeverity,
+        severity: resolveLogSeverity('WARNING'),
         ...(traceId ? {telemetryTraceId: traceId} : {}),
         ...(spanId ? {telemetrySpanId: spanId} : {}),
         other: attributes,
@@ -199,34 +251,65 @@ export function logCall<T = GaxCallResult>(
     );
   };
 
-  // Wrap user callback to log RPC errors before forwarding.
-  let callbackInvoked = false;
-  const loggedCallback: APICallback | undefined = userCallback
-    ? function (this: unknown, ...args: Parameters<APICallback>) {
-        callbackInvoked = true;
-        const err = args[0];
-        if (err) {
-          emitL3WarnLog(err);
-        }
-        userCallback.apply(this, args);
-      }
-    : undefined;
+  return executeLoggedOperation(fn, emitL3WarnLog, isStreamCall, userCallback);
+}
 
-  // Execute call and attach error handlers for streams, promises, or sync throws.
-  try {
-    const result = fn(loggedCallback);
-    const promiseTarget = !isStreamCall ? getPromiseTarget(result) : null;
-    if (isStreamCall && result instanceof EventEmitter) {
-      handleStream(result, emitL3WarnLog, () => {}, Boolean(loggedCallback));
-    } else if (promiseTarget) {
-      handlePromise(promiseTarget, emitL3WarnLog, () => {});
-    }
-    return result;
-  } catch (syncErr) {
-    // Avoid logging errors thrown inside userCallback after RPC completion.
-    if (!callbackInvoked) {
-      emitL3WarnLog(syncErr);
+/**
+ * Executes an individual RPC transport attempt and emits a `DEBUG` log entry
+ * if the attempt fails across promise, callback, stream, or synchronous execution boundaries.
+ *
+ * @template T
+ * @param {AttemptTraceContext} attemptDynamicArgs - Dynamic context for the RPC attempt.
+ * @param {StaticTraceContext} staticArgs - Static context for the client library.
+ * @param {function} fn - The transport attempt operation to execute. Receives the wrapped
+ *   callback when `userCallback` is supplied, otherwise `undefined`.
+ * @param {boolean} [isStreamCall=false] - Whether the operation returns a stream (`EventEmitter`).
+ * @param {APICallback} [userCallback] - Optional attempt callback.
+ * @returns {T} The return value of `fn`.
+ */
+export function logNetworkAttempt<T = GaxCallResult>(
+  attemptDynamicArgs: AttemptTraceContext,
+  staticArgs: StaticTraceContext,
+  fn: (loggedCallback?: APICallback) => T,
+  isStreamCall = false,
+  userCallback?: APICallback,
+): T {
+  // Snapshot attempt context in case async rejection loses active context.
+  const initialContext = context.active();
+
+  const emitL4DebugLog = (err: unknown) => {
+    // Resolve logger and active T4 attempt span context for this attempt.
+    const logger = getGaxLogger(staticArgs?.gcpClientService);
+    const attemptSpan =
+      getActiveTraceContext()?.attemptSpan ??
+      getActiveTraceContext(initialContext)?.attemptSpan;
+    const spanContext = attemptSpan?.spanContext();
+    const traceId = spanContext?.traceId;
+    const spanId = spanContext?.spanId;
+
+    // Build structured log attributes and attach attempt metadata payload.
+    const attributes = buildLogAttributes(err, attemptDynamicArgs, staticArgs);
+    const responsePayload = resolveResponsePayload(
+      (err as {metadata?: unknown} | null | undefined)?.metadata,
+    );
+    if (responsePayload !== undefined) {
+      attributes['response.payload'] = responsePayload;
     }
-    throw syncErr;
-  }
+
+    // Emit structured DEBUG log for the failed attempt.
+    const message = resolveExceptionMessage(err, attemptDynamicArgs.rpcType);
+    logger(
+      {
+        severity: resolveLogSeverity('DEBUG'),
+        ...(traceId ? {telemetryTraceId: traceId} : {}),
+        ...(spanId ? {telemetrySpanId: spanId} : {}),
+        other: attributes,
+      },
+      'RPC attempt failed: %s',
+      message || err,
+    );
+  };
+
+  // Intercept and dispatch execution boundaries for the attempt.
+  return executeLoggedOperation(fn, emitL4DebugLog, isStreamCall, userCallback);
 }

core/packages/gax/test/unit/loggerHelper.ts

@@ -23,6 +23,7 @@ import {
   buildLogAttributes,
   getGaxLogger,
   logCall,
+  logNetworkAttempt,
   LogPayload,
 } from '../../src/observability/LoggerHelper';
 import {traceAttempt, traceCall} from '../../src/observability/TracerHelper';
@@ -482,4 +483,148 @@ describe('LoggerHelper', () => {
       );
     });
   });
+
+  describe('logNetworkAttempt', () => {
+    it('does not log when an attempt succeeds across promise, callback, stream, or sync boundaries', async () => {
+      const asyncResult = await logNetworkAttempt(
+        dynamicArgs,
+        staticArgs,
+        async () => ({ok: true}),
+      );
+      assert.deepStrictEqual(asyncResult, {ok: true});
+
+      const syncResult = logNetworkAttempt(dynamicArgs, staticArgs, () => 42);
+      assert.strictEqual(syncResult, 42);
+
+      let callbackResult: unknown;
+      logNetworkAttempt(
+        dynamicArgs,
+        staticArgs,
+        loggedCallback => {
+          loggedCallback!(null, {cb: true});
+          return undefined as unknown as ResultTuple;
+        },
+        false,
+        (_err, res) => {
+          callbackResult = res;
+        },
+      );
+      assert.deepStrictEqual(callbackResult, {cb: true});
+
+      const stream = new EventEmitter();
+      logNetworkAttempt(dynamicArgs, staticArgs, () => stream, true);
+      stream.emit('end');
+
+      assert.strictEqual(sink.logs.length, 0);
+    });
+
+    it('emits a DEBUG log with response.payload when err.metadata is present and omits it when absent', async () => {
+      const errWithMeta = Object.assign(new GoogleError('Unavailable'), {
+        code: Status.UNAVAILABLE,
+        metadata: {'x-goog-request-id': 'req-123'},
+      });
+
+      await assert.rejects(
+        async () => {
+          await logNetworkAttempt(dynamicArgs, staticArgs, async () => {
+            throw errWithMeta;
+          });
+        },
+        (caught: unknown) => caught === errWithMeta,
+      );
+
+      assert.strictEqual(sink.logs.length, 1);
+      const entry = sink.logs[0];
+      assert.strictEqual(entry.namespace, 'storage');
+      assert.strictEqual(entry.fields.severity, 'DEBUG');
+      assert.deepStrictEqual(entry.args, [
+        'RPC attempt failed: %s',
+        'Unavailable',
+      ]);
+      const other = entry.fields.other as LogPayload;
+      assert.deepStrictEqual(other, {
+        'gcp.client.service': 'storage',
+        'gcp.client.version': '1.2.3',
+        'gcp.client.repo': DEFAULT_GCP_REPO,
+        'gcp.client.artifact': '@google-cloud/storage',
+        'gcp.client.language': 'node',
+        'rpc.system.name': 'grpc',
+        'rpc.method': 'google.storage.v2.Storage/GetObject',
+        'url.domain': 'storage.googleapis.com',
+        'rpc.response.status_code': 'UNAVAILABLE',
+        'error.type': 'UNAVAILABLE',
+        'response.payload': '{"x-goog-request-id":"req-123"}',
+      });
+      assert.strictEqual('exception.type' in other, false);
+      assert.strictEqual('exception.message' in other, false);
+      assert.strictEqual('exception.stacktrace' in other, false);
+
+      sink.reset();
+      const errWithoutMeta = new TypeError('client error');
+      assert.throws(
+        () =>
+          logNetworkAttempt(dynamicArgs, staticArgs, () => {
+            throw errWithoutMeta;
+          }),
+        errWithoutMeta,
+      );
+
+      assert.strictEqual(sink.logs.length, 1);
+      assert.strictEqual(sink.logs[0].fields.severity, 'DEBUG');
+      assert.deepStrictEqual(sink.logs[0].args, [
+        'RPC attempt failed: %s',
+        'client error',
+      ]);
+      const otherWithoutMeta = sink.logs[0].fields.other as LogPayload;
+      assert.strictEqual('response.payload' in otherWithoutMeta, false);
+      assert.strictEqual('exception.type' in otherWithoutMeta, false);
+      assert.strictEqual('exception.message' in otherWithoutMeta, false);
+      assert.strictEqual('exception.stacktrace' in otherWithoutMeta, false);
+    });
+
+    it('captures the attempt span context and logs on callback and stream failures', () => {
+      const stream = new EventEmitter();
+      let receivedCallback: APICallback | undefined;
+
+      traceCall(dynamicArgs, staticArgs, () =>
+        traceAttempt(dynamicArgs, staticArgs, () =>
+          logNetworkAttempt(
+            dynamicArgs,
+            staticArgs,
+            loggedCallback => {
+              receivedCallback = loggedCallback;
+              return stream;
+            },
+            true,
+            () => {},
+          ),
+        ),
+      );
+
+      const attemptErr = new GoogleError('attempt failed');
+      receivedCallback!(attemptErr);
+      stream.emit('error', attemptErr);
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 2);
+      const attemptSpan = spans[0];
+
+      assert.strictEqual(sink.logs.length, 2);
+      for (const logEntry of sink.logs) {
+        assert.strictEqual(logEntry.fields.severity, 'DEBUG');
+        assert.strictEqual(
+          logEntry.fields.telemetryTraceId,
+          attemptSpan.spanContext().traceId,
+        );
+        assert.strictEqual(
+          logEntry.fields.telemetrySpanId,
+          attemptSpan.spanContext().spanId,
+        );
+        assert.deepStrictEqual(logEntry.args, [
+          'RPC attempt failed: %s',
+          'attempt failed',
+        ]);
+      }
+    });
+  });
 });