通过服务总线消息传递进行分布式跟踪和关联

服务总线 .NET 客户端自动跟踪

ServiceBusProcessor适用于 .NET 的 Azure 消息传送服务总线客户端的 类提供了可通过跟踪系统或客户端代码片段挂接的跟踪检测点。 使用检测可以从客户端跟踪对服务总线消息传递服务发出的所有调用。 如果消息处理是使用 ProcessMessageAsync 的 ServiceBusProcessor（消息处理程序模式）完成的，则还会检测消息处理。

使用 Azure Application Insights 进行跟踪

Microsoft Application Insights 提供丰富的性能监视功能，包括自动请求和依赖项跟踪。

请根据项目类型安装 Application Insights SDK：

• ASP.NET - 安装版本 2.5-beta2 或更高版本
• ASP.NET Core - 安装版本 2.2.0-beta2 或更高版本。 这些链接提供了有关安装 SDK、创建资源和配置 SDK（如果需要）的详细信息。 针对非 ASP.NET 应用程序，请参阅适用于控制台应用程序的 Azure Application Insights 一文。

如果你使用 ProcessMessageAsync 的 ServiceBusProcessor（消息处理程序模式）来处理消息，则还会检测消息处理。 系统会自动跟踪由服务完成的所有服务总线调用，并将其与其他遥测项关联。 否则，请参考以下示例手动进行消息处理跟踪。

跟踪消息处理

async Task ProcessAsync(ProcessMessageEventArgs args)
{
    ServiceBusReceivedMessage message = args.Message;
    if (message.ApplicationProperties.TryGetValue("Diagnostic-Id", out var objectId) && objectId is string diagnosticId)
    {
        var activity = new Activity("ServiceBusProcessor.ProcessMessage");
        activity.SetParentId(diagnosticId);
        // If you're using Microsoft.ApplicationInsights package version 2.6-beta or higher, you should call StartOperation<RequestTelemetry>(activity) instead
        using (var operation = telemetryClient.StartOperation<RequestTelemetry>("Process", activity.RootId, activity.ParentId))
        {
            telemetryClient.TrackTrace("Received message");
            try 
            {
            // process message
            }
            catch (Exception ex)
            {
                telemetryClient.TrackException(ex);
                operation.Telemetry.Success = false;
                throw;
            }

            telemetryClient.TrackTrace("Done");
        }
    }
}

在本示例中，系统针对每个已处理的消息报告请求遥测，并提供时间戳、持续时间和结果（成功）。 遥测功能也会提供一组关联属性。 在消息处理期间报告的嵌套跟踪和异常也带有关联属性的戳记，代表它们是 RequestTelemetry 的“子级”。

如果在消息处理期间对支持的外部组件发出调用，则还会自动跟踪和关联这些调用。 请参阅使用 Application Insights .NET SDK 跟踪自定义操作来了解手动跟踪和关联。

如果除 Application Insights SDK 之外还运行了外部代码，则在查看 Application Insights 日志时，你会发现持续时间延长。

这并不意味着接收消息时存在延迟。 在这种情况下，消息已经被接收，因为消息以参数的形式传递给 SDK 代码。 而且，App Insights 日志（进程）中的“名称”标记指示消息正在由外部事件处理代码处理 。 此问题与 Azure 无关。 相反，这些指标指示的是外部代码的效率，前提是已从服务总线接收到消息。

使用 OpenTelemetry 进行跟踪

服务总线 .NET 客户端库版本 7.5.0 及更高版本支持实验模式下的 OpenTelemetry。 有关详细信息，请参阅 .NET SDK 中的分布式跟踪。

在没有跟踪系统的情况下进行跟踪

如果跟踪系统不支持自动服务总线调用跟踪，可以考虑将此类支持添加到跟踪系统或应用程序中。 本部分介绍服务总线 .NET 客户端发送的诊断事件。

使用 .NET 跟踪基元 System.Diagnostics.Activity 和 System.Diagnostics.DiagnosticSource 检测服务总线 .NET 客户端。

Activity 充当跟踪上下文，而 DiagnosticSource 是通知机制。

如果没有针对 DiagnosticSource 事件的侦听器，检测会关闭，以避免产生检测成本。 DiagnosticSource 将所有控制权授予侦听器：

• 侦听器控制要侦听的源和事件
• 侦听器控制事件速率和采样
• 事件连同某个有效负载一起发送，该有效负载可提供完整上下文，使你能够在事件发生期间访问和修改 Message 对象

在继续实现之前，请先通过 DiagnosticSource 用户指南熟悉相关流程。

