当前位置：首页 > news >正文

Orleans 可观测性实战：基于源码的指标与分布式追踪（含 Prometheus 集成）

news 2025/10/21 11:42:38

概览与原则
指标体系（Metrics）
- Meter 与命名规范
- 运行时与 Grain 指标
- 消息与网关指标
- 流（Streams）缓存与压力指标
- 如何本地验证
分布式追踪（Tracing）
- ActivitySource 与传播
- Silo/Client 开启方式
Prometheus 与 OpenTelemetry 集成
- Metrics 输出到 Prometheus
- Tracing 输出到 Zipkin/Jaeger
- 仪表盘与告警建议（PromQL）
实践建议
- 命名与维度规范
- 低开销采样与 Listener
- 与持久流（Persistent Streams）的结合点

概览与原则

Orleans 全面采用 .NET 的 System.Diagnostics.Metrics 与 Activity 生态输出指标与追踪，面向 OpenTelemetry 开放。
指标以域划分（runtime、messaging、streams、gateway 等），命名统一以 orleans-* 前缀，Meter 名称统一为 "Microsoft.Orleans"。
分布式追踪使用两个 ActivitySource："Microsoft.Orleans.Runtime" 与 "Microsoft.Orleans.Application"，通过 Grain 调用过滤器传播上下文。

参考文档（Orleans 7.0）：Orleans observability（官方）

指标体系（Metrics）

Meter 与命名规范

全局 Meter：Meter("Microsoft.Orleans")，所有指标均从此源产生：

public static class Instruments
{public static readonly Meter Meter = new("Microsoft.Orleans");
}

指标命名集中在 InstrumentNames，保持跨域一致性与可发现性（示例：Streams 队列缓存指标）：

public const string STREAMS_QUEUE_CACHE_MESSAGES_ADDED = "orleans-streams-queue-cache-messages-added";
public const string STREAMS_QUEUE_CACHE_MESSAGES_PURGED = "orleans-streams-queue-cache-messages-purged";
public const string STREAMS_QUEUE_CACHE_MEMORY_ALLOCATED = "orleans-streams-queue-cache-memory-allocated";
public const string STREAMS_QUEUE_CACHE_MEMORY_RELEASED = "orleans-streams-queue-cache-memory-released";
public const string STREAMS_QUEUE_CACHE_OLDEST_TO_NEWEST_DURATION = "orleans-streams-queue-cache-oldest-to-newest-duration";
public const string STREAMS_QUEUE_CACHE_OLDEST_AGE = "orleans-streams-queue-cache-oldest-age";
public const string STREAMS_QUEUE_CACHE_PRESSURE = "orleans-streams-queue-cache-pressure";
public const string STREAMS_QUEUE_CACHE_UNDER_PRESSURE = "orleans-streams-queue-cache-under-pressure";
public const string STREAMS_QUEUE_CACHE_PRESSURE_CONTRIBUTION_COUNT = "orleans-streams-queue-cache-pressure-contribution-count";

运行时与 Grain 指标

使用 UpDownCounter<int> 追踪 Grain/SystemTarget 数量（带 type 维度），并通过 MeterListener 做轻量聚合：

internal static UpDownCounter<int> GrainCounts = Instruments.Meter.CreateUpDownCounter<int>(InstrumentNames.GRAIN_COUNTS);
internal static void IncrementGrainCounts(string grainTypeName)
{GrainCounts.Add(1, new KeyValuePair<string, object>("type", grainTypeName));
}
internal static void DecrementGrainCounts(string grainTypeName)
{GrainCounts.Add(-1, new KeyValuePair<string, object>("type", grainTypeName));
}

MeterListener.InstrumentPublished = (instrument, listener) =>
{if (instrument.Name == InstrumentNames.GRAIN_COUNTS){listener.EnableMeasurementEvents(instrument);}
};
MeterListener.SetMeasurementEventCallback<int>(OnMeasurementRecorded);

运行时精选指标监听（例：网关连接数、消息尺⼨等），利于低开销仪表盘：

private static readonly string[] MetricNames =
{// orleansInstrumentNames.GATEWAY_CONNECTED_CLIENTS,InstrumentNames.MESSAGING_RECEIVED_MESSAGES_SIZE,
};static SiloRuntimeMetricsListener()
{MeterListener.InstrumentPublished = (instrument, listener) =>{if (instrument.Meter != Instruments.Meter){return;}if (MetricNames.Contains(instrument.Name)){listener.EnableMeasurementEvents(instrument);}};
}

消息与网关指标

消息层面指标（发送/接收大小、失败/丢弃、Dispatcher 收/处/转发等）采用 Counter/ObservableCounter/Histogram，命名均在 InstrumentNames 中统一管理（不赘引）。

