如何从C#应用直接导出OpenTelemetry(Otel)数据至Azure Data Explorer(ADX)后端？

方案概述 #

你可以通过自定义OpenTelemetry导出器结合Azure Data Explorer（ADX）的批量摄入API实现直接导出，无需依赖OpenTelemetry Collector。以下是具体实现步骤：

步骤1：准备ADX环境 #

• 创建ADX集群和目标数据库
• 为C#应用创建服务主体，并赋予该主体数据库的ingestor权限（用于写入数据）
• 在ADX中创建对应Otel数据类型的表，示例结构：

  // 轨迹表示例
  .create table Traces (Timestamp:datetime, TraceId:string, SpanId:string, ParentSpanId:string, Name:string, Duration:timespan, Attributes:dynamic, StatusCode:string, StatusDescription:string)
  // 指标表示例
  .create table Metrics (Timestamp:datetime, Name:string, Value:real, Labels:dynamic)
  // 日志表示例
  .create table Logs (Timestamp:datetime, Severity:string, Message:string, Attributes:dynamic)
  

步骤2：安装必要的NuGet包 #

• 基础OpenTelemetry包：

  Install-Package OpenTelemetry
  Install-Package OpenTelemetry.Extensions.Hosting
  Install-Package OpenTelemetry.Instrumentation.AspNetCore // Web应用需安装
  

• ADX SDK包（用于调用摄入API）：

  Install-Package Microsoft.Azure.Kusto.Data
  Install-Package Microsoft.Azure.Kusto.Ingest
  

步骤3：实现自定义OpenTelemetry导出器 #

以导出轨迹数据为例，创建实现BaseExporter<Activity>的导出器：

using OpenTelemetry;
using OpenTelemetry.Trace;
using Microsoft.Azure.Kusto.Ingest;
using System.Collections.Generic;
using System.Threading;
using Newtonsoft.Json;

public class AdxTraceExporter : BaseExporter<Activity>
{
    private readonly IKustoIngestClient _ingestClient;
    private readonly string _databaseName;
    private readonly string _tableName;

    public AdxTraceExporter(string adxIngestUri, string databaseName, string tableName, string clientId, string clientSecret, string tenantId)
    {
        _databaseName = databaseName;
        _tableName = tableName;

        var kustoConnString = new KustoConnectionStringBuilder(adxIngestUri)
            .WithAadApplicationKeyAuthentication(clientId, clientSecret, tenantId);
        
        _ingestClient = KustoIngestFactory.CreateQueuedIngestClient(kustoConnString);
    }

    public override ExportResult Export(in Batch<Activity> batch)
    {
        var traceRows = new List<string>();
        foreach (var activity in batch)
        {
            // 将Activity转换为ADX兼容的CSV行格式
            var attributesJson = JsonConvert.SerializeObject(activity.Attributes.ToDictionary(kv => kv.Key, kv => kv.Value));
            var row = $"{activity.StartTimeUtc:yyyy-MM-dd HH:mm:ss.fff},{activity.TraceId},{activity.SpanId},{activity.ParentSpanId}," +
                      $"\"{activity.Name}\",{activity.Duration.TotalMilliseconds},{attributesJson}," +
                      $"\"{activity.Status}\",\"{activity.StatusDescription ?? string.Empty}\"";
            traceRows.Add(row);
        }

        // 批量提交至ADX
        var ingestProps = new KustoIngestionProperties(_databaseName, _tableName)
        {
            Format = DataSourceFormat.Csv,
            IgnoreFirstRecord = false
        };

        _ingestClient.IngestFromString(traceRows, ingestProps);
        return ExportResult.Success;
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing)
        {
            _ingestClient?.Dispose();
        }
        base.Dispose(disposing);
    }
}

步骤4：在C#应用中配置OpenTelemetry #

在Program.cs中注册导出器并启用相关 instrumentation：

var builder = WebApplication.CreateBuilder(args);

// 配置OpenTelemetry追踪
builder.Services.AddOpenTelemetry()
    .WithTracing(tracing => tracing
        .AddAspNetCoreInstrumentation()
        .AddConsoleExporter() // 可选，本地调试用
        .AddExporter(sp => new AdxTraceExporter(
            adxIngestUri: builder.Configuration["Adx:IngestUri"],
            databaseName: builder.Configuration["Adx:Database"],
            tableName: "Traces",
            clientId: builder.Configuration["Adx:ClientId"],
            clientSecret: builder.Configuration["Adx:ClientSecret"],
            tenantId: builder.Configuration["Adx:TenantId"])));

var app = builder.Build();
// 其他应用配置...
app.Run();

关键注意事项 #

• 批量写入优化：ADX适合批量写入，建议在导出器中设置批量阈值（比如攒够100条再提交），避免频繁小批量写入影响性能
• 数据格式校验：确保Otel数据转换后的格式与ADX表结构完全匹配，比如datetime格式、dynamic类型的JSON序列化规则
• 权限验证：服务主体必须拥有ADX数据库的ingestor权限，否则会写入失败
• 异常处理：生产环境需添加异常捕获和重试逻辑，避免数据丢失

内容的提问来源于stack exchange，提问作者shivi