diff --git a/contrib/interceptors-opentelemetry-v2/src/client/index.ts b/contrib/interceptors-opentelemetry-v2/src/client/index.ts index cd8090b1fa..e2dc316f95 100644 --- a/contrib/interceptors-opentelemetry-v2/src/client/index.ts +++ b/contrib/interceptors-opentelemetry-v2/src/client/index.ts @@ -13,13 +13,18 @@ import type { WorkflowTerminateInput, WorkflowCancelInput, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowClientInterceptor, TerminateWorkflowExecutionResponse, RequestCancelWorkflowExecutionResponse, DescribeWorkflowExecutionResponse, + WorkflowExecutionInfo, } from '@temporalio/client'; +import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, + instrumentSync, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -217,4 +222,29 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt }, }); } + + async fetchHistory( + input: WorkflowFetchHistoryInput, + next: Next + ): Promise { + return await instrument({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_FETCH_HISTORY, + fn: async (span) => { + span.setAttribute(WORKFLOW_ID_ATTR_KEY, input.workflowExecution.workflowId); + if (input.workflowExecution.runId) { + span.setAttribute(RUN_ID_ATTR_KEY, input.workflowExecution.runId); + } + return await next(input); + }, + }); + } + + list(input: WorkflowListInput, next: Next): AsyncIterable { + return instrumentSync({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_LIST, + fn: () => next(input), + }); + } } diff --git a/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts b/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts index e56fadfc6b..f82b28c0a7 100644 --- a/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts +++ b/contrib/interceptors-opentelemetry-v2/src/workflow/definitions.ts @@ -108,6 +108,16 @@ export enum SpanName { */ WORKFLOW_DESCRIBE = 'DescribeWorkflow', + /** + * Workflow history is fetched + */ + WORKFLOW_FETCH_HISTORY = 'FetchWorkflowHistory', + + /** + * Workflows are listed + */ + WORKFLOW_LIST = 'ListWorkflows', + /** * Workflow run is executing */ diff --git a/contrib/interceptors-opentelemetry/src/client/index.ts b/contrib/interceptors-opentelemetry/src/client/index.ts index 3e78488d27..9d1a9be9f3 100644 --- a/contrib/interceptors-opentelemetry/src/client/index.ts +++ b/contrib/interceptors-opentelemetry/src/client/index.ts @@ -13,13 +13,18 @@ import type { WorkflowTerminateInput, WorkflowCancelInput, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowClientInterceptor, TerminateWorkflowExecutionResponse, RequestCancelWorkflowExecutionResponse, DescribeWorkflowExecutionResponse, + WorkflowExecutionInfo, } from '@temporalio/client'; +import type { History } from '@temporalio/common/lib/proto-utils'; import { instrument, + instrumentSync, headersWithContext, RUN_ID_ATTR_KEY, WORKFLOW_ID_ATTR_KEY, @@ -215,4 +220,29 @@ export class OpenTelemetryWorkflowClientInterceptor implements WorkflowClientInt }, }); } + + async fetchHistory( + input: WorkflowFetchHistoryInput, + next: Next + ): Promise { + return await instrument({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_FETCH_HISTORY, + fn: async (span) => { + span.setAttribute(WORKFLOW_ID_ATTR_KEY, input.workflowExecution.workflowId); + if (input.workflowExecution.runId) { + span.setAttribute(RUN_ID_ATTR_KEY, input.workflowExecution.runId); + } + return await next(input); + }, + }); + } + + list(input: WorkflowListInput, next: Next): AsyncIterable { + return instrumentSync({ + tracer: this.tracer, + spanName: SpanName.WORKFLOW_LIST, + fn: () => next(input), + }); + } } diff --git a/contrib/interceptors-opentelemetry/src/workflow/definitions.ts b/contrib/interceptors-opentelemetry/src/workflow/definitions.ts index 73c3a23b8c..4dd77ade58 100644 --- a/contrib/interceptors-opentelemetry/src/workflow/definitions.ts +++ b/contrib/interceptors-opentelemetry/src/workflow/definitions.ts @@ -109,6 +109,16 @@ export enum SpanName { */ WORKFLOW_DESCRIBE = 'DescribeWorkflow', + /** + * Workflow history is fetched + */ + WORKFLOW_FETCH_HISTORY = 'FetchWorkflowHistory', + + /** + * Workflows are listed + */ + WORKFLOW_LIST = 'ListWorkflows', + /** * Workflow run is executing */ diff --git a/packages/client/src/interceptors.ts b/packages/client/src/interceptors.ts index eae5e03667..bcc9909e14 100644 --- a/packages/client/src/interceptors.ts +++ b/packages/client/src/interceptors.ts @@ -6,6 +6,7 @@ import type { Duration, SearchAttributePair, TypedSearchAttributes } from '@temporalio/common'; import { Headers, Next } from '@temporalio/common'; +import type { History } from '@temporalio/common/lib/proto-utils'; import type { temporal } from '@temporalio/proto'; import type { NexusOperationHandle } from './nexus-client'; import type { @@ -24,6 +25,7 @@ import type { RequestCancelWorkflowExecutionResponse, TerminateWorkflowExecutionResponse, WorkflowExecution, + WorkflowExecutionInfo, } from './types'; import type { CompiledWorkflowOptions, WorkflowUpdateOptions } from './workflow-options'; import type { ActivityHandle, ActivityOptions } from './activity-client'; @@ -128,6 +130,17 @@ export interface WorkflowDescribeInput { readonly workflowExecution: WorkflowExecution; } +/** Input for WorkflowClientInterceptor.fetchHistory */ +export interface WorkflowFetchHistoryInput { + readonly workflowExecution: WorkflowExecution; +} + +/** Input for WorkflowClientInterceptor.list */ +export interface WorkflowListInput { + readonly query?: string; + readonly pageSize?: number; +} + /** * Implement any of these methods to intercept {@link WorkflowClient} outbound calls * @@ -198,6 +211,14 @@ export interface WorkflowClientInterceptor { * Intercept a service call to describeWorkflowExecution */ describe?: (input: WorkflowDescribeInput, next: Next) => Promise; + /** + * Intercept a service call to getWorkflowExecutionHistory + */ + fetchHistory?: (input: WorkflowFetchHistoryInput, next: Next) => Promise; + /** + * Intercept a service call to listWorkflowExecutions + */ + list?: (input: WorkflowListInput, next: Next) => AsyncIterable; } /** @deprecated: Use {@link WorkflowClientInterceptor} instead */ diff --git a/packages/client/src/workflow-client.ts b/packages/client/src/workflow-client.ts index 6146bcb5b8..37dc79345d 100644 --- a/packages/client/src/workflow-client.ts +++ b/packages/client/src/workflow-client.ts @@ -55,6 +55,8 @@ import type { WorkflowClientInterceptor, WorkflowClientInterceptors, WorkflowDescribeInput, + WorkflowFetchHistoryInput, + WorkflowListInput, WorkflowQueryInput, WorkflowSignalInput, WorkflowSignalWithStartInput, @@ -1442,6 +1444,28 @@ export class WorkflowClient extends BaseClient { } } + /** + * Uses given input to make getWorkflowExecutionHistory call(s) to the service + * + * Used as the final function of the fetchHistory interceptor chain + */ + protected async _fetchHistoryHandler(input: WorkflowFetchHistoryInput): Promise { + let nextPageToken: Uint8Array | undefined = undefined; + const events = Array(); + for (;;) { + const response: temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse = + await this.workflowService.getWorkflowExecutionHistory({ + nextPageToken, + namespace: this.options.namespace, + execution: input.workflowExecution, + }); + events.push(...(response.history?.events ?? [])); + nextPageToken = response.nextPageToken; + if (nextPageToken == null || nextPageToken.length === 0) break; + } + return temporal.api.history.v1.History.create({ events }); + } + /** * Create a new workflow handle for new or existing Workflow execution */ @@ -1530,20 +1554,11 @@ export class WorkflowClient extends BaseClient { }; }, async fetchHistory() { - let nextPageToken: Uint8Array | undefined = undefined; - const events = Array(); - for (;;) { - const response: temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse = - await this.client.workflowService.getWorkflowExecutionHistory({ - nextPageToken, - namespace: this.client.options.namespace, - execution: { workflowId, runId }, - }); - events.push(...(response.history?.events ?? [])); - nextPageToken = response.nextPageToken; - if (nextPageToken == null || nextPageToken.length === 0) break; - } - return temporal.api.history.v1.History.create({ events }); + const next = this.client._fetchHistoryHandler.bind(this.client); + const fn = composeInterceptors(interceptors, 'fetchHistory', next); + return await fn({ + workflowExecution: { workflowId, runId }, + }); }, async startUpdate( def: UpdateDefinition | string, @@ -1627,16 +1642,16 @@ export class WorkflowClient extends BaseClient { }); } - protected async *_list(options?: ListOptions): AsyncIterable { + protected async *_list(input: WorkflowListInput): AsyncIterable { let nextPageToken: Uint8Array = Buffer.alloc(0); for (;;) { let response: temporal.api.workflowservice.v1.ListWorkflowExecutionsResponse; try { response = await this.workflowService.listWorkflowExecutions({ namespace: this.options.namespace, - query: options?.query, + query: input.query, nextPageToken, - pageSize: options?.pageSize, + pageSize: input.pageSize, }); } catch (e) { this.rethrowGrpcError(e, 'Failed to list workflows', undefined); @@ -1661,11 +1676,20 @@ export class WorkflowClient extends BaseClient { * https://docs.temporal.io/visibility */ public list(options?: ListOptions): AsyncWorkflowListIterable { + const input: WorkflowListInput = { + query: options?.query, + pageSize: options?.pageSize, + }; + // Deprecated interceptor factories require a workflowId and are skipped for client-level list. + const interceptors = Array.isArray(this.options.interceptors) + ? (this.options.interceptors as WorkflowClientInterceptor[]) + : []; + const list = composeInterceptors(interceptors, 'list', this._list.bind(this)); return { - [Symbol.asyncIterator]: () => this._list(options)[Symbol.asyncIterator](), + [Symbol.asyncIterator]: () => list(input)[Symbol.asyncIterator](), intoHistories: (intoHistoriesOptions?: IntoHistoriesOptions) => { return mapAsyncIterable( - this._list(options), + list(input), async ({ workflowId, runId }) => ({ workflowId, history: await this.getHandle(workflowId, runId).fetchHistory(), diff --git a/packages/test/src/test-interceptors.ts b/packages/test/src/test-interceptors.ts index 79ce01934a..9c03a1b83d 100644 --- a/packages/test/src/test-interceptors.ts +++ b/packages/test/src/test-interceptors.ts @@ -217,6 +217,47 @@ if (RUN_INTEGRATION_TESTS) { }); }); + test.serial('WorkflowClientInterceptor intercepts list and fetchHistory', async (t) => { + const taskQueue = 'test-interceptor-list-and-fetch-history'; + const workflowId = randomUUID(); + const worker = await Worker.create({ + ...defaultOptions, + taskQueue, + }); + let listCalls = 0; + let fetchHistoryCalls = 0; + const client = new WorkflowClient({ + interceptors: [ + { + list(input, next) { + listCalls += 1; + return next(input); + }, + async fetchHistory(input, next) { + fetchHistoryCalls += 1; + return next(input); + }, + }, + ], + }); + + await worker.runUntil(async () => { + await client.execute(successString, { + taskQueue, + workflowId, + }); + + const history = await client.getHandle(workflowId).fetchHistory(); + t.true((history.events?.length ?? 0) > 0); + t.is(fetchHistoryCalls, 1); + + for await (const _ of client.list({ query: `WorkflowId = "${workflowId}"` })) { + // consume iterator + } + t.is(listCalls, 1); + }); + }); + test.serial('Workflow continueAsNew can be intercepted', async (t) => { const taskQueue = 'test-continue-as-new-interceptor'; const worker = await Worker.create({