让我们在 ASP.NET Core 应用中针对服务总线事件创建一个侦听器，以通过 Microsoft.Extension.Logger 写入日志。 该侦听器使用 System.Reactive.Core 库来订阅 DiagnosticSource（不使用它也能轻松订阅 DiagnosticSource）

public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory factory, IApplicationLifetime applicationLifetime)
{
    // configuration...

    var serviceBusLogger = factory.CreateLogger("Azure.Messaging.ServiceBus");

    IDisposable innerSubscription = null;
    IDisposable outerSubscription = DiagnosticListener.AllListeners.Subscribe(delegate (DiagnosticListener listener)
    {
        // subscribe to the Service Bus DiagnosticSource
        if (listener.Name == "Azure.Messaging.ServiceBus")
        {
            // receive event from Service Bus DiagnosticSource
            innerSubscription = listener.Subscribe(delegate (KeyValuePair<string, object> evnt)
            {
                // Log operation details once it's done
                if (evnt.Key.EndsWith("Stop"))
                {
                    Activity currentActivity = Activity.Current;
                    serviceBusLogger.LogInformation($"Operation {currentActivity.OperationName} is finished, Duration={currentActivity.Duration}, Id={currentActivity.Id}, StartTime={currentActivity.StartTimeUtc}");
                }
            });
        }
    });

    applicationLifetime.ApplicationStopping.Register(() =>
    {
        outerSubscription?.Dispose();
        innerSubscription?.Dispose();
    });
}

在此示例中，侦听器记录每个服务总线操作的持续时间、结果、唯一标识符和开始时间。

事件

所有事件都会具有以下属性，这些属性符合开放式遥测规范： https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/api.md 。

• message_bus.destination - 队列/主题/订阅路径
• peer.address - 完全限定的命名空间
• kind - 生成者、使用者或客户端。 发送消息时使用生成者，接收时使用使用者，处置时使用客户端。
• component - servicebus

所有事件还具有 Entity 和 Endpoint 属性。

• Entity - 实体（队列、主题等）的名称
• Endpoint - 服务总线终结点 URL

检测的操作

下面是检测的操作的完整列表：

筛选和采样

在某些情况下，我们可能只需记录一部分事件，以减少性能开销或存储耗用量。 可以只记录“Stop”事件（如上述示例所示），或者对特定百分比的事件采样。 使用 DiagnosticSource 可以通过 IsEnabled 谓词实现此目的。 有关详细信息，请参阅 DiagnosticSource 中基于上下文的筛选。

可以针对单个操作调用 IsEnabled 多次，以将性能影响降到最低。

按以下顺序调用 IsEnabled：

1. IsEnabled(<OperationName>, string entity, null)，例如 IsEnabled("ServiceBusSender.Send", "MyQueue1")。 请注意，末尾没有“Start”或“Stop”。 使用此语句可以筛选出特定的操作或队列。 如果回调方法返回 false，则表示未发送操作的事件。

   • 对于“Process”和“ProcessSession”操作，还会收到 IsEnabled(<OperationName>, string entity, Activity activity) 回调。 使用此回调可根据 activity.Id 或 Tags 属性筛选事件。

2. IsEnabled(<OperationName>.Start)，例如 IsEnabled("ServiceBusSender.Send.Start")。 检查是否应激发“Start”事件。 结果只影响“Start”事件，但进一步的检测不依赖它。

“Stop”事件没有 IsEnabled。

如果某个操作结果为异常，则会调用 IsEnabled("ServiceBusSender.Send.Exception")。 只能订阅“Exception”事件并阻止剩余的检测。 在此情况下，仍需处理此类异常。 由于其他检测已被禁用，因此跟踪上下文不会连同消息一起从使用者流向生成者。

也可以使用 IsEnabled 来实现采样策略。 基于 Activity.Id 或 Activity.RootId 的采样可确保所有尝试都获取一致的采样结果（前提是采样内容通过跟踪系统或你自己的代码传播）。

如果同一个源存在多个 DiagnosticSource 侦听器，则只需一个侦听器接受事件，因此无法保证调用 IsEnabled。

服务总线 .NET 客户端自动跟踪

从版本 3.0.0 开始，适用于 .NET 的 Azure 服务总线客户端提供可由跟踪系统或客户端代码片段挂接的跟踪检测点。 使用检测可以从客户端跟踪对服务总线消息传递服务发出的所有调用。 如果消息处理是通过消息处理程序模式完成的，则还会检测消息处理

使用 Azure Application Insights 进行跟踪

Microsoft Application Insights 提供丰富的性能监视功能，包括自动请求和依赖项跟踪。

请根据项目类型安装 Application Insights SDK：