流（Streams）缓存与压力指标

队列缓存（Queue Cache）指标覆盖：缓存大小、消息数、消息增删、内存分配/释放、最老到最新时距、最老消息年龄、实时压力、是否 UnderPressure、贡献计数；包含关键维度（如 QueueId）：

_queueCacheSizeCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_SIZE, () => new(_totalCacheSize, _dimensions), unit: "bytes");
_queueCacheMessageCountCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_LENGTH, () => new(_messageCount, _dimensions), unit: "messages");
_queueCacheMessagesAddedCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_MESSAGES_ADDED, () => new(_messagesAdded, _dimensions));
_queueCacheMessagesPurgedCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_MESSAGES_PURGED, () => new(_messagesPurged, _dimensions));
_queueCacheMemoryAllocatedCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_MEMORY_ALLOCATED, () => new(_memoryAllocated, _dimensions));
_queueCacheMemoryReleasedCounter = Instruments.Meter.CreateObservableCounter<long>(InstrumentNames.STREAMS_QUEUE_CACHE_MEMORY_RELEASED, () => new(_memoryReleased, _dimensions));
_oldestMessageReadEnqueueTimeToNowCounter = Instruments.Meter.CreateObservableGauge<long>(InstrumentNames.STREAMS_QUEUE_CACHE_OLDEST_TO_NEWEST_DURATION, GetOldestToNewestAge);
_newestMessageReadEnqueueTimeToNowCounter = Instruments.Meter.CreateObservableGauge<long>(InstrumentNames.STREAMS_QUEUE_CACHE_OLDEST_AGE, GetOldestAge);
_currentPressureCounter = Instruments.Meter.CreateObservableGauge<double>(InstrumentNames.STREAMS_QUEUE_CACHE_PRESSURE, () => GetPressureMonitorMeasurement(monitor => monitor.CurrentPressure));
_underPressureCounter = Instruments.Meter.CreateObservableGauge<int>(InstrumentNames.STREAMS_QUEUE_CACHE_UNDER_PRESSURE, () => GetPressureMonitorMeasurement(monitor => monitor.UnderPressure));
_pressureContributionCounter = Instruments.Meter.CreateObservableGauge<double>(InstrumentNames.STREAMS_QUEUE_CACHE_PRESSURE_CONTRIBUTION_COUNT, () => GetPressureMonitorMeasurement(monitor => monitor.PressureContributionCount));

压力来源来自各实现的 PressureMonitor，以实现类型作为维度进行区分与合并（PressureMonitorStatistics 内部支持）。
与持久流（Persistent Streams）处理链的结合点说明：
- PersistentStreamPullingAgent 拉取与分发批次期间会驱动缓存与压力监控逻辑（例如队列消息取入、缓存老化/清理、背压触发）。你当前聚焦位置为该文件第 477 行的拉取调用：
  - orleans-main/src/Orleans.Streaming/PersistentStreams/PersistentStreamPullingAgent.cs 行 477：IList<IBatchContainer> multiBatch = await rcvr.GetQueueMessagesAsync(maxCacheAddCount);
- 配套的缓存实现（如 SimpleQueueCache）与监控器（DefaultCacheMonitor）共同产出上述指标。

如何本地验证

使用 .NET 诊断工具快速观察 Orleans Meters：

dotnet counters monitor -n <ProcessName> --counters Microsoft.Orleans

常见输出包含目录环、网关收发速率、Dispatcher 处理计数、应用请求延迟分布等（详见官方示例输出）。

分布式追踪（Tracing）

ActivitySource 与传播

Orleans 按命名空间区分 Runtime/Application 两个 ActivitySource，并在 Grain 调用前后打上标准化 RPC 语义标签与 Orleans 自定义标签：

internal const string ApplicationGrainActivitySourceName = "Microsoft.Orleans.Application";
internal const string RuntimeActivitySourceName = "Microsoft.Orleans.Runtime";protected static readonly ActivitySource ApplicationGrainSource = new(ApplicationGrainActivitySourceName, "1.0.0");
protected static readonly ActivitySource RuntimeGrainSource = new(RuntimeActivitySourceName, "1.0.0");protected static ActivitySource GetActivitySource(IGrainCallContext context) =>context.Request.GetInterfaceType().Namespace?.StartsWith(OrleansNamespacePrefix) == true? RuntimeGrainSource: ApplicationGrainSource;

传播过程（请求前设置 rpc.system/service/method 以及 orleans target/source id，异常时设置 Status/Error）均在过滤器中完成。

Silo/Client 开启方式

Silo 侧：

