YMJake.RocketMQ.Client.OpenTelemetry 1.1.10

YMJake.RocketMQ.Client.OpenTelemetry

OpenTelemetry instrumentation for YMJake.RocketMQ.Client - Automatic distributed tracing and metrics for Apache RocketMQ 5.x messaging operations.

Features

• ✅ Automatic tracing for message publish, receive, and process operations, including LitePushConsumer and LiteSimpleConsumer
• ✅ Built-in metrics for send/receive/consume latency histograms and broker-controlled OTLP metric export
• ✅ W3C Trace Context propagation (traceparent/tracestate)
• ✅ OpenTelemetry semantic conventions for messaging
• ✅ RocketMQ-specific attributes (message type, tags, etc.)
• ✅ Batch message support
• ✅ Customizable filtering and captured message headers
• ✅ Zero code changes required in your application

Supports .NET 8.0 (LTS) and .NET 10.0

Installation

dotnet add package YMJake.RocketMQ.Client.OpenTelemetry

Quick Start

using OpenTelemetry;
using OpenTelemetry.Resources;
using OpenTelemetry.Metrics;
using OpenTelemetry.Trace;
using RocketMQ.Client.OpenTelemetry;

builder.Services.AddRocketMqMetricsInstrumentation();

var tracerProvider = Sdk.CreateTracerProviderBuilder()
    .SetResourceBuilder(ResourceBuilder.CreateDefault()
        .AddService("my-service"))
    .AddRocketMqInstrumentation()
    .AddOtlpExporter()  // or AddConsoleExporter(), AddJaegerExporter(), etc.
    .Build();

var meterProvider = Sdk.CreateMeterProviderBuilder()
    .AddRocketMqMetrics()
    .AddOtlpExporter()
    .Build();

That's it! The tracing instrumentation automatically listens to DiagnosticListener events emitted by the RocketMQ client and produces spans for publish, receive, and process operations, while the metrics instrumentation publishes RocketMQ histograms into the OpenTelemetry metrics pipeline and reacts to broker metric settings.

If you want a runnable WebAPI example, see RocketMQ.Client.OpenTelemetry.Sample.

Advanced Configuration

Filtering Messages

.AddRocketMqInstrumentation(options =>
{
    // Only trace messages from specific topics
    options.Filter = data => 
        data.Messages.Any(m => m.Topic.StartsWith("important-"));
})

Captured Headers

.AddRocketMqInstrumentation(options =>
{
    options.CapturedHeaders.Add("tenant-id");
    options.CapturedHeaders.Add("order-type");
})

Captured message properties are recorded as messaging.header.{name} attributes.

Metrics

RocketMQ metrics are exposed as histograms under the Apache.RocketMQ.Client meter.

Application Pipeline Registration

builder.Services.AddRocketMqMetricsInstrumentation();

builder.Services.AddOpenTelemetry()
    .WithMetrics(metrics =>
    {
        metrics.AddRocketMqMetrics();
        metrics.AddOtlpExporter();
    });

Broker-Controlled Exporter

AddRocketMqMetricsInstrumentation() also listens for BrokerMetricSettings diagnostics emitted by the core client. When RocketMQ broker or proxy enables metric export in telemetry Settings, the package automatically creates or resets an OTLP metric exporter using the broker-provided endpoint, timeout, and signed headers.

No extra application code is required beyond registering AddRocketMqMetricsInstrumentation().

If you also register AddRocketMqMetrics() in your own MeterProvider, the same RocketMQ histograms can be exported through both your application metrics pipeline and the broker-controlled exporter.

Available Histograms

• rocketmq_send_cost_time
• rocketmq_delivery_latency
• rocketmq_await_time
• rocketmq_process_time

Tags by Histogram

• rocketmq_send_cost_time
  • topic
  • client_id
  • invocation_status
• rocketmq_delivery_latency
  • topic
  • client_id
  • consumer_group
• rocketmq_await_time
  • topic
  • client_id
  • consumer_group
• rocketmq_process_time
  • topic
  • client_id
  • consumer_group
  • invocation_status

Span Attributes

The instrumentation follows OpenTelemetry semantic conventions and adds the following attributes:

SemConv Alignment Notes

The package is aligned with RocketMQ's messaging semantic conventions for client-side observability:

• span name: {topic} publish | {topic} receive | {topic} process
• messaging.system = "rocketmq"
• messaging.operation = "publish" | "receive" | "process"
• messaging.destination.name = topic name
• messaging.rocketmq.client_group = consumer group (receive / process)
• messaging.batch.message_count = batch size (receive)
• messaging.message.id = message ID (publish / process)
• messaging.message.body.size = message body size in bytes (publish / process)
• messaging.rocketmq.message.type = "normal" | "fifo" | "delay" | "transaction" (publish)
• messaging.rocketmq.message.tag = message tag
• messaging.rocketmq.message.group = message group
• messaging.rocketmq.message.keys = message keys
• messaging.rocketmq.message.delivery_timestamp = scheduled delivery timestamp
• messaging.header.{name} = captured message property values (publish / process)

Client-side metrics are intentionally focused on send, delivery, await, and process latency. Broker-side labels such as cluster or broker node identifiers are intentionally not included in this package.

If the core RocketMQ client later exposes stable endpoint and error metadata in diagnostics, the instrumentation can be extended with additional standard attributes such as server.address, server.port, and error.type.

Standard Messaging Attributes

• messaging.system = "rocketmq"
• messaging.operation = "publish" | "receive" | "process"
• messaging.destination.name = topic name
• messaging.message.id = message ID (publish / process)
• messaging.message.body.size = message body size in bytes (publish / process)
• messaging.batch.message_count = number of messages in batch (receive)

RocketMQ-Specific Attributes

• messaging.rocketmq.client_group = consumer group (receive / process)
• messaging.rocketmq.message.type = "normal" | "fifo" | "delay" | "transaction" (publish)
• messaging.rocketmq.message.tag = message tag
• messaging.rocketmq.message.group = message group (for FIFO)
• messaging.rocketmq.message.keys = message keys
• messaging.rocketmq.message.delivery_timestamp = scheduled delivery timestamp (Unix milliseconds)
• messaging.header.{name} = captured message property values (publish / process)

How It Works

The instrumentation uses .NET's DiagnosticListener to subscribe to events emitted by the RocketMQ client. It automatically:

1. Creates spans for publish, receive, and process operations
2. Propagates trace context via message properties (traceparent/tracestate)
3. Makes process spans children of receive spans when receive instrumentation is enabled, while linking them back to producer spans
4. Skips receive spans for successful empty polls, while still creating error receive spans for failed receive attempts
5. Sets appropriate span status based on operation results

This applies to the client-side consumer variants supported by the core package, including SimpleConsumer, PushConsumer, LitePushConsumer, and LiteSimpleConsumer.

No changes are required in your application code beyond adding the instrumentation to your TracerProvider.

License

Apache License 2.0. Based on Apache RocketMQ Clients.

Links

• Main Package: YMJake.RocketMQ.Client
• Repository: Gitee
• OpenTelemetry Java RocketMQ instrumentation: rocketmq-client-5.0
• OpenTelemetry RocketMQ SemConv: messaging/rocketmq
• OpenTelemetry Messaging Attributes Registry: registry/attributes/messaging