• ASP.NET - 安装版本 2.5-beta2 或更高版本
• ASP.NET Core - 安装版本 2.2.0-beta2 或更高版本。 这些链接提供了有关安装 SDK、创建资源和配置 SDK（如果需要）的详细信息。 针对非 ASP.NET 应用程序，请参阅适用于控制台应用程序的 Azure Application Insights 一文。

如果你使用消息处理程序模式来处理消息，则无需执行其他操作，系统会自动跟踪由服务完成的所有服务总线调用，并将其与其他遥测项关联。 否则，请参考以下示例手动进行消息处理跟踪。

跟踪消息处理

private readonly TelemetryClient telemetryClient;

async Task ProcessAsync(Message message)
{
    var activity = message.ExtractActivity();
    
    // If you're using Microsoft.ApplicationInsights package version 2.6-beta or higher, you should call StartOperation<RequestTelemetry>(activity) instead
    using (var operation = telemetryClient.StartOperation<RequestTelemetry>("Process", activity.RootId, activity.ParentId))
    {
        telemetryClient.TrackTrace("Received message");
        try 
        {
           // process message
        }
        catch (Exception ex)
        {
             telemetryClient.TrackException(ex);
             operation.Telemetry.Success = false;
             throw;
        }

        telemetryClient.TrackTrace("Done");
   }
}

在本示例中，系统针对每个已处理的消息报告 RequestTelemetry，并提供时间戳、持续时间和结果（成功）。 遥测功能也会提供一组关联属性。 在消息处理期间报告的嵌套跟踪和异常也带有关联属性的戳记，代表它们是 RequestTelemetry 的“子级”。

如果在消息处理期间对支持的外部组件发出调用，则还会自动跟踪和关联这些调用。 请参阅使用 Application Insights .NET SDK 跟踪自定义操作来了解手动跟踪和关联。

如果除 Application Insights SDK 之外还运行了外部代码，则在查看 Application Insights 日志时，你会发现持续时间延长。

这并不意味着接收消息时存在延迟。 在这种情况下，消息已经被接收，因为消息以参数的形式传递给 SDK 代码。 而且，App Insights 日志（进程）中的“名称”标记指示消息正在由外部事件处理代码处理 。 此问题与 Azure 无关。 相反，这些指标指示的是外部代码的效率，前提是已从服务总线接收到消息。 请参阅 GitHub 上的此文件，了解从服务总线接收到消息后，生成和分配“进程”标记的位置。

在没有跟踪系统的情况下进行跟踪

如果跟踪系统不支持自动服务总线调用跟踪，可以考虑将此类支持添加到跟踪系统或应用程序中。 本部分介绍服务总线 .NET 客户端发送的诊断事件。

使用 .NET 跟踪基元 System.Diagnostics.Activity 和 System.Diagnostics.DiagnosticSource 检测服务总线 .NET 客户端。

Activity 充当跟踪上下文，而 DiagnosticSource 是通知机制。

如果没有针对 DiagnosticSource 事件的侦听器，检测会关闭，以避免产生检测成本。 DiagnosticSource 将所有控制权授予侦听器：

• 侦听器控制要侦听的源和事件
• 侦听器控制事件速率和采样
• 事件连同某个有效负载一起发送，该有效负载可提供完整上下文，使你能够在事件发生期间访问和修改 Message 对象

在继续实现之前，请先通过 DiagnosticSource 用户指南熟悉相关流程。

让我们在 ASP.NET Core 应用中针对服务总线事件创建一个侦听器，以通过 Microsoft.Extension.Logger 写入日志。 该侦听器使用 System.Reactive.Core 库来订阅 DiagnosticSource（不使用它也能轻松订阅 DiagnosticSource）

public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory factory, IApplicationLifetime applicationLifetime)
{
    // configuration...

    var serviceBusLogger = factory.CreateLogger("Microsoft.Azure.ServiceBus");

    IDisposable innerSubscription = null;
    IDisposable outerSubscription = DiagnosticListener.AllListeners.Subscribe(delegate (DiagnosticListener listener)
    {
        // subscribe to the Service Bus DiagnosticSource
        if (listener.Name == "Microsoft.Azure.ServiceBus")
        {
            // receive event from Service Bus DiagnosticSource
            innerSubscription = listener.Subscribe(delegate (KeyValuePair<string, object> evnt)
            {
                // Log operation details once it's done
                if (evnt.Key.EndsWith("Stop"))
                {
                    Activity currentActivity = Activity.Current;
                    TaskStatus status = (TaskStatus)evnt.Value.GetProperty("Status");
                    serviceBusLogger.LogInformation($"Operation {currentActivity.OperationName} is finished, Duration={currentActivity.Duration}, Status={status}, Id={currentActivity.Id}, StartTime={currentActivity.StartTimeUtc}");
                }
            });
        }
    });

    applicationLifetime.ApplicationStopping.Register(() =>
    {
        outerSubscription?.Dispose();
        innerSubscription?.Dispose();
    });
}

