使用 Durable Task Scheduler 的 OpenTelemetry 和分布式跟踪

分布式跟踪提供业务流程执行的端到端可见性。使用 Durable Task Scheduler 启用 OpenTelemetry 时，每个编排、活动和子编排都会生成关联的跨度，显示整个工作流中的计时、排序和错误。可以将这些跟踪导出到任何与 OpenTelemetry 兼容的后端，例如 Azure Monitor Application Insights、Jaeger 或 Zipkin。

Durable Functions 和独立的 Durable Task SDK 在使用 Durable Task Scheduler 作为后端时，均支持 OpenTelemetry 分布式跟踪。

工作原理

Durable Task SDKs 使用 OpenTelemetry span 自动为业务流程和活动提供检测。 SDK 为每个业务流程创建一个父范围，为每个活动调用、子业务流程和计时器创建子范围。跟踪上下文会在所有这些操作中自动传播，因此可以获得整个工作流程的单一关联的跟踪。

生成的跟踪树如下所示：

create_orchestration (client)
  └─ orchestration (server)
       ├─ activity:Step1
       ├─ activity:Step2
       └─ activity:Step3

无需将自定义仪表添加到协调器或活动代码中。将 Microsoft.DurableTask 活动源注册到 OpenTelemetry 配置，SDK 将处理其余内容。

先决条件

具有 Durable Functions 扩展 2.13.0 或更高版本的 Azure Functions 项目。
配置为函数应用的存储后端的持久任务计划程序。
用于查看跟踪的 OpenTelemetry 兼容后端（Application Insights、Jaeger 或其他 OTLP 收集器）。

.NET 8 SDK 或更高版本。
Microsoft.DurableTask.Worker.AzureManaged和 Microsoft.DurableTask.Client.AzureManaged NuGet 包。
OpenTelemetry、OpenTelemetry.Extensions.Hosting 和 OpenTelemetry.Exporter.OpenTelemetryProtocol NuGet 包。

与 OpenTelemetry 兼容的后端，用于查看跟踪，例如用于生产的 Application Insights 或用于本地开发的 Jaeger。

启用分布式跟踪

若要在 Durable Functions 中启用分布式跟踪，请更新 host.json 并配置与 OpenTelemetry 兼容的遥测后端。

更新 host.json

在 tracing 下的 durableTask 文件中添加节：

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "tracing": {
        "DistributedTracingEnabled": true,
        "Version": "V2"
      }
    }
  }
}

配置 Application Insights

在 APPLICATIONINSIGHTS_CONNECTION_STRING 函数应用中设置环境变量。

对于本地开发，请将其添加到 local.settings.json：

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "dotnet-isolated",
    "APPLICATIONINSIGHTS_CONNECTION_STRING": "<your-connection-string>"
  }
}

对于Azure托管的应用，请在 Azure 门户中的 Configuration 下将其添加为应用程序设置。

注意

如果您之前使用过 APPINSIGHTS_INSTRUMENTATIONKEY，请切换到 APPLICATIONINSIGHTS_CONNECTION_STRING 以获得最新功能。

减少遥测干扰

若要防止 Application Insights 采样跟踪数据，请从 Request中的采样规则中排除：

{
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  }
}

将 Microsoft.DurableTask 活动源注册到 OpenTelemetry 配置。当您注册此源时，Durable Task SDK 会自动为协调和活动创建时间跨度。

在工作进程的 Program.cs 中，使用持久任务活动源添加 OpenTelemetry 跟踪：

using Microsoft.DurableTask;
using Microsoft.DurableTask.Worker;
using Microsoft.DurableTask.Worker.AzureManaged;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using OpenTelemetry;
using OpenTelemetry.Resources;
using OpenTelemetry.Trace;

var builder = Host.CreateApplicationBuilder(args);

// Configure OpenTelemetry tracing
builder.Services.AddOpenTelemetry()
    .ConfigureResource(resource => resource.AddService("durable-worker"))
    .WithTracing(tracing =>
    {
        tracing
            .AddSource("Microsoft.DurableTask")
            .AddOtlpExporter(opts =>
            {
                opts.Endpoint = new Uri(
                    Environment.GetEnvironmentVariable("OTEL_EXPORTER_OTLP_ENDPOINT")
                    ?? "http://localhost:4317");
            });
    });

