core: add span links + message ids to queue spans (#472)

* core: add span links + message ids to queue spans

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* add queue overhead to metrics

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* fix types

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* rename messageId to message.id

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

* add changeset

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

---------

Signed-off-by: Gal Schlezinger <gal@spitfire.co.il>

.changeset/neat-comics-care.md

@@ -0,0 +1,6 @@
+---
+"@workflow/world": patch
+"@workflow/core": patch
+---
+
+track queue overhead with opentelemetry

packages/core/src/runtime/resume-hook.ts

@@ -1,7 +1,6 @@
 import { waitUntil } from '@vercel/functions';
 import { ERROR_SLUGS, WorkflowRuntimeError } from '@workflow/errors';
-import type { Hook } from '@workflow/world';
-import type { WorkflowInvokePayload } from '../schemas.js';
+import type { Hook, WorkflowInvokePayload } from '@workflow/world';
 import {
   dehydrateStepReturnValue,
   hydrateStepArguments,

packages/core/src/runtime/start.ts

@@ -1,8 +1,9 @@
 import { waitUntil } from '@vercel/functions';
 import { WorkflowRuntimeError } from '@workflow/errors';
 import { withResolvers } from '@workflow/utils';
+import type { WorkflowInvokePayload } from '@workflow/world';
 import { Run } from '../runtime.js';
-import type { Serializable, WorkflowInvokePayload } from '../schemas.js';
+import type { Serializable } from '../schemas.js';
 import { dehydrateWorkflowArguments } from '../serialization.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { serializeTraceCarrier, trace } from '../telemetry.js';

packages/core/src/schemas.ts

@@ -1,24 +1,3 @@
-import { z } from 'zod';
-
-// OpenTelemetry trace context for distributed tracing
-const TraceCarrierSchema = z.record(z.string(), z.string());
-
-export const WorkflowInvokePayloadSchema = z.object({
-  runId: z.string(),
-  traceCarrier: TraceCarrierSchema.optional(),
-});
-
-export const StepInvokePayloadSchema = z.object({
-  workflowName: z.string(),
-  workflowRunId: z.string(),
-  workflowStartedAt: z.number(),
-  stepId: z.string(),
-  traceCarrier: TraceCarrierSchema.optional(),
-});
-
-export type WorkflowInvokePayload = z.infer<typeof WorkflowInvokePayloadSchema>;
-export type StepInvokePayload = z.infer<typeof StepInvokePayloadSchema>;
-
 /**
  * A serializable value:
  * Any valid JSON object is serializable

packages/core/src/telemetry.ts

@@ -1,4 +1,5 @@
-import type { Span, SpanOptions } from '@opentelemetry/api';
+import type * as api from '@opentelemetry/api';
+import type { Span, SpanKind, SpanOptions } from '@opentelemetry/api';
 import { once } from '@workflow/utils';
 
 // ============================================================
@@ -118,9 +119,7 @@ export async function getSpanContextForTraceCarrier(
 }
 
 export async function getActiveSpan() {
-  const otel = await OtelApi.value;
-  if (!otel) return null;
-  return otel.trace.getActiveSpan();
+  return await withOtel((otel) => otel.trace.getActiveSpan());
 }
 
 export function instrumentObject<T extends object>(prefix: string, o: T): T {
@@ -137,3 +136,23 @@ export function instrumentObject<T extends object>(prefix: string, o: T): T {
   }
   return handlers;
 }
+
+export async function getSpanKind(field: keyof typeof SpanKind) {
+  return withOtel((x) => x.SpanKind[field]);
+}
+
+export async function withOtel<T>(
+  fn: (otel: typeof api) => T
+): Promise<Awaited<T> | undefined> {
+  const otel = await OtelApi.value;
+  if (!otel) return undefined;
+  return await fn(otel);
+}
+
+export function linkToCurrentContext(): Promise<[api.Link] | undefined> {
+  return withOtel((otel): [api.Link] | undefined => {
+    const context = otel.trace.getActiveSpan()?.spanContext();
+    if (!context) return;
+    return [{ context }];
+  });
+}

packages/core/src/telemetry/semantic-conventions.ts

@@ -37,15 +37,16 @@
  * @packageDocumentation
  */
 
-import type { Step, WorkflowRun } from '@workflow/world';
+import type { MessageId, Step, WorkflowRun } from '@workflow/world';
 
 /**
  * Creates a semantic convention function that returns an attribute object.
  * @param name - The attribute name following OpenTelemetry semantic conventions
  * @returns A function that takes a value and returns an attribute object
  */
-function SemanticConvention<T>(name: string) {
-  return (value: T) => ({ [name]: value });
+function SemanticConvention<T>(...names: string[]) {
+  return (value: T) =>
+    Object.fromEntries(names.map((name) => [name, value] as const));
 }
 
 // Workflow attributes
@@ -173,6 +174,15 @@ export const StepRetryWillRetry = SemanticConvention<boolean>(
 /** Name of the queue being used for message processing */
 export const QueueName = SemanticConvention<string>('queue.name');
 
+/** The message id being handled */
+export const QueueMessageId = SemanticConvention<MessageId>(
+  'messaging.message.id',
+  'queue.message.id'
+);
+
+/** Time taken to enqueue the message in milliseconds */
+export const QueueOverheadMs = SemanticConvention<number>('queue.overhead_ms');
+
 // Deployment attributes
 
 /** Unique identifier for the deployment environment */

packages/world/src/index.ts

@@ -13,7 +13,9 @@ export {
   MessageId,
   QueuePayloadSchema,
   QueuePrefix,
+  StepInvokePayloadSchema,
   ValidQueueName,
+  WorkflowInvokePayloadSchema,
 } from './queue.js';
 export type * from './runs.js';
 export {

packages/world/src/queue.ts

@@ -24,6 +24,7 @@ export type TraceCarrier = z.infer<typeof TraceCarrierSchema>;
 export const WorkflowInvokePayloadSchema = z.object({
   runId: z.string(),
   traceCarrier: TraceCarrierSchema.optional(),
+  requestedAt: z.coerce.date().optional(),
 });
 
 export const StepInvokePayloadSchema = z.object({
@@ -32,6 +33,7 @@ export const StepInvokePayloadSchema = z.object({
   workflowStartedAt: z.number(),
   stepId: z.string(),
   traceCarrier: TraceCarrierSchema.optional(),
+  requestedAt: z.coerce.date().optional(),
 });
 
 export type WorkflowInvokePayload = z.infer<typeof WorkflowInvokePayloadSchema>;