public static ISiloBuilder AddActivityPropagation(this ISiloBuilder builder)
{builder.Services.TryAddSingleton(DistributedContextPropagator.Current);return builder.AddOutgoingGrainCallFilter<ActivityPropagationOutgoingGrainCallFilter>().AddIncomingGrainCallFilter<ActivityPropagationIncomingGrainCallFilter>();
}

Client 侧：

public static IClientBuilder AddActivityPropagation(this IClientBuilder builder)
{builder.Services.TryAddSingleton(DistributedContextPropagator.Current);return builder.AddOutgoingGrainCallFilter<ActivityPropagationOutgoingGrainCallFilter>();
}

也可通过配置项打开：

if (bool.TryParse(cfg["EnableDistributedTracing"], out var enableDistributedTracing) && enableDistributedTracing)
{builder.AddActivityPropagation();
}

Prometheus 与 OpenTelemetry 集成

Metrics 输出到 Prometheus

Orleans 指标源为 Meter("Microsoft.Orleans")。在主机构建时将其加入 OpenTelemetry Metrics 管道，并使用 Prometheus Exporter 暴露：

// Program.cs
builder.Services.AddOpenTelemetry().WithMetrics(metrics =>{metrics.AddPrometheusExporter().AddMeter("Microsoft.Orleans"); // 订阅 Orleans 指标源});var app = builder.Build();
app.MapPrometheusScrapingEndpoint(); // 暴露 /metrics，供 Prometheus 抓取
app.Run();

注意：官方当前 Prometheus Exporter（OpenTelemetry.Exporter.Prometheus / .AspNetCore）在发布候选阶段，生产需评估稳定性。官方文档说明

Tracing 输出到 Zipkin/Jaeger

添加 OpenTelemetry Tracing，订阅 Orleans 的 ActivitySource，并配置 Zipkin 导出（Jaeger 兼容 Zipkin 格式）：

builder.Services.AddOpenTelemetry().WithTracing(tracing =>{tracing.SetResourceBuilder(ResourceBuilder.CreateDefault().AddService("YourServiceName")).AddSource("Microsoft.Orleans.Runtime").AddSource("Microsoft.Orleans.Application").AddZipkinExporter(zipkin =>{zipkin.Endpoint = new Uri("http://localhost:9411/api/v2/spans");});});

与官方文档对齐，生产使用前同样需评估 exporter 的成熟度。官方文档说明

仪表盘与告警建议（PromQL）

指标命名前缀通常在 Prometheus 中会被下划线转换（因 exporter 实现而异），示例仅供参考：
速率与吞吐
- 消息发送速率：rate(orleans_streams_queue_cache_messages_added_total[5m])
- 网关请求吞吐：rate(orleans_gateway_sent_total[1m])
背压与积压
- 队列压力：avg(orleans_streams_queue_cache_pressure)
- UnderPressure 比例：sum(orleans_streams_queue_cache_under_pressure) / count(orleans_streams_queue_cache_under_pressure)
- 缓存长度阈值告警：orleans_streams_queue_cache_length > 10000
时延分布
- 应用请求分位：基于 orleans-app-requests-latency-*（如 sum/bucket，根据 exporter 输出具体名称做适配）

实践建议

命名与维度规范
- 固定订阅 Meter("Microsoft.Orleans")，在查询中善用维度（如 QueueId、pressureMonitorType、Grain type）用于分组与定位。
- 指标命名遵循 orleans-<domain>-<entity>-<metric>，便于跨域归类与仪表盘组织。
低开销采样与 Listener
- 尽量使用 ObservableGauge/ObservableCounter 与内部累加器，避免高频同步打点导致的锁竞争。
- 合理使用 MeterListener 聚合精选指标，降低抓取与渲染成本（参考 SiloRuntimeMetricsListener）。
与持久流（Persistent Streams）的结合点
- 拉取代理 PersistentStreamPullingAgent 在拉取与分发批次期间会驱动缓存统计及压力监控（你当前查看位置在第 477 行的拉取入口），配套 SimpleQueueCache 与 DefaultCacheMonitor 产出完整的缓存与压力指标。
- 建议重点看与缓存操作、背压决策、清理策略有关的代码路径，以建立“消息入/出/清理/背压”的指标链路闭环。

—
简短总结

Orleans 使用 System.Diagnostics.Metrics 与 Activity 原生输出指标与追踪，Meter 名为 Microsoft.Orleans，ActivitySource 为 Microsoft.Orleans.Runtime/Application。
Streams 的队列缓存与背压指标覆盖齐全，带关键维度，可直接用于容量/告警/调优。
通过 OpenTelemetry 可无缝接入 Prometheus（Metrics）与 Zipkin/Jaeger（Tracing），配置简洁、生态兼容性好。参考官方文档获取更多细节与注意事项。

查看全文

http://www.dtcms.com/a/508299.html