// Build connection string from environment variables
string endpoint = Environment.GetEnvironmentVariable("ENDPOINT") ?? "http://localhost:8080";
string taskHub = Environment.GetEnvironmentVariable("TASKHUB") ?? "default";
string connectionString = endpoint.Contains("localhost")
    ? $"Endpoint={endpoint};TaskHub={taskHub};Authentication=None"
    : $"Endpoint={endpoint};TaskHub={taskHub};Authentication=DefaultAzure";

// Configure Durable Task worker
builder.Services.AddDurableTaskWorker()
    .AddTasks(tasks =>
    {
        tasks.AddOrchestratorFunc<string, string>(
            "OrderProcessingOrchestration", async (ctx, input) =>
        {
            var validated = await ctx.CallActivityAsync<string>("ValidateOrder", input);
            var payment = await ctx.CallActivityAsync<string>("ProcessPayment", validated);
            var shipment = await ctx.CallActivityAsync<string>("ShipOrder", payment);
            var result = await ctx.CallActivityAsync<string>("SendNotification", shipment);
            return result;
        });

        tasks.AddActivityFunc<string, string>("ValidateOrder", (ctx, input) =>
            Task.FromResult($"Validated({input})"));
        tasks.AddActivityFunc<string, string>("ProcessPayment", (ctx, input) =>
            Task.FromResult($"Paid({input})"));
        tasks.AddActivityFunc<string, string>("ShipOrder", (ctx, input) =>
            Task.FromResult($"Shipped({input})"));
        tasks.AddActivityFunc<string, string>("SendNotification", (ctx, input) =>
            Task.FromResult($"Notified({input})"));
    })
    .UseDurableTaskScheduler(connectionString);

var host = builder.Build();
await host.RunAsync();

关键行是 .AddSource("Microsoft.DurableTask")，这告诉 OpenTelemetry 捕获 Durable Task SDK 发出的 span。

在导入任何 Durable Task 模块之前初始化 OpenTelemetry SDK。此顺序确保全局 TracerProvider 在 SDK 加载之前被注册。

// 1. OpenTelemetry SDK bootstrap — must run before other imports
import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base";
import { resourceFromAttributes } from "@opentelemetry/resources";
import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions";

const otlpEndpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT || "http://localhost:4318";

const sdk = new NodeSDK({
  resource: resourceFromAttributes({
    [ATTR_SERVICE_NAME]: "durable-worker",
  }),
  spanProcessors: [
    new SimpleSpanProcessor(
      new OTLPTraceExporter({ url: `${otlpEndpoint}/v1/traces` })
    ),
  ],
});
sdk.start();

