diff --git a/core/packages/gax/src/createApiCall.ts b/core/packages/gax/src/createApiCall.ts index e161879d5c9..80fbed668c5 100644 --- a/core/packages/gax/src/createApiCall.ts +++ b/core/packages/gax/src/createApiCall.ts @@ -33,6 +33,12 @@ import {retryable} from './normalCalls/retries'; import {addTimeoutArg} from './normalCalls/timeout'; import {StreamingApiCaller} from './streamingCalls/streamingApiCaller'; import {warn} from './warnings'; +import { + traceCall, + StaticTraceContext, + DynamicTraceContext, +} from './observability/TracerHelper'; +import {checkTelemetryEnabled} from './util'; /** * Converts an rpc call into an API call governed by the settings. @@ -66,6 +72,9 @@ export function createApiCall( const funcPromise = typeof func === 'function' ? Promise.resolve(func) : func; // the following apiCaller will be used for all calls of this function... const apiCaller = createAPICaller(settings, descriptor); + + const tracingEnabled = checkTelemetryEnabled(settings); + const invokeCall = ( request: RequestType, callOptions?: CallOptions, @@ -168,5 +177,45 @@ export function createApiCall( // or to cancel the ongoing call. return currentApiCaller.result(ongoingCall); }; - return invokeCall; + + if (tracingEnabled) { + const staticArgs: StaticTraceContext = { + gcpClientService: + settings.otherArgs?.internalTelemetryInfo?.gcpClientService, + gcpVersion: settings.otherArgs?.internalTelemetryInfo?.gcpVersion, + gcpRepo: settings.otherArgs?.internalTelemetryInfo?.gcpRepo, + gcpArtifact: settings.otherArgs?.internalTelemetryInfo?.gcpArtifact, + }; + + const serviceName = settings.apiName?.split('.').pop() ?? ''; + const isFallback = Boolean(_fallback); + const dynamicArgs: DynamicTraceContext = { + clientName: serviceName ? `${serviceName}Client` : '', + methodName: settings.otherArgs?.internalMethodName ?? '', + rpcType: isFallback ? 'http' : 'grpc', + }; + const isStreamingCall = apiCaller instanceof StreamingApiCaller; + return ( + request: RequestType, + callOptions?: CallOptions, + callback?: APICallback, + ) => { + return traceCall( + dynamicArgs, + staticArgs, + (tracedCallback?: APICallback) => { + // `traceCall` wraps the user's callback whenever one was supplied, + // for stream and non-stream calls alike, and that wrapper is what + // closes the span. It is undefined only when there is no callback to + // wrap, in which case the span is bound to the returned promise or + // stream instead; the fallback keeps this correct either way. + return invokeCall(request, callOptions, tracedCallback ?? callback); + }, + isStreamingCall, + callback, + ); + }; + } else { + return invokeCall; + } } diff --git a/core/packages/gax/src/fallback.ts b/core/packages/gax/src/fallback.ts index 32d122b3a67..fa9d154348e 100644 --- a/core/packages/gax/src/fallback.ts +++ b/core/packages/gax/src/fallback.ts @@ -447,7 +447,7 @@ export function createApiCall( ); }; } - return _createApiCall(func, settings, descriptor); + return _createApiCall(func, settings, descriptor, true); } export {protobuf}; diff --git a/core/packages/gax/src/fallbackRest.ts b/core/packages/gax/src/fallbackRest.ts index b6e5f7621ea..14c9cef23e8 100644 --- a/core/packages/gax/src/fallbackRest.ts +++ b/core/packages/gax/src/fallbackRest.ts @@ -90,6 +90,7 @@ export function decodeResponse( rpc: protobuf.Method, ok: boolean, response: Buffer | ArrayBuffer, + httpStatusCode?: number, ): {} { // eslint-disable-next-line n/no-unsupported-features/node-builtins const decodedString = new TextDecoder().decode(response); @@ -99,6 +100,17 @@ export function decodeResponse( const json = JSON.parse(decodedString); if (!ok) { const error = GoogleError.parseHttpError(json); + // `parseHttpError` reads the status out of the response body and maps it + // onto the gRPC `code`, keeping no record of the status the transport + // actually received — and the body's status can differ from it, or be + // missing entirely. Record the received one when the caller knows it. + // + // Optional because `decodeResponse` is also called from + // `streamArrayParser`, which only ever decodes an already-successful body + // and so has no status to pass. + if (httpStatusCode !== undefined) { + error.httpStatusCode = httpStatusCode; + } throw error; } const message = serializer.fromProto3JSON(rpc.resolvedResponseType!, json); diff --git a/core/packages/gax/src/fallbackServiceStub.ts b/core/packages/gax/src/fallbackServiceStub.ts index 119e3b3aab4..d58d4605297 100644 --- a/core/packages/gax/src/fallbackServiceStub.ts +++ b/core/packages/gax/src/fallbackServiceStub.ts @@ -14,7 +14,9 @@ * limitations under the License. */ -import type {Response as NodeFetchResponse} from 'node-fetch' with {'resolution-mode': 'import'}; +import type {Response as NodeFetchResponse} from 'node-fetch' with { + 'resolution-mode': 'import', +}; import {AuthClient, GoogleAuth, gaxios} from 'google-auth-library'; import * as serializer from 'proto3-json-serializer'; @@ -22,6 +24,8 @@ import * as serializer from 'proto3-json-serializer'; import {isNodeJS} from './featureDetection'; import {StreamArrayParser} from './streamArrayParser'; import {defaultToObjectOptions} from './fallback'; +import {GoogleError} from './googleError'; +import {Status} from './status'; import {pipeline, PipelineSource} from 'stream'; import type {Agent as HttpAgent} from 'http'; import type {Agent as HttpsAgent} from 'https'; @@ -33,8 +37,7 @@ import type {Agent as HttpsAgent} from 'https'; // - https://github.com/node-fetch/node-fetch#custom-agent // - https://github.com/googleapis/gax-nodejs/pull/1534 let agentOption: - | ((parsedUrl: {protocol: string}) => HttpAgent | HttpsAgent) - | null = null; + ((parsedUrl: {protocol: string}) => HttpAgent | HttpsAgent) | null = null; if (isNodeJS()) { const http = require('http'); const https = require('https'); @@ -82,6 +85,41 @@ function _formatEmptyResponse(rpc: protobuf.Method) { return resp; } +/** + * Reports an expired deadline the way the rest of gax expects. + * + * `retryCodes` matching, caller `err.code` comparisons and telemetry's + * `error.type` all key off the numeric gRPC status, and gRPC reports this + * condition as DEADLINE_EXCEEDED, so a REST deadline is surfaced identically + * rather than leaking a transport-specific error for a failure both transports + * share. The original error is kept as `cause`. + * + * Whether the deadline expired is passed in rather than inferred from `err`, + * because the error carries no usable evidence of it. Measured against a server + * that accepts a connection and never replies: node-fetch discards + * `signal.reason` and throws its own AbortError, gaxios wraps that in a + * GaxiosError which never sets `name` (so it stays the inherited 'Error') and + * only copies `code` from a DOMException cause, which this is not. The result + * is `name: 'Error'`, `code: undefined` — byte-identical to what `cancel()` + * produces. Only the caller, which armed the timer, knows which happened. + */ +function toDeadlineExceeded( + err: unknown, + rpcName: string, + timeoutMs: number | undefined, + timedOut: boolean, +): unknown { + if (!timedOut || timeoutMs === undefined) { + return err; + } + const error = new GoogleError( + `Deadline exceeded: ${rpcName} did not respond within ${timeoutMs} milliseconds.`, + {cause: err}, + ); + error.code = Status.DEADLINE_EXCEEDED; + return error; +} + export function generateServiceStub( rpcs: {[name: string]: protobuf.Method}, protocol: string, @@ -101,6 +139,7 @@ export function generateServiceStub( rpc: protobuf.Method, ok: boolean, response: Buffer | ArrayBuffer, + httpStatusCode?: number, ) => {}, numericEnums: boolean, minifyJson: boolean, @@ -112,13 +151,38 @@ export function generateServiceStub( }, }; for (const [rpcName, rpc] of Object.entries(rpcs)) { + // Named for what gax actually passes, which is its `UnaryCall` order: + // (request, metadata, options, callback). `FallbackServiceStub` declares + // the middle two the other way round, so the third argument — the one gRPC + // calls `options`, carrying the deadline — reads as metadata there and was + // long ignored as such. serviceStub[rpcName] = ( request: {}, - options?: {[name: string]: string}, - _metadata?: {} | Function, + metadata?: {[name: string]: string | string[]}, + callOptions?: {deadline?: Date}, callback?: Function, ) => { - options ??= {}; + metadata ??= {}; + + // `addTimeoutArg` sets a deadline on every call and `CallSettings.timeout` + // defaults to 30s, so one is essentially always present. gRPC enforces its + // own deadline, but nothing here ever read this one, so an endpoint that + // accepted the connection and then went quiet left the request — and the + // promise or callback waiting on it — outstanding forever. Convert it to + // the remaining duration and arm an abort signal with it below. + // + // Server-streaming RPCs are deliberately excluded. Their response is + // long-lived by design and the signal stays armed once the body starts + // flowing, so forwarding the deadline would abort a healthy stream + // mid-read. Bounding those is a separate, user-visible change. + let timeoutMs: number | undefined; + if (callOptions?.deadline && !rpc.responseStream) { + // `AbortSignal.timeout` rejects a negative delay with a RangeError, so + // an already-expired deadline is clamped. Zero is a fine value here: it + // aborts on the next tick, which is the right answer for a deadline + // that has already passed. + timeoutMs = Math.max(0, callOptions.deadline.getTime() - Date.now()); + } // We cannot use async-await in this function because we need to return the canceller object as soon as possible. // Using plain old promises instead. @@ -148,10 +212,40 @@ export function generateServiceStub( const cancelController = new AbortController(); const cancelSignal = cancelController.signal as AbortSignal; let cancelRequested = false; + + // Arm the deadline here rather than handing `timeout` to gaxios, which + // would build the identical `AbortSignal.timeout` internally. The + // difference is bookkeeping: both a deadline expiry and a `cancel()` + // abort the same request and surface the same error, so unless we record + // which one fired, the handlers below cannot tell them apart. + let timedOut = false; + let requestSignal = cancelSignal; + if (timeoutMs !== undefined) { + const timeoutSignal = AbortSignal.timeout(timeoutMs); + timeoutSignal.addEventListener('abort', () => (timedOut = true), { + once: true, + }); + requestSignal = AbortSignal.any([cancelSignal, timeoutSignal]); + } + const url = fetchParameters.url; const headers = new Headers(fetchParameters.headers); - for (const key of Object.keys(options)) { - headers.set(key, options[key][0]); + // gRPC metadata is multi-valued, and `buildMetadata` normalizes every + // value to an array for exactly that reason. This used to read + // `metadata[key][0]`, which dropped every value after the first and, for + // a value that was a plain string rather than an array, sent only its + // first character. Replace whatever the request encoder set, as the + // single-value `set` did, then keep all of the values. + for (const key of Object.keys(metadata)) { + const value = metadata[key]; + if (Array.isArray(value)) { + headers.delete(key); + for (const item of value) { + headers.append(key, String(item)); + } + } else { + headers.set(key, String(value)); + } } const streamArrayParser = new StreamArrayParser(rpc); let response204Ok = false; @@ -162,7 +256,7 @@ export function generateServiceStub( ? fetchParameters.body : Buffer.from(fetchParameters.body), method: fetchParameters.method, - signal: cancelSignal, + signal: requestSignal, responseType: 'stream', // ensure gaxios returns the data directly so that it handle data/streams itself agent: agentOption || undefined, }; @@ -206,21 +300,45 @@ export function generateServiceStub( ); return; } else { + // Captured here because the decoded value below is also named + // `response` and shadows the fetch response. + const httpStatusCode = response.status; return Promise.all([ Promise.resolve(response.ok), response.arrayBuffer(), ]) .then(([ok, buffer]: [boolean, Buffer | ArrayBuffer]) => { - const response = responseDecoder(rpc, ok, buffer); + const response = responseDecoder( + rpc, + ok, + buffer, + httpStatusCode, + ); callback!(null, response); + return; }) .catch((err: Error) => { - if (!cancelRequested || err.name !== 'AbortError') { + // The deadline can expire after the response headers arrive but + // before the body is fully read, which rejects here rather than + // in the outer handler. + const callErr = toDeadlineExceeded( + err, + rpcName, + timeoutMs, + timedOut, + ); + // A caller that cancelled does not need the resulting abort + // reported back to it, but a deadline always does. This used to + // test `err.name !== 'AbortError'`; gaxios wraps node-fetch's + // AbortError and never sets its own `name`, leaving the + // inherited 'Error', so the check never matched and cancelled + // calls still reported an error. Use the state we recorded. + if (timedOut || !cancelRequested) { if (rpc.responseStream) { if (callback) { - callback(err); + callback(callErr); } - streamArrayParser.emit('error', err); + streamArrayParser.emit('error', callErr); } else { // This supports a legacy Apiary behavior that allows // empty 204 responses. If we do not intercept this potential error @@ -232,7 +350,7 @@ export function generateServiceStub( if (!response204Ok) { // by this point, we're guaranteed to have added a callback // it is added in the library before calling this.innerApiCalls - callback!(err); + callback!(callErr); } else { const resp = _formatEmptyResponse(rpc); // by this point, we're guaranteed to have added a callback @@ -244,7 +362,10 @@ export function generateServiceStub( }); } }) - .catch((err: unknown) => { + .catch((rawErr: unknown) => { + // The usual timeout path: the deadline expired before any response + // was received, so the fetch itself rejects. + const err = toDeadlineExceeded(rawErr, rpcName, timeoutMs, timedOut); if (rpc.responseStream) { if (callback) { callback(err); diff --git a/core/packages/gax/src/googleError.ts b/core/packages/gax/src/googleError.ts index 42a64f60b1a..961fab6bdf7 100644 --- a/core/packages/gax/src/googleError.ts +++ b/core/packages/gax/src/googleError.ts @@ -32,6 +32,19 @@ const NUM_OF_PARTS_IN_PROTO_TYPE_NAME = 2; export class GoogleError extends Error { code?: Status; + /** + * The HTTP response status received by the REST fallback transport. + * + * `code` holds the gRPC status the response was mapped to, which is lossy: + * `rpcCodeFromHttpStatusCode` collapses whole ranges (every unmapped 5xx + * becomes INTERNAL), so the original status cannot be recovered from it. + * Telemetry reports the two separately, so the received status is kept here + * as well. + * + * Undefined for gRPC calls, and for fallback failures that never produced a + * response at all, such as an expired deadline or a connection error. + */ + httpStatusCode?: number; note?: string; metadata?: Metadata; statusDetails?: string | protobuf.Message<{}>[]; diff --git a/core/packages/gax/src/observability/TracerHelper.ts b/core/packages/gax/src/observability/TracerHelper.ts index c444b40360f..9c7487f48a4 100644 --- a/core/packages/gax/src/observability/TracerHelper.ts +++ b/core/packages/gax/src/observability/TracerHelper.ts @@ -15,7 +15,13 @@ */ import {EventEmitter} from 'events'; -import {Span, SpanStatusCode, trace, Tracer} from '@opentelemetry/api'; +import { + Attributes, + Span, + SpanStatusCode, + trace, + Tracer, +} from '@opentelemetry/api'; import {APICallback, GaxCallResult} from '../apitypes'; import {Status} from '../status'; @@ -105,6 +111,33 @@ function resolveErrorType(e: Error): string { return e.constructor?.name ?? e.name; } +/** + * Resolves the gRPC status reported for a failed call, as its name. + * + * Zero is treated as absent rather than as `OK`, for the same reason as in + * `resolveErrorType`: it is the proto3 default for an unset field, so a failed + * call must not be labelled `OK`. + */ +function resolveRpcStatusName(e: unknown): string { + const code = (e as {code?: unknown} | null)?.code; + if ( + typeof code === 'number' && + code !== Status.OK && + Status[code] !== undefined + ) { + return Status[code]; + } + return Status[Status.UNKNOWN]; +} + +/** + * Reads the HTTP response status recorded on a fallback error. + */ +function resolveHttpStatusCode(e: unknown): number | undefined { + const code = (e as {httpStatusCode?: unknown} | null)?.httpStatusCode; + return typeof code === 'number' ? code : undefined; +} + /** * Checks if a value behaves like a Promise or Thenable. * @@ -333,30 +366,76 @@ export function traceCall( let spanEnded = false; let errorRecorded = false; + // Resolved from the error when one is reported, and defaulted to success + // in endSpan otherwise. Held here rather than written immediately so that + // every completion path — promise, stream, callback, synchronous throw — + // emits them from the same place. + let rpcStatusName: string | undefined; + let httpStatusCode: number | undefined; + // Marks the span failed. Kept separate from recordError so paths that are // failures but not exceptions can set the status without emitting a // misleading exception event. + // + // The human-readable message is reported here and nowhere else. There is + // deliberately no `error.message` attribute: semconv deprecated it and + // calls it NOT RECOMMENDED on spans, because it has unbounded cardinality + // and restates the status description that already carries it. const setErrorStatus = (message: string) => { errorRecorded = true; span.setStatus({code: SpanStatusCode.ERROR, message}); }; - // Every path ends here, so the status is resolved in one place: ERROR if - // anything reported a failure, OK otherwise. + // The gRPC status is reported for both transports, because it is the one + // status gax resolves on every call and the only one a caller can compare + // across them. The transport-specific attribute is an alias of it on gRPC, + // and the received HTTP status on the fallback, which is a different value + // rather than a restatement of the same one. + // + // Written from one place so the two can never disagree. + const setStatusAttributes = () => { + const attributes: Attributes = { + 'rpc.response.status_code': rpcStatusName, + }; + if (dynamicArgs.rpcType === 'grpc') { + attributes['grpc.response.status_code'] = rpcStatusName; + } else if (httpStatusCode !== undefined) { + attributes['http.response.status_code'] = httpStatusCode; + } + span.setAttributes(attributes); + }; + + // Every path ends here, so the outcome is resolved in one place: ERROR if + // anything reported a failure, and left unset otherwise. + // + // A successful call deliberately does not set OK. Per OTel semconv the + // span status "MUST be left unset if the instrumented operation has ended + // without any errors"; `OK` is reserved for an application explicitly + // overriding the instrumentation's judgement, and a library must never + // claim it on the application's behalf. Unset already reads as success. const endSpan = () => { if (!spanEnded) { spanEnded = true; if (!errorRecorded) { - span.setStatus({code: SpanStatusCode.OK}); + rpcStatusName = Status[Status.OK]; + // Nothing carries the response status back on a successful fallback + // call, and success means a 2xx, so 200 is the only value available. + // A legacy Apiary 204 is therefore also reported as 200. + httpStatusCode = 200; } + setStatusAttributes(); span.end(); } }; const recordError = (e: unknown) => { + // Resolved for every failure, including non-Error throws: those carry no + // status, and resolveRpcStatusName reports UNKNOWN for them, which is + // the right answer for a call that failed for an unmapped reason. + rpcStatusName = resolveRpcStatusName(e); + httpStatusCode = resolveHttpStatusCode(e); if (e instanceof Error) { span.setAttributes({ - 'error.message': e.message, 'error.type': resolveErrorType(e), }); // recordException emits the `exception` event, which carries @@ -366,12 +445,18 @@ export function traceCall( span.recordException(e); setErrorStatus(e.message); } else { - const message = String(e); + // A non-Error throw carries no type, no message and no stack. `_OTHER` + // is the fallback semconv defines for exactly this, and reporting + // something matters: error.type is the dimension error-rate queries + // group on, so a failure missing it is invisible to them. + // + // No exception event is emitted here. recordException on a bare string + // yields an event with no exception.type and no stacktrace, which adds + // nothing the status description does not already carry. span.setAttributes({ - 'error.message': message, + 'error.type': '_OTHER', }); - span.recordException(message); - setErrorStatus(message); + setErrorStatus(String(e)); } }; diff --git a/core/packages/gax/test/unit/apiCallable.ts b/core/packages/gax/test/unit/apiCallable.ts index f7cfaed4148..7d85bdb2ea0 100644 --- a/core/packages/gax/test/unit/apiCallable.ts +++ b/core/packages/gax/test/unit/apiCallable.ts @@ -15,13 +15,21 @@ */ import assert from 'assert'; +import {PassThrough} from 'stream'; import {status} from '@grpc/grpc-js'; -import {afterEach, describe, it} from 'mocha'; +import {afterEach, beforeEach, describe, it} from 'mocha'; import * as sinon from 'sinon'; -import {RequestType} from '../../src/apitypes'; +import {CancellableStream, GRPCCall, RequestType} from '../../src/apitypes'; +import {createApiCall as gaxCreateApiCall} from '../../src/createApiCall'; +import {createApiCall as fallbackCreateApiCall} from '../../src/fallback'; +import {StreamDescriptor} from '../../src/descriptor'; +import {StreamType} from '../../src/streamingCalls/streaming'; import * as gax from '../../src/gax'; import {GoogleError} from '../../src/googleError'; +import {OtelHarness} from './otelHarness'; +import * as tracerHelper from '../../src/observability/TracerHelper'; +import {StaticTraceContext} from '../../src/observability/TracerHelper'; import * as utils from './utils'; import * as retries from '../../src/normalCalls/retries'; @@ -331,27 +339,832 @@ describe('createApiCall', () => { }); describe('in regards to OpenTelemetry Tracing', () => { + let harness: OtelHarness; + + const telemetryInfo: StaticTraceContext = { + gcpClientService: 'echo.googleapis.com', + gcpVersion: '1.2.3', + gcpRepo: 'googleapis/google-cloud-node', + gcpArtifact: '@google-cloud/echo', + }; + + beforeEach(() => { + harness = new OtelHarness(); + harness.setup(); + }); + afterEach(() => { + harness.teardown(); delete process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED; }); - it('creates an api call when GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED and CallSettings field is set', () => { + it('calls traceCall with dynamicArgs, staticArgs, and isStreamingCall when tracing is enabled', async () => { process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; - const mockCallOptions: gax.CallOptions = { + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', enableTelemetryTracing: true, otherArgs: { - internalTelemetryInfo: { - gcpClientService: 'test.googleapis.com', + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return {cancel: () => {}}; + } + + const apiCall = gaxCreateApiCall(func, settings); + await apiCall({param: 'test'}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [dynamicArgs, staticArgs, fn, isStreamingCall] = + traceCallSpy.firstCall.args; + + assert.deepStrictEqual(dynamicArgs, { + clientName: 'EchoClient', + methodName: 'Echo', + rpcType: 'grpc', + }); + assert.deepStrictEqual(staticArgs, telemetryInfo); + assert.strictEqual(typeof fn, 'function'); + assert.strictEqual(isStreamingCall, false); + }); + + it('passes isStreamingCall as true to traceCall for streaming calls when tracing is enabled', () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + const spy = sinon.spy(() => { + const s = new PassThrough({objectMode: true}); + s.push(null); + return Object.assign(s, {cancel: () => {}}); + }); + + const apiCall = gaxCreateApiCall( + spy as unknown as GRPCCall, + settings, + new StreamDescriptor(StreamType.SERVER_STREAMING, true), + ); + void apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [, , , isStreamingCall] = traceCallSpy.firstCall.args; + assert.strictEqual(isStreamingCall, true); + }); + + it('gracefully handles missing apiName and internalMethodName when tracing is enabled', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return {cancel: () => {}}; + } + + const apiCall = gaxCreateApiCall(func, settings); + await apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [dynamicArgs] = traceCallSpy.firstCall.args; + assert.deepStrictEqual(dynamicArgs, { + clientName: '', + methodName: '', + rpcType: 'grpc', + }); + }); + + it('returns invokeCall directly without calling traceCall when tracing is disabled', async () => { + delete process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: false, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return {cancel: () => {}}; + } + + const apiCall = gaxCreateApiCall(func, settings); + await apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.called, false); + assert.strictEqual(harness.getSpans('google-gax').length, 0); + }); + + it('correctly pipes telemetry information into the active span for gRPC calls', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings); + const [response] = (await apiCall({}, undefined)) as [ + {data: string}, + unknown, + unknown, + ]; + assert.deepStrictEqual(response, {data: 'hello'}); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.name, 'EchoClient.Echo'); + assert.strictEqual(span.ended, true); + assert.strictEqual( + span.attributes['gcp.client.service'], + 'echo.googleapis.com', + ); + assert.strictEqual(span.attributes['gcp.client.version'], '1.2.3'); + assert.strictEqual( + span.attributes['gcp.repo'], + 'googleapis/google-cloud-node', + ); + assert.strictEqual(span.attributes['gcp.artifact'], '@google-cloud/echo'); + assert.strictEqual(span.attributes['gcp.method.name'], 'Echo'); + assert.strictEqual(span.attributes['gcp.method.type'], 'grpc'); + }); + + it('correctly pipes telemetry information for HTTP fallback calls', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings, undefined, true); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.name, 'EchoClient.Echo'); + assert.strictEqual(span.ended, true); + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + }); + + it('passes fallback flag through when using fallback createApiCall with default options', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = fallbackCreateApiCall(func, settings); + await apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [dynamicArgs] = traceCallSpy.firstCall.args; + assert.strictEqual(dynamicArgs.rpcType, 'http'); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.name, 'EchoClient.Echo'); + assert.strictEqual(span.ended, true); + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + }); + + it('overrides an explicit _fallback argument, since the call is a fallback call by definition', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + // `_fallback` is documented as "unused; for compatibility only" and is + // never read. Reaching this function at all means the call is going over + // the fallback transport, so `false` must not be able to mislabel it. + const apiCall = fallbackCreateApiCall(func, settings, undefined, false); + await apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [dynamicArgs] = traceCallSpy.firstCall.args; + assert.strictEqual(dynamicArgs.rpcType, 'http'); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + }); + + it('ends the span and labels it DEADLINE_EXCEEDED when a fallback call times out', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + // The error the REST transport produces when a call exceeds its + // deadline: a GoogleError carrying the numeric gRPC status, so that + // retryCodes, caller `err.code` checks and telemetry all agree across + // transports. An unenforced deadline is what leaves a span open + // forever, which is the failure this tracing work has to surface + // rather than hide. + function timingOutFunc( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + const error = new GoogleError( + 'Deadline exceeded: Echo did not respond within 100 milliseconds.', + ); + error.code = status.DEADLINE_EXCEEDED; + setImmediate(() => callback(error)); + return { + cancel: () => {}, + }; + } + + const apiCall = fallbackCreateApiCall(timingOutFunc, settings); + const promise = apiCall({}, undefined); + + // The span must not be closed while the call is still outstanding; a + // fabricated end time would report a duration the RPC never took. + assert.strictEqual(harness.getSpans('google-gax').length, 0); + + await assert.rejects( + async () => { + await promise; + }, + (err: GoogleError) => { + assert.strictEqual(err.code, status.DEADLINE_EXCEEDED); + return true; + }, + ); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + // `resolveErrorType` maps the numeric code through the Status enum, so a + // deadline is reported by name rather than as the transport's own error + // class. Before the REST transport enforced the deadline this attribute + // would have read 'GaxiosError', and only if the call completed at all. + assert.strictEqual(span.attributes['error.type'], 'DEADLINE_EXCEEDED'); + assert.strictEqual(span.events.length, 1); + assert.strictEqual(span.events[0].name, 'exception'); + }); + + it('passes fallback flag and isStreamingCall as true for server-streaming fallback calls', () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const traceCallSpy = sinon.spy(tracerHelper, 'traceCall'); + + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + const spy = sinon.spy(() => { + const s = new PassThrough({objectMode: true}); + s.push(null); + return Object.assign(s, {cancel: () => {}}); + }); + + const apiCall = fallbackCreateApiCall( + spy as unknown as GRPCCall, + settings, + new StreamDescriptor(StreamType.SERVER_STREAMING, true), + ); + void apiCall({}, undefined); + + assert.strictEqual(traceCallSpy.calledOnce, true); + const [dynamicArgs, , , isStreamingCall] = traceCallSpy.firstCall.args; + assert.strictEqual(dynamicArgs.rpcType, 'http'); + assert.strictEqual(isStreamingCall, true); + }); + + it('sets rpcType to grpc when _fallback is boolean false', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings, undefined, false); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.attributes['gcp.method.type'], 'grpc'); + }); + + it('sets rpcType to http when _fallback is "rest"', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings, undefined, 'rest'); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + }); + + it('sets rpcType to http when _fallback is "proto"', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings, undefined, 'proto'); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.attributes['gcp.method.type'], 'http'); + }); + + it('pipes telemetry information configured via constructSettings', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const serviceName = 'google.example.v1.Echo'; + const defaults = gax.constructSettings( + serviceName, + { + interfaces: { + [serviceName]: { + methods: { + Echo: {}, + }, + }, }, }, - }; - const apiCall = createApiCall(() => {}, {settings: mockCallOptions}); - assert.strictEqual(typeof apiCall, 'function'); + {}, + {}, + undefined, + true, + telemetryInfo, + ); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, defaults.echo); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.name, 'EchoClient.Echo'); + assert.strictEqual(span.ended, true); + assert.strictEqual( + span.attributes['gcp.client.service'], + 'echo.googleapis.com', + ); + assert.strictEqual(span.attributes['gcp.client.version'], '1.2.3'); + assert.strictEqual( + span.attributes['gcp.repo'], + 'googleapis/google-cloud-node', + ); + assert.strictEqual(span.attributes['gcp.artifact'], '@google-cloud/echo'); + assert.strictEqual(span.attributes['gcp.method.name'], 'Echo'); + assert.strictEqual(span.attributes['gcp.method.type'], 'grpc'); + }); + + it('records error details on the span when the API call fails', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function failingFunc( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + const error = new GoogleError('RPC test failure'); + setImmediate(() => { + callback(error); + }); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(failingFunc, settings); + const promise = apiCall({}, undefined); + assert.strictEqual(harness.getSpans('google-gax').length, 0); + + await assert.rejects( + async () => { + await promise; + }, + (err: GoogleError) => { + assert.strictEqual(err.message, 'RPC test failure'); + return true; + }, + ); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + assert.strictEqual(span.status.message, 'RPC test failure'); + assert.strictEqual(span.events.length, 1); + assert.strictEqual(span.events[0].name, 'exception'); + }); + + it('does not end span prematurely for successful asynchronous API calls', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function asyncFunc( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + setImmediate(() => { + callback(null, {data: 'hello'}); + }); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(asyncFunc, settings); + const promise = apiCall({}, undefined); + + // Verify the span is not ended prematurely while the call is in flight + assert.strictEqual(harness.getSpans('google-gax').length, 0); + + const [response] = (await promise) as [{data: string}, unknown, unknown]; + assert.deepStrictEqual(response, {data: 'hello'}); + + // Span must only be ended after completion + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + assert.strictEqual(span.name, 'EchoClient.Echo'); + }); + + it('cancels the call and ends the span', async () => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function cancellableFunc( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + const timeoutId = setTimeout(() => { + callback(null, {data: 'done'}); + }, 5000); + return { + cancel: () => { + clearTimeout(timeoutId); + const err = new GoogleError('cancelled'); + err.code = status.CANCELLED; + callback(err); + }, + }; + } + + const apiCall = gaxCreateApiCall(cancellableFunc, settings); + const promise = apiCall({}, undefined); + assert.strictEqual(typeof promise.cancel, 'function'); + promise.cancel(); + + await assert.rejects(async () => { + await promise; + }); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + }); + + it('does not create any spans when tracing is disabled', async () => { + delete process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: false, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + function func( + argument: {}, + metadata: {}, + options: {}, + callback: (err: GoogleError | null, resp?: unknown) => void, + ) { + callback(null, {data: 'hello'}); + return { + cancel: () => {}, + }; + } + + const apiCall = gaxCreateApiCall(func, settings); + await apiCall({}, undefined); + + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 0); + }); + + it('manages span lifetime for streaming API calls until stream ends', done => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + const spy = sinon.spy(() => { + const s = new PassThrough({ + objectMode: true, + }); + s.push({data: 'chunk1'}); + s.push({data: 'chunk2'}); + s.push(null); + return Object.assign(s, {cancel: () => {}}); + }); + + const apiCall = gaxCreateApiCall( + spy as unknown as GRPCCall, + settings, + new StreamDescriptor(StreamType.SERVER_STREAMING, true), + ); + const stream = apiCall({}, undefined) as CancellableStream; + assert.strictEqual(harness.getSpans('google-gax').length, 0); + + const received: unknown[] = []; + stream.on('data', chunk => { + received.push(chunk); + // Span must remain active while streaming chunks + assert.strictEqual(harness.getSpans('google-gax').length, 0); + }); + stream.on('end', () => { + try { + assert.strictEqual(received.length, 2); + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + assert.strictEqual(span.name, 'EchoClient.Echo'); + assert.strictEqual(span.attributes['gcp.method.type'], 'grpc'); + done(); + } catch (e) { + done(e); + } + }); }); - it('creates an api call when GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED is not set', () => { - const apiCall = createApiCall(() => {}); - assert.strictEqual(typeof apiCall, 'function'); + it('records error details on the span when a streaming API call errors', done => { + process.env.GOOGLE_SDK_NODE_EXPERIMENTAL_O11Y_ENABLED = 'true'; + const settings = new gax.CallSettings({ + apiName: 'google.example.v1.Echo', + enableTelemetryTracing: true, + otherArgs: { + internalTelemetryInfo: telemetryInfo, + internalMethodName: 'Echo', + }, + }); + + const spy = sinon.spy(() => { + const s = new PassThrough({ + objectMode: true, + }); + setImmediate(() => { + s.emit('error', new GoogleError('streaming test failure')); + }); + return Object.assign(s, {cancel: () => {}}); + }); + + const apiCall = gaxCreateApiCall( + spy as unknown as GRPCCall, + settings, + new StreamDescriptor(StreamType.SERVER_STREAMING, true), + ); + const stream = apiCall({}, undefined) as CancellableStream; + assert.strictEqual(harness.getSpans('google-gax').length, 0); + + stream.on('error', (err: GoogleError) => { + try { + assert.strictEqual(err.message, 'streaming test failure'); + const spans = harness.getSpans('google-gax'); + assert.strictEqual(spans.length, 1); + const span = spans[0]; + assert.strictEqual(span.ended, true); + assert.strictEqual(span.status.message, 'streaming test failure'); + assert.strictEqual(span.events.length, 1); + assert.strictEqual(span.events[0].name, 'exception'); + done(); + } catch (e) { + done(e); + } + }); }); }); }); diff --git a/core/packages/gax/test/unit/grpc-fallback.ts b/core/packages/gax/test/unit/grpc-fallback.ts index 732c52e298d..1bbb628e317 100644 --- a/core/packages/gax/test/unit/grpc-fallback.ts +++ b/core/packages/gax/test/unit/grpc-fallback.ts @@ -21,10 +21,20 @@ import assert from 'assert'; import {describe, it, beforeEach, afterEach, after} from 'mocha'; import * as protobuf from 'protobufjs'; import * as sinon from 'sinon'; +import * as stream from 'stream'; import echoProtoJson = require('../fixtures/echo.json'); import {GrpcClient} from '../../src/fallback'; -import {ClientStubOptions, GoogleAuth, GoogleError} from '../../src'; -import {PassThroughClient} from 'google-auth-library'; +import { + CallSettings, + ClientStubOptions, + GoogleAuth, + GoogleError, + Status, + createApiCall, +} from '../../src'; +import {GRPCCall} from '../../src/apitypes'; +import {StreamArrayParser} from '../../src/streamArrayParser'; +import {gaxios, PassThroughClient} from 'google-auth-library'; import {setMockFallbackResponse} from './utils'; let authClient = new PassThroughClient(); @@ -444,6 +454,35 @@ describe('grpc-fallback', () => { }); }); + it('should record the received http status on the error', async () => { + const requestObject = {content: 'test-content'}; + + // The body reports 400 while the response itself is a 503. `code` is + // derived from the body, so a status read back off the error can only be + // the received one if the two differ. + setMockFallbackResponse( + gaxGrpc, + new Response( + JSON.stringify({error: {code: 400, message: 'mismatched status'}}), + {status: 503}, + ), + ); + + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + await new Promise((resolve, reject) => { + echoStub.echo(requestObject, {}, {}, (err?: Error) => { + try { + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.INVALID_ARGUMENT); + assert.strictEqual(err.httpStatusCode, 503); + resolve(); + } catch (e) { + reject(e); + } + }); + }); + }); + it('should promote ErrorInfo if exist in fallback-rest error', async () => { const requestObject = {content: 'test-content'}; // example of an actual google.rpc.Status error message returned by Translate API @@ -539,4 +578,407 @@ describe('grpc-fallback', () => { const stub = await gaxGrpc.createStub(echoService, stubOptions); stub.close({}, {}, {}, () => {}); }); + + // `setMockFallbackResponse` discards the options it is handed, but the + // deadline and metadata handling under test are only observable there, so + // record them. + function recordRequests( + client: GrpcClient, + response: Response, + ): gaxios.GaxiosOptions[] { + const requests: gaxios.GaxiosOptions[] = []; + class RecordingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + requests.push(opts); + return Object.assign(response, { + config: { + headers: response.headers, + url: new URL(opts.url || 'https://example.com'), + }, + data: response.body as T, + }); + } + } + client.auth = new GoogleAuth({authClient: new RecordingAuthClient()}); + return requests; + } + + describe('call metadata', () => { + async function headersSentFor(metadata: { + [name: string]: string | string[]; + }): Promise { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo({content: 'test'}, metadata, {}, () => resolve()); + }); + + return requests[0].headers as Headers; + } + + it('should send every value of a multi-valued header', async () => { + const headers = await headersSentFor({'x-multi': ['a', 'b', 'c']}); + + // gRPC metadata is multi-valued and `buildMetadata` normalizes every + // value to an array precisely because of that. Reading index 0 silently + // dropped the rest. The Headers API joins repeated values with ', '. + assert.strictEqual(headers.get('x-multi'), 'a, b, c'); + }); + + it('should send a single-valued header as its only value', async () => { + const headers = await headersSentFor({'x-single': ['one']}); + + assert.strictEqual(headers.get('x-single'), 'one'); + }); + + it('should send a plain string value whole', async () => { + const headers = await headersSentFor({'x-plain': 'hello'}); + + // Indexing a string yields its first character, so this used to arrive + // as 'h'. The declared parameter type said the values were strings while + // the code indexed them as arrays; both could not be right. + assert.strictEqual(headers.get('x-plain'), 'hello'); + }); + + it('should let metadata replace a header set by the request encoder', async () => { + const headers = await headersSentFor({ + 'content-type': ['application/x-custom'], + }); + + // The previous `headers.set` replaced rather than appended, so keeping + // every value must not turn an override into an accumulation. + assert.strictEqual(headers.get('content-type'), 'application/x-custom'); + }); + }); + + describe('call deadline', () => { + function signalOf(request: gaxios.GaxiosOptions): AbortSignal | undefined { + return request.signal as AbortSignal | undefined; + } + + // Resolves true if the signal aborts within the budget, false if it does + // not. A budget rather than a bare `aborted` read, because the abort is + // asynchronous and a test that only sampled it would pass for the wrong + // reason. + function abortedWithin( + signal: AbortSignal | undefined, + ms: number, + ): Promise { + if (!signal) { + return Promise.resolve(false); + } + if (signal.aborted) { + return Promise.resolve(true); + } + return new Promise(resolve => { + const timer = setTimeout(() => resolve(false), ms); + signal.addEventListener( + 'abort', + () => { + clearTimeout(timer); + resolve(true); + }, + {once: true}, + ); + }); + } + + // The error an aborted request actually produces, measured end to end + // against a server that accepts the connection and never replies: + // node-fetch discards `signal.reason` and throws its own AbortError, + // gaxios wraps that without setting `name` (so it stays the inherited + // 'Error') and only copies `code` from a DOMException cause, which this is + // not. A `cancel()` produces a byte-identical error. Earlier versions of + // these tests fabricated a `TimeoutError` that never occurs in production + // and so passed against a translation that was dead code. + function abortError(): Error { + const cause = new Error('The operation was aborted.'); + cause.name = 'AbortError'; + return new Error('The operation was aborted.', {cause}); + } + + // Rejects as soon as the request is aborted. `cancel()` can run before the + // asynchronous auth chain ever reaches the transport, and a listener added + // to an already-aborted signal never fires, so check the state first. + function rejectWhenAborted( + signal: AbortSignal | undefined, + ): Promise { + return new Promise((_resolve, reject) => { + if (!signal) { + return; + } + if (signal.aborted) { + reject(abortError()); + } else { + signal.addEventListener('abort', () => reject(abortError()), { + once: true, + }); + } + }); + } + + // A transport that is actually bound by the signal: it stays pending until + // the request is aborted, then rejects the way the real one does. + function rejectOnAbort(client: GrpcClient): gaxios.GaxiosOptions[] { + const requests: gaxios.GaxiosOptions[] = []; + class AbortingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + requests.push(opts); + return rejectWhenAborted(signalOf(opts)); + } + } + client.auth = new GoogleAuth({authClient: new AbortingAuthClient()}); + return requests; + } + + // Aborts after the response headers arrive but before the body is read, + // which rejects in the stub's inner handler rather than the outer one. + function rejectDuringBodyRead(client: GrpcClient) { + class BodyAbortingAuthClient extends PassThroughClient { + async request( + opts: gaxios.GaxiosOptions, + ): Promise> { + const signal = signalOf(opts); + return { + ok: true, + status: 200, + headers: new Headers(), + arrayBuffer: () => rejectWhenAborted(signal), + } as unknown as gaxios.GaxiosResponse; + } + } + client.auth = new GoogleAuth({authClient: new BodyAbortingAuthClient()}); + } + + it('should abort the in-flight request when the deadline expires', async () => { + const requests = rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + () => resolve(), + ); + }); + + // The transport never answered. Before this, nothing read the deadline, + // so the request and the callback waiting on it stayed outstanding. + assert.strictEqual(signalOf(requests[0])?.aborted, true); + }); + + it('should carry CallSettings.timeout through to the transport', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + // Every other test here hands the stub a deadline directly, which only + // exercises the stub itself. This one goes through `createApiCall`, the + // path a generated client takes, so `addTimeoutArg` is what produces the + // deadline. That hand-off is the seam where the deadline used to be + // dropped, and no direct call to the stub can see it. + const apiCall = createApiCall( + Promise.resolve(echoStub.echo as unknown as GRPCCall), + new CallSettings({timeout: 50}), + ); + + await assert.rejects( + apiCall({content: 'test'}, {}) as unknown as Promise, + (err: unknown) => { + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + return true; + }, + ); + }); + + it('should not abort a call that has no deadline', async () => { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + await new Promise(resolve => { + echoStub.echo({content: 'test'}, {}, {}, () => resolve()); + }); + + assert.strictEqual( + await abortedWithin(signalOf(requests[0]), 100), + false, + ); + }); + + it('should abort promptly, and not throw, for an already-expired deadline', async () => { + const requests = recordRequests( + gaxGrpc, + new Response(Buffer.from(JSON.stringify({content: 'test'}))), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + // `AbortSignal.timeout` rejects a negative delay with a RangeError, so + // an expired deadline that was not clamped would throw out of the stub + // before the request was ever made. Zero is the right clamp: the call + // has no time left, so it should abort on the next tick. + await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() - 60000)}, + () => resolve(), + ); + }); + + assert.strictEqual(await abortedWithin(signalOf(requests[0]), 100), true); + }); + + it('should not bound server-streaming calls by the deadline', async () => { + const responseStream = new stream.Readable(); + responseStream.push(JSON.stringify([{content: 'test'}])); + responseStream.push(null); + const requests = recordRequests( + gaxGrpc, + new Response(responseStream as unknown as BodyInit), + ); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const responses = echoStub.expand( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + () => {}, + ) as StreamArrayParser; + await new Promise((resolve, reject) => { + responses.on('data', () => {}); + responses.on('error', reject); + responses.on('end', resolve); + }); + + // A server stream is long-lived by design; the signal would stay armed + // once the body starts flowing and abort a healthy read. + assert.strictEqual( + await abortedWithin(signalOf(requests[0]), 100), + false, + ); + }); + + it('should report an expired deadline as DEADLINE_EXCEEDED', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + (err?: Error) => resolve(err), + ); + }); + + // gRPC reports this condition with a numeric status, and retryCodes, + // caller `err.code` checks and telemetry all key off that, so the REST + // path must not leak the transport's own error shape. + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + assert.match(err.message, /Deadline exceeded/); + }); + + it('should report a deadline that expires while the body is being read', async () => { + rejectDuringBodyRead(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 50)}, + (err?: Error) => resolve(err), + ); + }); + + // Headers arriving in time does not mean the call met its deadline. + assert(err instanceof GoogleError); + assert.strictEqual(err.code, Status.DEADLINE_EXCEEDED); + }); + + it('should not report a cancelled call as DEADLINE_EXCEEDED', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + const call = echoStub.echo( + {content: 'test'}, + {}, + {deadline: new Date(Date.now() + 5000)}, + (err?: Error) => resolve(err), + ); + (call as {cancel: () => void}).cancel(); + }); + + // A deadline was armed here but never expired; the caller gave up first. + // This is the case no amount of error inspection can get right, because + // the abort a cancel produces is byte-identical to the one a timeout + // produces. Only the stub, which armed the timer, knows which fired. + assert(!(err instanceof GoogleError)); + }); + + it('should leave an abort error alone when no deadline was forwarded', async () => { + rejectOnAbort(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + const err = await new Promise(resolve => { + const call = echoStub.echo({content: 'test'}, {}, {}, (err?: Error) => + resolve(err), + ); + (call as {cancel: () => void}).cancel(); + }); + + // Nothing armed a deadline, so this abort came from the caller, and + // reporting a deadline that was never set would be a fabrication. The + // error itself is byte-identical to a timeout's, which is why the + // translation is gated on the flag the stub records rather than on + // anything read back off the error. + assert(!(err instanceof GoogleError)); + assert.strictEqual((err?.cause as Error | undefined)?.name, 'AbortError'); + }); + + it('should not report an error when the caller cancelled', async () => { + rejectDuringBodyRead(gaxGrpc); + const echoStub = await gaxGrpc.createStub(echoService, stubOptions); + + let callbackErr: unknown; + let callbackCalled = false; + const call = echoStub.echo( + {content: 'test'}, + {}, + {}, + (err?: Error, resp?: {}) => { + callbackCalled = true; + callbackErr = err ?? resp; + }, + ); + (call as {cancel: () => void}).cancel(); + + await new Promise(resolve => setTimeout(resolve, 100)); + + // A caller that cancelled does not need the resulting abort reported + // back to it. The guard here used to test `err.name !== 'AbortError'`, + // but gaxios wraps node-fetch's AbortError and never sets its own + // `name`, leaving the inherited 'Error', so the check never matched and + // cancelled calls reported an error anyway. + assert.strictEqual( + callbackCalled, + false, + `callback was invoked with ${callbackErr}`, + ); + }); + }); }); diff --git a/core/packages/gax/test/unit/otelHarness.ts b/core/packages/gax/test/unit/otelHarness.ts index 77c9e98e0c1..0686de71804 100644 --- a/core/packages/gax/test/unit/otelHarness.ts +++ b/core/packages/gax/test/unit/otelHarness.ts @@ -199,6 +199,123 @@ export class OtelHarness { ); } } + + /** + * The three response status attributes carried by a traced call. + * + * @param {ReadableSpan} span - The span to read. + * @returns {ResponseStatusAttributes} The attributes, each undefined if absent. + */ + responseStatus(span: ReadableSpan): ResponseStatusAttributes { + return { + rpc: span.attributes['rpc.response.status_code'] as string | undefined, + grpc: span.attributes['grpc.response.status_code'] as string | undefined, + http: span.attributes['http.response.status_code'] as number | undefined, + }; + } + + /** + * Asserts the response status attributes of a traced call. + * + * The transport-specific attribute is not named by the caller. It is derived + * from the span's own `gcp.method.type`, so a test cannot assert a + * combination the tracer is not supposed to produce — such as an HTTP status + * on a gRPC span. Both the presence of the attribute that applies and the + * absence of the one that does not are checked, because the second half is + * what catches an attribute leaking onto the wrong transport. + * + * `httpStatus` is only meaningful on a fallback span. Omitting it there + * asserts that no HTTP status was reported, which is the expected result for + * a failure that never received a response, such as an expired deadline. + * + * @param {object} expected - Expected status values. + * @param {string} expected.rpcStatus - gRPC status name, e.g. 'OK' or 'NOT_FOUND'. + * @param {number} [expected.httpStatus] - HTTP status expected on a fallback span. + * @param {object} [options] - Span selection. + * @param {string} [options.tracerName] - Restrict the lookup to one instrumentation scope. + * @param {ReadableSpan} [options.span] - Span to check; defaults to the only exported span. + */ + assertResponseStatus( + expected: {rpcStatus: string; httpStatus?: number}, + options: {tracerName?: string; span?: ReadableSpan} = {}, + ): void { + const target = options.span ?? this.requireSingleSpan(options.tracerName); + const actual = this.responseStatus(target); + const transport = target.attributes['gcp.method.type']; + const where = `span '${target.name}'`; + + assert.ok( + transport === 'grpc' || transport === 'http', + `${where} has gcp.method.type ${JSON.stringify(transport)}; the ` + + 'transport-specific status attribute cannot be checked without it. ' + + 'Was this span produced by traceCall?', + ); + + assert.strictEqual( + actual.rpc, + expected.rpcStatus, + `expected ${where} to report rpc.response.status_code ` + + `${JSON.stringify(expected.rpcStatus)}, got ${JSON.stringify(actual.rpc)}. ` + + 'This attribute is reported on every call, on both transports.', + ); + + if (transport === 'grpc') { + assert.strictEqual( + actual.grpc, + expected.rpcStatus, + `expected ${where} to report grpc.response.status_code ` + + `${JSON.stringify(expected.rpcStatus)}, got ${JSON.stringify(actual.grpc)}. ` + + 'On a gRPC span it mirrors rpc.response.status_code.', + ); + assert.strictEqual( + actual.http, + undefined, + `${where} is a gRPC span but reported http.response.status_code ` + + `${JSON.stringify(actual.http)}. A gRPC call has no HTTP status, ` + + 'not even a synthesized one.', + ); + assert.strictEqual( + expected.httpStatus, + undefined, + 'assertResponseStatus was given an expected httpStatus for a gRPC ' + + 'span, which can never hold one. Drop it, or assert against a ' + + 'fallback span.', + ); + return; + } + + assert.strictEqual( + actual.grpc, + undefined, + `${where} is a fallback span but reported grpc.response.status_code ` + + `${JSON.stringify(actual.grpc)}. The gRPC status is reported as ` + + 'rpc.response.status_code there, not under the grpc.* name.', + ); + assert.strictEqual( + actual.http, + expected.httpStatus, + expected.httpStatus === undefined + ? `expected ${where} to report no http.response.status_code, got ` + + `${JSON.stringify(actual.http)}. It is only reported when a ` + + 'response was actually received.' + : `expected ${where} to report http.response.status_code ` + + `${expected.httpStatus}, got ${JSON.stringify(actual.http)}. ` + + 'This is the status the transport received, which is not ' + + 'recoverable from the gRPC status it was mapped to.', + ); + } +} + +/** + * The response status attributes read off a traced span. + */ +export interface ResponseStatusAttributes { + /** `rpc.response.status_code`: gRPC status name, reported on both transports. */ + rpc: string | undefined; + /** `grpc.response.status_code`: gRPC spans only. */ + grpc: string | undefined; + /** `http.response.status_code`: fallback spans that received a response. */ + http: number | undefined; } /** diff --git a/core/packages/gax/test/unit/tracerHelper.ts b/core/packages/gax/test/unit/tracerHelper.ts index 7e87577f9bf..9ca9056f4bb 100644 --- a/core/packages/gax/test/unit/tracerHelper.ts +++ b/core/packages/gax/test/unit/tracerHelper.ts @@ -107,6 +107,10 @@ describe('TracerHelper', () => { ); assert.strictEqual(span.attributes['gcp.method.name'], 'GetObject'); assert.strictEqual(span.attributes['gcp.method.type'], 'grpc'); + // A successful call reports no error.type, and leaves the status unset + // rather than claiming OK on the application's behalf. + assert.strictEqual(span.attributes['error.type'], undefined); + assert.strictEqual(span.status.code, SpanStatusCode.UNSET); assert.strictEqual(span.events.length, 0); }); @@ -132,7 +136,10 @@ describe('TracerHelper', () => { const span = spans[0]; assert.strictEqual(span.name, 'StorageClient.GetObject'); assert.strictEqual(span.ended, true); - assert.strictEqual(span.attributes['error.message'], 'RPC Failed'); + // The message is carried by the status description. `error.message` is + // deprecated and NOT RECOMMENDED on spans, so it must not appear. + assert.strictEqual(span.status.message, 'RPC Failed'); + assert.strictEqual(span.attributes['error.message'], undefined); // No status code on this error, so error.type falls back to the class. assert.strictEqual(span.attributes['error.type'], 'Error'); // exception.* belongs on the exception event, not on the span. @@ -260,7 +267,7 @@ describe('TracerHelper', () => { assert.strictEqual(span.attributes['error.type'], 'GoogleError'); }); - it('omits error.type entirely when a non-Error is thrown', async () => { + it('reports the _OTHER error.type when a non-Error is thrown', async () => { await assert.rejects(async () => { await traceCall(dynamicArgs, staticArgs, async () => { throw 'plain string failure'; @@ -268,13 +275,163 @@ describe('TracerHelper', () => { }); const span = harness.requireSingleSpan('google-gax'); - assert.strictEqual( - span.attributes['error.message'], - 'plain string failure', - ); - assert.strictEqual(span.attributes['error.type'], undefined); + // Something must be reported, or the failure is invisible to any + // error-rate query that groups on error.type. + assert.strictEqual(span.attributes['error.type'], '_OTHER'); assert.strictEqual(span.attributes['exception.type'], undefined); assert.strictEqual(span.status.code, SpanStatusCode.ERROR); + // The thrown value survives as the status description. + assert.strictEqual(span.status.message, 'plain string failure'); + // No exception event: recordException on a bare string yields one with + // no type and no stacktrace, which adds nothing to the status above. + assert.strictEqual(span.events.length, 0); + }); + + // The two signals answer different questions, and the split between them + // is the part most easily broken by a well-meaning edit. Error information + // (span status + error.type) says how the operation ended and is what + // error-rate queries group on, so it must stay low-cardinality and must + // exist for every failure. Exception information (the `exception` event) + // says what was thrown, carries the unbounded detail, and only exists when + // something actually was thrown. + describe('error and exception reporting', () => { + const failWith = async (thrown: unknown) => { + await assert.rejects(async () => { + await traceCall(dynamicArgs, staticArgs, async () => { + throw thrown; + }); + }); + return harness.requireSingleSpan('google-gax'); + }; + + it('keeps error information on the span and exception detail on the event', async () => { + const error = new GoogleError('object does not exist'); + error.code = Status.NOT_FOUND; + + const span = await failWith(error); + + // Error information: the outcome, on the span itself. + assert.strictEqual(span.status.code, SpanStatusCode.ERROR); + assert.strictEqual(span.status.message, 'object does not exist'); + assert.strictEqual(span.attributes['error.type'], 'NOT_FOUND'); + + // Exception information: the detail, on the event. + assert.strictEqual(span.events.length, 1); + const event = span.events[0]; + assert.strictEqual(event.name, 'exception'); + // OTel derives exception.type from `code` when the error carries one, + // falling back to `name` otherwise, so a coded gax error reports the + // bare number '5' here. That is precisely why error.type is resolved + // separately: 'NOT_FOUND' above is the value worth querying on. + assert.strictEqual(event.attributes?.['exception.type'], '5'); + assert.strictEqual( + event.attributes?.['exception.message'], + 'object does not exist', + ); + + // Neither may leak into the other. exception.* on the span would + // duplicate the event at span cardinality, and error.* on the event + // would split the dimension that error-rate queries group on. + assert.strictEqual(span.attributes['exception.type'], undefined); + assert.strictEqual(span.attributes['exception.message'], undefined); + assert.strictEqual(span.attributes['exception.stacktrace'], undefined); + assert.strictEqual(event.attributes?.['error.type'], undefined); + }); + + it('never sets the deprecated error.message attribute', async () => { + // semconv deprecated it and calls it NOT RECOMMENDED on spans: it has + // unbounded cardinality and restates the status description. The + // message must be reachable, just not from here. + const span = await failWith(new Error('quota exceeded')); + + assert.strictEqual(span.attributes['error.message'], undefined); + assert.strictEqual(span.status.message, 'quota exceeded'); + assert.strictEqual( + span.events[0].attributes?.['exception.message'], + 'quota exceeded', + ); + }); + + it('carries the stacktrace on the exception event', async () => { + // The stacktrace is the reason the event exists at all: it is the one + // piece of detail no span attribute is allowed to hold. + const span = await failWith(new Error('boom')); + + const stacktrace = span.events[0].attributes?.['exception.stacktrace']; + assert.strictEqual(typeof stacktrace, 'string'); + assert.ok( + (stacktrace as string).includes('boom'), + `expected a stacktrace mentioning the failure, got ${JSON.stringify( + stacktrace, + )}`, + ); + }); + + it('reports error.type consistently with the RPC status', async () => { + // semconv asks that error.type be applied consistently across the + // signals a single operation reports. The two are resolved by separate + // helpers, so nothing but a test keeps them from drifting apart. + const error = Object.assign(new Error('5 NOT_FOUND: gone'), {code: 5}); + + const span = await failWith(error); + + assert.strictEqual(span.attributes['error.type'], 'NOT_FOUND'); + harness.assertResponseStatus({rpcStatus: 'NOT_FOUND'}, {span}); + }); + + it('records one exception event however many completion signals arrive', async () => { + // A stream can report 'error' and then still emit 'end' and 'close'. + // Only the first may be recorded: a second event would double-count + // the failure, and a later success signal must not overwrite it. + const emitter = new EventEmitter(); + traceCall(dynamicArgs, staticArgs, () => emitter, true); + + emitter.emit('error', new Error('stream broke')); + emitter.emit('end'); + emitter.emit('close'); + + const span = harness.requireSingleSpan('google-gax'); + assert.strictEqual(span.events.length, 1); + assert.strictEqual(span.status.code, SpanStatusCode.ERROR); + assert.strictEqual(span.status.message, 'stream broke'); + assert.strictEqual(span.attributes['error.type'], 'Error'); + }); + + it('still resolves the RPC status for a non-Error carrying a code', async () => { + // error.type falls back to _OTHER because a non-Error has no class + // worth reporting, but the domain status is resolved independently and + // is still recoverable. The two do not have to agree here. + const span = await failWith({code: Status.NOT_FOUND}); + + assert.strictEqual(span.attributes['error.type'], '_OTHER'); + harness.assertResponseStatus({rpcStatus: 'NOT_FOUND'}, {span}); + assert.strictEqual(span.events.length, 0); + }); + + it('survives a thrown null', async () => { + // resolveRpcStatusName and String() both have to tolerate it; a throw + // inside recordError would lose the span entirely. + const span = await failWith(null); + + assert.strictEqual(span.status.code, SpanStatusCode.ERROR); + assert.strictEqual(span.status.message, 'null'); + assert.strictEqual(span.attributes['error.type'], '_OTHER'); + harness.assertResponseStatus({rpcStatus: 'UNKNOWN'}, {span}); + }); + + it('reports no error information at all when the call succeeds', async () => { + await traceCall(dynamicArgs, staticArgs, async () => ({ok: true})); + + const span = harness.requireSingleSpan('google-gax'); + // semconv: instrumentation SHOULD NOT set error.type on success, and + // the status MUST be left unset. An UNSET status with no error.type is + // what lets a consumer filter failures out cleanly. + assert.strictEqual(span.attributes['error.type'], undefined); + assert.strictEqual(span.attributes['error.message'], undefined); + assert.strictEqual(span.status.code, SpanStatusCode.UNSET); + assert.strictEqual(span.status.message, undefined); + assert.strictEqual(span.events.length, 0); + }); }); it('handles missing optional static arguments gracefully', async () => { @@ -315,6 +472,169 @@ describe('TracerHelper', () => { assert.strictEqual(spans[0].attributes['gcp.method.type'], 'http'); }); + describe('response status attributes', () => { + const httpDynamicArgs: DynamicTraceContext = { + clientName: 'ComputeClient', + methodName: 'InsertInstance', + rpcType: 'http', + }; + + it('reports OK on a successful grpc call', async () => { + await traceCall(dynamicArgs, staticArgs, async () => 'ok'); + + harness.assertResponseStatus({rpcStatus: 'OK'}); + }); + + it('reports OK and 200 on a successful http call', async () => { + await traceCall(httpDynamicArgs, staticArgs, async () => 'ok'); + + harness.assertResponseStatus({rpcStatus: 'OK', httpStatus: 200}); + }); + + it('reports the gRPC status name on a failed grpc call', async () => { + // Shape produced by grpc-js callErrorFromStatus. + const error = Object.assign(new Error('5 NOT_FOUND'), {code: 5}); + + await assert.rejects(async () => { + await traceCall(dynamicArgs, staticArgs, async () => { + throw error; + }); + }); + + harness.assertResponseStatus({rpcStatus: 'NOT_FOUND'}); + }); + + it('reports the received http status alongside the mapped gRPC status', async () => { + // 418 is unmapped, so rpcCodeFromHttpStatusCode collapses it to + // FAILED_PRECONDITION. The received status is therefore not + // recoverable from `code`, which is why it is carried separately. + const error = new GoogleError('teapot'); + error.code = Status.FAILED_PRECONDITION; + error.httpStatusCode = 418; + + await assert.rejects(async () => { + await traceCall(httpDynamicArgs, staticArgs, async () => { + throw error; + }); + }); + + harness.assertResponseStatus({ + rpcStatus: 'FAILED_PRECONDITION', + httpStatus: 418, + }); + }); + + it('omits the http status when no response was received', async () => { + // What toDeadlineExceeded produces: a real gRPC status, but no + // response and so no HTTP status to report. Omitting httpStatus + // asserts the attribute is absent. + const error = new GoogleError('Deadline exceeded'); + error.code = Status.DEADLINE_EXCEEDED; + + await assert.rejects(async () => { + await traceCall(httpDynamicArgs, staticArgs, async () => { + throw error; + }); + }); + + harness.assertResponseStatus({rpcStatus: 'DEADLINE_EXCEEDED'}); + }); + + it('reports UNKNOWN for a failure carrying no gRPC status', async () => { + const error = Object.assign(new Error('connect ECONNREFUSED'), { + code: 'ECONNREFUSED', + }); + + await assert.rejects(async () => { + await traceCall(dynamicArgs, staticArgs, async () => { + throw error; + }); + }); + + harness.assertResponseStatus({rpcStatus: 'UNKNOWN'}); + }); + + it('reports UNKNOWN rather than OK for a zero status code', async () => { + // Zero is the proto3 default for an unset code, so a failed call must + // not be labelled OK. + const error = Object.assign(new Error('unset code'), {code: 0}); + + await assert.rejects(async () => { + await traceCall(dynamicArgs, staticArgs, async () => { + throw error; + }); + }); + + harness.assertResponseStatus({rpcStatus: 'UNKNOWN'}); + }); + + it('reports UNKNOWN when a non-Error is thrown', async () => { + await assert.rejects(async () => { + await traceCall(dynamicArgs, staticArgs, async () => { + throw 'plain string failure'; + }); + }); + + harness.assertResponseStatus({rpcStatus: 'UNKNOWN'}); + }); + + it('reports the status of a stream failure', () => { + const emitter = new EventEmitter(); + const error = new GoogleError('stream failed'); + error.code = Status.UNAVAILABLE; + + traceCall(dynamicArgs, staticArgs, () => emitter, true); + emitter.emit('error', error); + + harness.assertResponseStatus({rpcStatus: 'UNAVAILABLE'}); + }); + + it('reports the status of a failure delivered to a callback', done => { + const error = new GoogleError('permission denied'); + error.code = Status.PERMISSION_DENIED; + + // Mimics an API caller that returns OngoingCall (no `.promise`), so + // the span is bound to the callback rather than to a promise. + const returned = traceCall( + dynamicArgs, + staticArgs, + tracedCallback => { + tracedCallback!(error); + return undefined as unknown as ResultTuple; + }, + false, + () => { + harness.assertResponseStatus({rpcStatus: 'PERMISSION_DENIED'}); + done(); + }, + ); + + assert.strictEqual(returned, undefined); + }); + + // The harness derives the transport from the span, so these guard the + // guard: a regression that moved an attribute onto the wrong transport + // has to be caught rather than quietly accepted. + it('rejects an expected http status on a grpc span', async () => { + await traceCall(dynamicArgs, staticArgs, async () => 'ok'); + + assert.throws( + () => + harness.assertResponseStatus({rpcStatus: 'OK', httpStatus: 200}), + /can never hold one/, + ); + }); + + it('rejects a grpc status name that does not match', async () => { + await traceCall(dynamicArgs, staticArgs, async () => 'ok'); + + assert.throws( + () => harness.assertResponseStatus({rpcStatus: 'NOT_FOUND'}), + /rpc\.response\.status_code/, + ); + }); + }); + it('manages span lifetime for resolved promises', async () => { const result = await traceCall(dynamicArgs, staticArgs, () => Promise.resolve('async-result'), @@ -382,15 +702,18 @@ describe('TracerHelper', () => { cancel(): void {} then( onfulfilled?: - ((value: string) => TResult1 | PromiseLike) | null, + | ((value: string) => TResult1 | PromiseLike) + | null, onrejected?: - ((reason: unknown) => TResult2 | PromiseLike) | null, + | ((reason: unknown) => TResult2 | PromiseLike) + | null, ): Promise { return this.promise.then(onfulfilled, onrejected); } catch( onrejected?: - ((reason: unknown) => TResult | PromiseLike) | null, + | ((reason: unknown) => TResult | PromiseLike) + | null, ): Promise { return this.promise.catch(onrejected); } @@ -455,10 +778,7 @@ describe('TracerHelper', () => { const spans = harness.getSpans('google-gax'); assert.strictEqual(spans.length, 1); assert.strictEqual(spans[0].ended, true); - assert.strictEqual( - spans[0].attributes['error.message'], - 'ongoing call failed', - ); + assert.strictEqual(spans[0].status.message, 'ongoing call failed'); }); it('ends span synchronously if result is not a Promise', () => { @@ -490,10 +810,7 @@ describe('TracerHelper', () => { const spans = harness.getSpans('google-gax'); assert.strictEqual(spans.length, 1); assert.strictEqual(spans[0].ended, true); - assert.strictEqual( - spans[0].attributes['error.message'], - 'async promise failure', - ); + assert.strictEqual(spans[0].status.message, 'async promise failure'); assert.strictEqual(spans[0].events.length, 1); }); @@ -535,10 +852,7 @@ describe('TracerHelper', () => { const spans = harness.getSpans('google-gax'); assert.strictEqual(spans.length, 1); assert.strictEqual(spans[0].ended, true); - assert.strictEqual( - spans[0].attributes['error.message'], - 'stream failure', - ); + assert.strictEqual(spans[0].status.message, 'stream failure'); assert.strictEqual(spans[0].events.length, 1); assert.strictEqual(spans[0].events[0].name, 'exception'); }); @@ -647,7 +961,7 @@ describe('TracerHelper', () => { assert.strictEqual(spansAfterAttempt1.length, 1); assert.strictEqual(spansAfterAttempt1[0].ended, true); assert.strictEqual( - spansAfterAttempt1[0].attributes['error.message'], + spansAfterAttempt1[0].status.message, 'transient stream failure', ); assert.strictEqual(spansAfterAttempt1[0].events.length, 1); @@ -749,10 +1063,7 @@ describe('TracerHelper', () => { const spans = harness.getSpans('google-gax'); assert.strictEqual(spans.length, 1); assert.strictEqual(spans[0].ended, true); - assert.strictEqual( - spans[0].attributes['error.message'], - 'RPC Failed', - ); + assert.strictEqual(spans[0].status.message, 'RPC Failed'); assert.strictEqual( spans[0].attributes['exception.type'], undefined, @@ -915,18 +1226,21 @@ describe('TracerHelper', () => { return spans[0].status; }; - it('sets OK for a synchronous non-promise result', () => { + // A successful call leaves the status UNSET rather than setting OK. + // semconv reserves OK for an application overriding the + // instrumentation's judgement, so a library must never emit it. + it('leaves the status unset for a synchronous non-promise result', () => { traceCall( dynamicArgs, staticArgs, () => ({data: 1}) as unknown as ResultTuple, ); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); }); - it('sets OK when the promise resolves', async () => { + it('leaves the status unset when the promise resolves', async () => { await traceCall(dynamicArgs, staticArgs, async () => ({data: 1})); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); }); it('sets ERROR when the promise rejects', async () => { @@ -951,11 +1265,11 @@ describe('TracerHelper', () => { assert.strictEqual(status.message, 'sync boom'); }); - it('sets OK when the stream ends cleanly', () => { + it('leaves the status unset when the stream ends cleanly', () => { const emitter = new EventEmitter(); traceCall(dynamicArgs, staticArgs, () => emitter, true); emitter.emit('end'); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); }); it('sets ERROR when the stream errors', () => { @@ -967,7 +1281,7 @@ describe('TracerHelper', () => { assert.strictEqual(status.message, 'stream boom'); }); - it('sets OK when a client-streaming call finishes', async () => { + it('leaves the status unset when a client-streaming call finishes', async () => { const writable = new Writable({ objectMode: true, write(_chunk, _enc, cb) { @@ -978,10 +1292,10 @@ describe('TracerHelper', () => { writable.end(); await new Promise(resolve => setImmediate(resolve)); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); }); - it('sets OK when the callback reports success', () => { + it('leaves the status unset when the callback reports success', () => { let invokedCallback: APICallback | undefined; traceCall( dynamicArgs, @@ -994,7 +1308,7 @@ describe('TracerHelper', () => { () => {}, ); invokedCallback!(null, {ok: true}); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); }); it('sets ERROR when the callback reports failure', () => { @@ -1015,8 +1329,9 @@ describe('TracerHelper', () => { assert.strictEqual(status.message, 'callback boom'); }); - it('does not downgrade an ERROR status to OK when the span ends', () => { - // endSpan resolves the status centrally; a recorded error must win. + it('does not clear an ERROR status when the span ends', () => { + // endSpan resolves the outcome centrally; a recorded error must win + // over the success path, which would otherwise leave it UNSET. const emitter = new EventEmitter(); traceCall(dynamicArgs, staticArgs, () => emitter, true); emitter.emit('error', new Error('stream boom')); @@ -1048,13 +1363,13 @@ describe('TracerHelper', () => { ); invokedCallback!(null, {ok: true}); - assert.strictEqual(lastStatus().code, SpanStatusCode.OK); + assert.strictEqual(lastStatus().code, SpanStatusCode.UNSET); emitter.emit('error', new Error('too late')); const spans = harness.getSpans('google-gax'); assert.strictEqual(spans.length, 1); - assert.strictEqual(spans[0].status.code, SpanStatusCode.OK); + assert.strictEqual(spans[0].status.code, SpanStatusCode.UNSET); // No phantom exception event tacked onto the finished span. assert.strictEqual( spans[0].events.filter(e => e.name === 'exception').length,