feat: integrate OpenTelemetry tracing with @microlabs/otel-cf-workers (#1971)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Adam <[email protected]>
Co-authored-by: Aj Wazzan <[email protected]>

Author: devin-ai-integration[bot]

apps/server/package.json

@@ -35,9 +35,11 @@
     "@googleapis/gmail": "12.0.0",
     "@googleapis/people": "3.0.9",
     "@hono/trpc-server": "^0.3.4",
+    "@microlabs/otel-cf-workers": "1.0.0-rc.52",
     "@microsoft/microsoft-graph-client": "^3.0.7",
     "@microsoft/microsoft-graph-types": "^2.40.0",
     "@modelcontextprotocol/sdk": "1.15.1",
+    "@opentelemetry/api": "1.9.0",
     "@react-email/components": "^0.0.41",
     "@react-email/render": "1.1.0",
     "@sentry/cloudflare": "9.43.0",

apps/server/src/env.ts

@@ -94,6 +94,9 @@ export type ZeroEnv = {
   MEET_AUTH_HEADER: string;
   MEET_API_URL: string;
   ENABLE_MEET: 'true' | 'false';
+  OTEL_EXPORTER_OTLP_ENDPOINT?: string;
+  OTEL_EXPORTER_OTLP_HEADERS?: string;
+  OTEL_SERVICE_NAME?: string;
 };
 
 const env = _env as ZeroEnv;

apps/server/src/lib/tracing.ts

@@ -0,0 +1,9 @@
+import { trace, type Tracer, type Attributes } from '@opentelemetry/api';
+
+export const initTracing = (): Tracer => {
+  return trace.getTracer('zero-email-server', '1.0.0');
+};
+
+export const createSpan = (tracer: Tracer, name: string, attributes?: Attributes) => {
+  return tracer.startSpan(name, { attributes });
+};

apps/server/src/main.ts

@@ -22,6 +22,7 @@ import {
 } from './lib/attachments';
 import { SyncThreadsCoordinatorWorkflow } from './workflows/sync-threads-coordinator-workflow';
 import { WorkerEntrypoint, DurableObject, RpcTarget } from 'cloudflare:workers';
+import { instrument, type ResolveConfigFn } from '@microlabs/otel-cf-workers';
 import { getZeroAgent, getZeroDB, verifyToken } from './lib/server-utils';
 import { SyncThreadsWorkflow } from './workflows/sync-threads-workflow';
 import { ShardRegistry, ZeroAgent, ZeroDriver } from './routes/agent';
@@ -40,6 +41,7 @@ import { ZeroMCP } from './routes/agent/mcp';
 import { publicRouter } from './routes/auth';
 import { WorkflowRunner } from './pipelines';
 import { autumnApi } from './routes/autumn';
+import { initTracing } from './lib/tracing';
 import { env, type ZeroEnv } from './env';
 import type { HonoContext } from './ctx';
 import { createDb, type DB } from './db';
@@ -747,49 +749,111 @@ const app = new Hono<HonoContext>()
     }
   })
   .post('/a8n/notify/:providerId', async (c) => {
-    if (!c.req.header('Authorization')) return c.json({ error: 'Unauthorized' }, { status: 401 });
-    if (env.DISABLE_WORKFLOWS === 'true') return c.json({ message: 'OK' }, { status: 200 });
-    const providerId = c.req.param('providerId');
-    if (providerId === EProviders.google) {
-      const body = await c.req.json<{ historyId: string }>();
-      const subHeader = c.req.header('x-goog-pubsub-subscription-name');
-      if (!subHeader) {
-        console.log('[GOOGLE] no subscription header', body);
-        return c.json({}, { status: 200 });
+    const tracer = initTracing();
+    const span = tracer.startSpan('a8n_notify', {
+      attributes: {
+        'provider.id': c.req.param('providerId'),
+        'notification.type': 'email_notification',
+        'http.method': c.req.method,
+        'http.url': c.req.url,
+      },
+    });
+
+    try {
+      if (!c.req.header('Authorization')) {
+        span.setAttributes({ 'auth.status': 'missing' });
+        return c.json({ error: 'Unauthorized' }, { status: 401 });
       }
-      const isValid = await verifyToken(c.req.header('Authorization')!.split(' ')[1]);
-      if (!isValid) {
-        console.log('[GOOGLE] invalid request', body);
-        return c.json({}, { status: 200 });
+      if (env.DISABLE_WORKFLOWS === 'true') {
+        span.setAttributes({ 'workflows.disabled': true });
+        return c.json({ message: 'OK' }, { status: 200 });
       }
-      try {
-        await env.thread_queue.send({
-          providerId,
-          historyId: body.historyId,
-          subscriptionName: subHeader,
-        });
-      } catch (error) {
-        console.error('Error sending to thread queue', error, {
-          providerId,
-          historyId: body.historyId,
-          subscriptionName: subHeader,
+      const providerId = c.req.param('providerId');
+      if (providerId === EProviders.google) {
+        const body = await c.req.json<{ historyId: string }>();
+        const subHeader = c.req.header('x-goog-pubsub-subscription-name');
+
+        span.setAttributes({
+          'history.id': body.historyId,
+          'subscription.name': subHeader || 'missing',
         });
+
+        if (!subHeader) {
+          console.log('[GOOGLE] no subscription header', body);
+          span.setAttributes({ 'error.type': 'missing_subscription_header' });
+          return c.json({}, { status: 200 });
+        }
+        const isValid = await verifyToken(c.req.header('Authorization')!.split(' ')[1]);
+        if (!isValid) {
+          console.log('[GOOGLE] invalid request', body);
+          span.setAttributes({ 'auth.status': 'invalid' });
+          return c.json({}, { status: 200 });
+        }
+
+        span.setAttributes({ 'auth.status': 'valid' });
+
+        try {
+          await env.thread_queue.send({
+            providerId,
+            historyId: body.historyId,
+            subscriptionName: subHeader,
+          });
+          span.setAttributes({ 'queue.message_sent': true });
+        } catch (error) {
+          console.error('Error sending to thread queue', error, {
+            providerId,
+            historyId: body.historyId,
+            subscriptionName: subHeader,
+          });
+          span.recordException(error as Error);
+          span.setStatus({ code: 2, message: (error as Error).message });
+        }
+        return c.json({ message: 'OK' }, { status: 200 });
       }
-      return c.json({ message: 'OK' }, { status: 200 });
+    } catch (error) {
+      span.recordException(error as Error);
+      span.setStatus({ code: 2, message: (error as Error).message });
+      throw error;
+    } finally {
+      span.end();
     }
   });
+const handler = {
+  async fetch(request: Request, env: ZeroEnv, ctx: ExecutionContext): Promise<Response> {
+    return app.fetch(request, env, ctx);
+  },
+};
+
+const config: ResolveConfigFn = (env: ZeroEnv) => {
+  return {
+    exporter: {
+      url: env.OTEL_EXPORTER_OTLP_ENDPOINT || 'https://api.axiom.co/v1/traces',
+      headers: env.OTEL_EXPORTER_OTLP_HEADERS
+        ? Object.fromEntries(
+            env.OTEL_EXPORTER_OTLP_HEADERS.split(',').map((header: string) => {
+              const [key, value] = header.split('=');
+              return [key.trim(), value.trim()];
+            }),
+          )
+        : {},
+    },
+    service: {
+      name: env.OTEL_SERVICE_NAME || 'zero-email-server',
+      version: '1.0.0',
+    },
+  };
+};
+
 export default class Entry extends WorkerEntrypoint<ZeroEnv> {
   async fetch(request: Request): Promise<Response> {
-    // const url = new URL(request.url);
-    // if (url.pathname === '/__studio') {
-    //   return await studio(request, env.ZERO_DRIVER, {
-    //     basicAuth: { username: 'admin', password: 'password' },
-    //   });
-    // }
-    return app.fetch(request, this.env, this.ctx);
+    const instrumentedHandler = instrument(handler, config);
+    if (instrumentedHandler && instrumentedHandler.fetch) {
+      return instrumentedHandler.fetch(request as any, this.env, this.ctx);
+    }
+    return handler.fetch(request, this.env, this.ctx);
   }
   async queue(
-    batch: MessageBatch<any> | { queue: string; messages: Array<{ body: IEmailSendBatch }> },
+    batch: MessageBatch<unknown> | { queue: string; messages: Array<{ body: IEmailSendBatch }> },
   ) {
     switch (true) {
       case batch.queue.startsWith('subscribe-queue'): {
@@ -817,7 +881,7 @@ export default class Entry extends WorkerEntrypoint<ZeroEnv> {
             const { messageId, connectionId, mail } = msg.body;
 
             const { pending_emails_status: statusKV, pending_emails_payload: payloadKV } = this
-              .env as any;
+              .env as { pending_emails_status: KVNamespace; pending_emails_payload: KVNamespace };
 
             const status = await statusKV.get(messageId);
             if (status === 'cancelled') {
@@ -881,22 +945,41 @@ export default class Entry extends WorkerEntrypoint<ZeroEnv> {
         return;
       }
       case batch.queue.startsWith('thread-queue'): {
+        const tracer = initTracing();
+
         await Promise.all(
           batch.messages.map(async (msg: any) => {
-            const providerId = msg.body.providerId;
-            const historyId = msg.body.historyId;
-            const subscriptionName = msg.body.subscriptionName;
+            const span = tracer.startSpan('thread_queue_processing', {
+              attributes: {
+                'provider.id': msg.body.providerId,
+                'history.id': msg.body.historyId,
+                'subscription.name': msg.body.subscriptionName,
+                'queue.name': batch.queue,
+              },
+            });
 
             try {
+              const providerId = msg.body.providerId;
+              const historyId = msg.body.historyId;
+              const subscriptionName = msg.body.subscriptionName;
+
               const workflowRunner = env.WORKFLOW_RUNNER.get(env.WORKFLOW_RUNNER.newUniqueId());
               const result = await workflowRunner.runMainWorkflow({
                 providerId,
                 historyId,
                 subscriptionName,
               });
               console.log('[THREAD_QUEUE] result', result);
+              span.setAttributes({
+                'workflow.result': typeof result === 'string' ? result : JSON.stringify(result),
+                'workflow.success': true,
+              });
             } catch (error) {
               console.error('Error running workflow', error);
+              span.recordException(error as Error);
+              span.setStatus({ code: 2, message: (error as Error).message });
+            } finally {
+              span.end();
             }
           }),
         );
@@ -914,7 +997,10 @@ export default class Entry extends WorkerEntrypoint<ZeroEnv> {
 
   private async processScheduledEmails() {
     console.log('Checking for scheduled emails ready to be queued...');
-    const { scheduled_emails: scheduledKV, send_email_queue } = this.env as any;
+    const { scheduled_emails: scheduledKV, send_email_queue } = this.env as {
+      scheduled_emails: KVNamespace;
+      send_email_queue: Queue<IEmailSendBatch>;
+    };
 
     try {
       const now = Date.now();
@@ -991,7 +1077,7 @@ export default class Entry extends WorkerEntrypoint<ZeroEnv> {
 
       for (const key of listResp.keys) {
         try {
-          const wakeAtIso = (key as any).metadata?.wakeAt as string | undefined;
+          const wakeAtIso = key.metadata?.wakeAt as string | undefined;
           if (!wakeAtIso) continue;
           const wakeAt = new Date(wakeAtIso).getTime();
           if (wakeAt > nowTs) continue;

apps/server/src/pipelines.ts

@@ -24,6 +24,7 @@ import { Effect, Console, Logger } from 'effect';
 import { connection } from './db/schema';
 import { EProviders } from './types';
 import type { ZeroEnv } from './env';
+import { initTracing } from './lib/tracing';
 import { EPrompts } from './types';
 import { eq } from 'drizzle-orm';
 import { createDb } from './db';
@@ -140,6 +141,15 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
    * @returns
    */
   public runMainWorkflow(params: MainWorkflowParams) {
+    const tracer = initTracing();
+    const span = tracer.startSpan('workflow_main', {
+      attributes: {
+        'provider.id': params.providerId,
+        'history.id': params.historyId,
+        'subscription.name': params.subscriptionName
+      }
+    });
+
     return Effect.gen(this, function* () {
       yield* Console.log('[MAIN_WORKFLOW] Starting workflow with payload:', params);
 
@@ -148,9 +158,11 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
       const serviceAccount = getServiceAccount();
 
       const connectionId = yield* validateArguments(params, serviceAccount);
+      span.setAttributes({ 'connection.id': connectionId });
 
       if (!isValidUUID(connectionId)) {
         yield* Console.log('[MAIN_WORKFLOW] Invalid connection id format:', connectionId);
+        span.setAttributes({ 'error.type': 'invalid_connection_id' });
         return yield* Effect.fail({
           _tag: 'InvalidConnectionId' as const,
           connectionId,
@@ -165,6 +177,8 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
         }),
       }).pipe(Effect.orElse(() => Effect.succeed(null)));
 
+      span.setAttributes({ 'history.previous_id': previousHistoryId || 'none' });
+
       if (providerId === EProviders.google) {
         yield* Console.log('[MAIN_WORKFLOW] Processing Google provider workflow');
         yield* Console.log('[MAIN_WORKFLOW] Previous history ID:', previousHistoryId);
@@ -181,17 +195,26 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
         });
 
         yield* Console.log('[MAIN_WORKFLOW] Zero workflow result:', result);
+        span.setAttributes({ 'workflow.result': typeof result === 'string' ? result : JSON.stringify(result) });
       } else {
         yield* Console.log('[MAIN_WORKFLOW] Unsupported provider:', providerId);
+        span.setAttributes({ 'error.type': 'unsupported_provider' });
         return yield* Effect.fail({
           _tag: 'UnsupportedProvider' as const,
           providerId,
         });
       }
 
       yield* Console.log('[MAIN_WORKFLOW] Workflow completed successfully');
+      span.setAttributes({ 'workflow.success': true });
       return 'Workflow completed successfully';
     }).pipe(
+      Effect.tap(() => Effect.sync(() => span.end())),
+      Effect.tapError((error) => Effect.sync(() => {
+        span.recordException(error as unknown as Error);
+        span.setStatus({ code: 2, message: String(error) });
+        span.end();
+      })),
       Effect.tapError((error) => Console.log('[MAIN_WORKFLOW] Error in workflow:', error)),
       Effect.provide(loggerLayer),
       Effect.runPromise,
@@ -600,7 +623,8 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
           threadId: threadId.toString(),
           thread,
           foundConnection,
-          results: new Map<string, any>(),
+          results: new Map<string, unknown>(),
+          env: this.env,
         };
 
         // Execute configured workflows using the workflow engine
@@ -758,12 +782,13 @@ export class WorkflowRunner extends DurableObject<ZeroEnv> {
           threadId: threadId.toString(),
           thread,
           foundConnection,
-          results: new Map<string, any>(),
+          results: new Map<string, unknown>(),
+          env: this.env,
         };
 
         let workflowResults;
         try {
-          const allResults = new Map<string, any>();
+          const allResults = new Map<string, unknown>();
           const allErrors = new Map<string, Error>();
 
           const workflowNames = workflowEngine.getWorkflowNames();