使用 Temporal 进行追踪

import asyncio
from temporalio.client import Client
from temporalio.contrib.opentelemetry import TracingInterceptor

# 导入相同的追踪器设置
from worker import init_tracer_provider


async def main():
    provider = init_tracer_provider()

    try:
        client = await Client.connect(
            "localhost:7233",
            interceptors=[TracingInterceptor()],
        )

        # 执行工作流
        result = await client.execute_workflow(
            MyWorkflow.run,
            "Hello World",
            id="my-workflow-1",
            task_queue="my-task-queue",
        )
        print(f"Workflow result: {result}")

    finally:
        provider.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';

export function initTracerProvider(): NodeTracerProvider {
  // 为 LangSmith 创建 OTLP 导出器
  const exporter = new OTLPTraceExporter({
    url: 'https://api.smith.langchain.com/otel/v1/traces',
    headers: {
      'x-api-key': process.env.LANGSMITH_API_KEY || '',
      'Langsmith-Project': process.env.LANGSMITH_PROJECT || 'default',
    },
  });

  // 创建 TracerProvider
  const provider = new NodeTracerProvider({
    resource: new Resource({
      [ATTR_SERVICE_NAME]: 'temporal-worker',
    }),
  });

  provider.addSpanProcessor(new BatchSpanProcessor(exporter));
  provider.register();

  return provider;
}

import { proxyActivities } from '@temporalio/workflow';
import type * as activities from './activities';

const { processActivity } = proxyActivities<typeof activities>({
  startToCloseTimeout: '10 seconds',
});

export async function myWorkflow(input: string): Promise<string> {
  return await processActivity(input);
}

import { log } from '@temporalio/activity';
import { trace } from '@opentelemetry/api';

export async function processActivity(input: string): Promise<string> {
  log.info('Processing', { input });

  // 获取当前跨度并添加 Gen AI 属性
  const span = trace.getActiveSpan();
  span?.setAttribute('gen_ai.prompt', input);
  span?.setAttribute('gen_ai.operation.name', 'chat');

  const result = `Processed: ${input}`;

  span?.setAttribute('gen_ai.completion', result);

  return result;
}

import { Worker, NativeConnection } from '@temporalio/worker';
import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import {
  makeWorkflowExporter,
  OpenTelemetryActivityInboundInterceptor,
} from '@temporalio/interceptors-opentelemetry';
import { trace } from '@opentelemetry/api';

import * as activities from './activities';
import { initTracerProvider } from './tracer';

async function run() {
  const provider = initTracerProvider();

  try {
    const connection = await NativeConnection.connect({
      address: 'localhost:7233',
    });

    const worker = await Worker.create({
      connection,
      namespace: 'default',
      taskQueue: 'my-task-queue',
      workflowsPath: require.resolve('./workflows'),
      activities,
      sinks: {
        exporter: makeWorkflowExporter(
          trace.getTracer('temporal-app'),
          new Resource({ [ATTR_SERVICE_NAME]: 'temporal-worker' })
        ),
      },
      interceptors: {
        activity: [() => ({ inbound: new OpenTelemetryActivityInboundInterceptor() })],
      },
    });

    console.log('Starting worker...');
    await worker.run();
  } finally {
    await provider.shutdown();
  }
}

run().catch((err) => {
  console.error(err);
  process.exit(1);
});

import { Client, Connection } from '@temporalio/client';
import { initTracerProvider } from './tracer';

async function run() {
  // 初始化追踪
  const provider = initTracerProvider();

  try {
    const connection = await Connection.connect({ address: 'localhost:7233' });
    const client = new Client({ connection });

    const result = await client.workflow.execute('myWorkflow', {
      taskQueue: 'my-task-queue',
      workflowId: 'my-workflow-1',
      args: ['Hello World'],
    });

    console.log('Workflow result:', result);
  } finally {
    await provider.shutdown();
  }
}

run().catch(console.error);