在此示例中，侦听器记录每个服务总线操作的持续时间、结果、唯一标识符和开始时间。

事件

对于每个操作，将发送两个事件：“Start”和“Stop”。 你很可能只对“Stop”事件感兴趣。 这些事件提供操作的结果，并以 Activity 属性的形式提供开始时间和持续时间。

每个事件有效负载为侦听器提供操作上下文，并复制 API 传入参数和返回值。 “Stop”事件有效负载具有“Start”事件有效负载的所有属性，因此可以完全忽略“Start”事件。

所有事件还具有“Entity”和“Endpoint”属性。

• string Entity - 实体（队列、主题等）的名称
• Uri Endpoint - 服务总线终结点 URL

每个“Stop”事件具有表示 Status 异步操作已完成的 TaskStatus 属性，为便于阅读，下表中也省略了该属性。

下面是检测的操作的完整列表：

在每个事件中，可以访问保存当前操作上下文的 Activity.Current。

将更多属性记录到日志中

Activity.Current 提供当前操作及其父级的详细上下文。 有关详细信息，请参阅 Activity 文档。 服务总线检测在 Activity.Current.Tags 中提供更多信息 - 这些信息包含 MessageId 和 SessionId（如果已提供）。

跟踪“Receive”、“Peek”和“ReceiveDeferred”事件的活动还可能带有 RelatedTo 标记。 该标记包含作为结果收到的消息的 Diagnostic-Id 相异列表。 此类操作可能导致收到多个不相关的消息。 此外，当操作启动时，Diagnostic-Id 未知，因此，只能使用此标记将“Receive”操作关联到“Process”操作。 此标记在分析性能问题时相当有用，可以检查接收消息所花费的时间。

记录标记的有效方法是循环访问它们，将标记添加到上述示例后，结果如下所示：

Activity currentActivity = Activity.Current;
TaskStatus status = (TaskStatus)evnt.Value.GetProperty("Status");

var tagsList = new StringBuilder();
foreach (var tags in currentActivity.Tags)
{
    tagsList.Append($", {tags.Key}={tags.Value}");
}

serviceBusLogger.LogInformation($"{currentActivity.OperationName} is finished, Duration={currentActivity.Duration}, Status={status}, Id={currentActivity.Id}, StartTime={currentActivity.StartTimeUtc}{tagsList}");

筛选和采样

在某些情况下，我们可能只需记录一部分事件，以减少性能开销或存储耗用量。 可以只记录“Stop”事件（如上述示例所示），或者对特定百分比的事件采样。 使用 DiagnosticSource 可以通过 IsEnabled 谓词实现此目的。 有关详细信息，请参阅 DiagnosticSource 中基于上下文的筛选。

可以针对单个操作调用 IsEnabled 多次，以将性能影响降到最低。

按以下顺序调用 IsEnabled：

1. IsEnabled(<OperationName>, string entity, null)，例如 IsEnabled("Microsoft.Azure.ServiceBus.Send", "MyQueue1")。 请注意，末尾没有“Start”或“Stop”。 使用此语句可以筛选出特定的操作或队列。 如果回调方法返回 false，则表示未发送操作的事件。

   • 对于“Process”和“ProcessSession”操作，还会收到 IsEnabled(<OperationName>, string entity, Activity activity) 回调。 使用此回调可根据 activity.Id 或 Tags 属性筛选事件。

2. IsEnabled(<OperationName>.Start)，例如 IsEnabled("Microsoft.Azure.ServiceBus.Send.Start")。 检查是否应激发“Start”事件。 结果只影响“Start”事件，但进一步的检测不依赖它。

“Stop”事件没有 IsEnabled。

如果某个操作结果为异常，则会调用 IsEnabled("Microsoft.Azure.ServiceBus.Exception")。 只能订阅“Exception”事件并阻止剩余的检测。 在此情况下，仍需处理此类异常。 由于其他检测已被禁用，因此跟踪上下文不会连同消息一起从使用者流向生成者。

也可以使用 IsEnabled 来实现采样策略。 基于 Activity.Id 或 Activity.RootId 的采样可确保所有尝试都获取一致的采样结果（前提是采样内容通过跟踪系统或你自己的代码传播）。

如果同一个源存在多个 DiagnosticSource 侦听器，则只需一个侦听器接受事件，因此无法保证调用 IsEnabled。