// 2. Import Durable Task modules after OpenTelemetry is initialized
import {
  DurableTaskAzureManagedClientBuilder,
  DurableTaskAzureManagedWorkerBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

当全局注册 TracerProvider 时，Durable Task JS SDK 会自动发出跨度。

在创建 Durable Task 辅助角色之前配置 OpenTelemetry TracerProvider ：

import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import durabletask.worker as worker
import durabletask.task as task

# Configure OpenTelemetry
resource = Resource.create({"service.name": "durable-worker"})
provider = TracerProvider(resource=resource)
otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

# Define activities
def validate_order(ctx: task.ActivityContext, order_id: str) -> str:
    return f"Validated({order_id})"

def process_payment(ctx: task.ActivityContext, input: str) -> str:
    return f"Paid({input})"

def ship_order(ctx: task.ActivityContext, input: str) -> str:
    return f"Shipped({input})"

def send_notification(ctx: task.ActivityContext, input: str) -> str:
    return f"Notified({input})"

# Define orchestrator
def order_processing(ctx: task.OrchestrationContext, order_id: str):
    validated = yield ctx.call_activity(validate_order, input=order_id)
    paid = yield ctx.call_activity(process_payment, input=validated)
    shipped = yield ctx.call_activity(ship_order, input=paid)
    result = yield ctx.call_activity(send_notification, input=shipped)
    return result

# Start worker
if __name__ == "__main__":
    endpoint = os.environ.get("ENDPOINT", "http://localhost:8080")
    taskhub = os.environ.get("TASKHUB", "default")

    with worker.DurableTaskWorker(
        host_address=endpoint,
        secure_channel=not endpoint.startswith("http://localhost"),
        taskhub=taskhub,
    ) as w:
        w.add_orchestrator(order_processing)
        w.add_activity(validate_order)
        w.add_activity(process_payment)
        w.add_activity(ship_order)
        w.add_activity(send_notification)
        w.start()

当你配置 TracerProvider 时，持久任务 Python SDK 自动检测业务流程和活动。

当你全局注册OpenTelemetrySdk时，Java Durable Task SDK 会自动使用 OpenTelemetry 跨度来仪表化业务流程和活动。 SDK 使用 Microsoft.DurableTask 追踪器名称（与 .NET 活动源匹配），并创建 Server 类型的跨度用于编排和活动执行。它还跨客户端、业务流程、活动和子业务流程边界传播 W3C 跟踪上下文（traceparent/tracestate）。

将 OpenTelemetry 依赖项添加到 build.gradle：

def openTelemetryVersion = '1.58.0'

dependencies {
    implementation "com.microsoft:durabletask-client:1.7.0"
    implementation "com.microsoft:durabletask-azuremanaged:1.7.0"
    implementation "io.opentelemetry:opentelemetry-api:${openTelemetryVersion}"
    implementation "io.opentelemetry:opentelemetry-sdk:${openTelemetryVersion}"
    implementation "io.opentelemetry:opentelemetry-exporter-otlp:${openTelemetryVersion}"
}

在应用程序中使用 OTLP 导出程序配置 OpenTelemetry SDK：

import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
import io.opentelemetry.sdk.resources.Resource;
import io.opentelemetry.api.trace.Tracer;

String otlpEndpoint = System.getenv("OTEL_EXPORTER_OTLP_ENDPOINT");
if (otlpEndpoint == null) {
    otlpEndpoint = "http://localhost:4317";
}

OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder()
    .setEndpoint(otlpEndpoint)
    .build();

SdkTracerProvider tracerProvider = SdkTracerProvider.builder()
    .setResource(Resource.builder().put("service.name", "durable-worker").build())
    .addSpanProcessor(BatchSpanProcessor.builder(spanExporter).build())
    .build();

OpenTelemetrySdk openTelemetry = OpenTelemetrySdk.builder()
    .setTracerProvider(tracerProvider)
    .buildAndRegisterGlobal();

Tracer tracer = openTelemetry.getTracer("durable-worker");

buildAndRegisterGlobal() 调用将 OpenTelemetry SDK 全局注册。 Durable Task SDK 在内部使用 GlobalOpenTelemetry.getTracer("Microsoft.DurableTask") ，因此会自动选取此配置，并为业务流程和活动创建范围。

在 Application Insights 中查看跟踪

对于生产工作负荷，建议使用 Application Insights 遥测后端。

当在DistributedTracingEnabled中将true设置为Version并将V2设置为host.json之后，Durable Functions 应用程序会向 Application Insights 发出相关的跨度。若要查看 Azure 门户中的完整编排跟踪，请进行以下操作：

请在 Azure 门户中访问您的 Application Insights 资源。
打开 事务搜索 ，按名字或实例标识搜索编排。
选择一条追踪以查看端到端事务及其所有关联跨度。

跟踪显示编排为父跨度，子跨度对应于每个活动调用、子编排和计时器等待。以下模式生成不同的跟踪形状：

Pattern	跟踪形状
函数链式调用	顺序活动在协调器范围下嵌套。
扇出/扇入	并行活动在时间上重叠。
人际互动	一个执行协调的跨度，需要长时间等待外部事件。
Monitor	活动重复周期，计时器在各次迭代之间进行等待。

将 OTLP 导出程序配置为使用 Azure Monitor OpenTelemetry 导出程序将追踪数据发送到 Application Insights，或者通过 OTLP 将其导出到一个 OpenTelemetry 收集器，然后再转发到 Application Insights。

安装 Azure.Monitor.OpenTelemetry.Exporter NuGet 包，并将 OTLP 导出程序替换为：

builder.Services.AddOpenTelemetry()
    .ConfigureResource(resource => resource.AddService("durable-worker"))
    .WithTracing(tracing =>
    {
        tracing
            .AddSource("Microsoft.DurableTask")
            .AddAzureMonitorTraceExporter(opts =>
            {
                opts.ConnectionString = Environment.GetEnvironmentVariable(
                    "APPLICATIONINSIGHTS_CONNECTION_STRING");
            });
    });

安装 @azure/monitor-opentelemetry-exporter npm 包并配置导出程序：

import { AzureMonitorTraceExporter } from "@azure/monitor-opentelemetry-exporter";

const exporter = new AzureMonitorTraceExporter({
  connectionString: process.env.APPLICATIONINSIGHTS_CONNECTION_STRING,
});

const sdk = new NodeSDK({
  resource: resourceFromAttributes({
    [ATTR_SERVICE_NAME]: "durable-worker",
  }),
  spanProcessors: [new SimpleSpanProcessor(exporter)],
});
sdk.start();

azure-monitor-opentelemetry-exporter安装 pip 包并配置：

from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter

exporter = AzureMonitorTraceExporter(
    connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"]
)
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

将 Azure Monitor OpenTelemetry 出口程序用于 Java，或使用 Application Insights 连接字符串配置 OpenTelemetry Java 代理。

// Using the Azure Monitor OpenTelemetry exporter
// Add com.azure:azure-monitor-opentelemetry-exporter to your dependencies
import com.azure.monitor.opentelemetry.exporter.AzureMonitorExporterOptions;
import com.azure.monitor.opentelemetry.exporter.AzureMonitorTraceExporter;

AzureMonitorTraceExporter azureExporter = new AzureMonitorTraceExporter(
    new AzureMonitorExporterOptions()
        .connectionString(System.getenv("APPLICATIONINSIGHTS_CONNECTION_STRING")));

使用 Jaeger 在本地查看跟踪

若要进行本地开发，请使用持久任务计划程序模拟器及 Jaeger 来查看跟踪。使用一个 docker-compose.yml 来启动两个服务：

services:
  dts-emulator:
    image: mcr.microsoft.com/dts/dts-emulator:latest
    ports:
      - "8080:8080"  # gRPC
      - "8082:8082"  # Dashboard
  jaeger:
    image: jaegertracing/jaeger:latest
    ports:
      - "16686:16686"  # Jaeger UI
      - "4317:4317"    # OTLP gRPC
      - "4318:4318"    # OTLP HTTP

启动基础结构：

docker compose up -d

运行应用程序后，在 http://localhost:16686 打开 Jaeger UI，并搜索您的服务名称（例如，durable-worker）以查看跟踪。

对于使用 Durable Functions 进行本地开发，分布式跟踪数据将发送到 Application Insights。若要查看 Application Insights 中的跟踪，请转到 Azure 门户中的 Transaction 搜索或 Application Map。

对于本地可见性，你还可以通过向函数应用的 Program.cs 添加适当的 OpenTelemetry 包和导出器配置，在 Application Insights 旁边设置 OTLP 导出器。

跟踪数据的显示内容

持久任务 SDK 生成的跟踪数据包括：

范围类型	Description
`create_orchestration`	当安排新编排时发出的客户端追踪跨度
`orchestration`	涵盖完整执行生命周期的服务器端编排跨度
`activity:<name>`	每个活动调用的时间间隔，显示计时和结果
`sub_orchestration:<name>`	每个子编排调用的跨度
`timer`	持久计时器等待的时间段

每个范围包括属性，例如durabletask.type，durabletask.task.name和durabletask.task.instance_iddurabletask.task.task_id。失败的活动和编排流程包含在跨度状态和事件中的错误详细信息。

故障排除

問题	解决方案
没有任何痕迹出现	检查是否已注册 `Microsoft.DurableTask` 活动源并可访问导出器终结点。
跟踪日志不完整	请确保在初始化 Durable Task SDK（尤其是在 JavaScript/TypeScript 中）之前初始化 OpenTelemetry SDK。
Application Insights 中缺少时间段	禁用或调整采样设置，以防止删除跟踪数据。
跟踪信息不相关	检查是否使用持久任务计划程序作为后端。跟踪上下文传播需要调度器。

代码示例

持久化函数的分布式追踪 V2 示例（C#）

后续步骤

了解 Durable Functions 中的诊断信息

了解 Durable Task SDK 中的诊断

Last updated on 2026-04-24