YMJake.RocketMQ.Client.OpenTelemetry 1.1.4

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 send and receive operations, including LiteSimpleConsumer
• ✅ Built-in metrics for send/receive/consume latency histograms
• ✅ W3C Trace Context propagation (traceparent/tracestate)
• ✅ OpenTelemetry semantic conventions for messaging
• ✅ RocketMQ-specific attributes (namespace, message type, tags, etc.)
• ✅ Batch message support
• ✅ Customizable filtering and enrichment
• ✅ 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 send and receive operations, while the metrics instrumentation publishes RocketMQ histograms into the OpenTelemetry metrics pipeline.

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-"));
})

Custom Tag Conversion

.AddRocketMqInstrumentation(options =>
{
    // Truncate message IDs for readability
    options.TagValueConverter = (name, value) =>
        name == "messaging.message.id" ? value?.Substring(0, 8) : value;
})

Additional Attributes

.AddRocketMqInstrumentation(options =>
{
    options.AdditionalAttributes["environment"] = () => "production";
    options.AdditionalAttributes["version"] = () => "1.0.0";
})

Metrics

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

Registration

builder.Services.AddRocketMqMetricsInstrumentation();

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

Available Histograms

• rocketmq_send_cost_time
• rocketmq_delivery_latency
• rocketmq_await_time
• rocketmq_process_time

Common Tags

• topic
• client_id
• namespace
• 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:

• messaging.system = "rocketmq"
• messaging.operation.name = "send" | "receive"
• messaging.operation.type = "send" | "receive"
• messaging.destination.name = topic name
• messaging.client.id = RocketMQ client ID
• messaging.consumer.group.name = consumer group
• messaging.rocketmq.namespace = RocketMQ namespace
• messaging.rocketmq.message.type = "normal" | "fifo" | "delay"
• 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.batch.message_count = batch size

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.name = "send" | "receive"
• messaging.operation.type = "send" | "receive"
• messaging.destination.name = topic name
• messaging.message.id = message ID
• messaging.message.body.size = message body size in bytes
• messaging.batch.message_count = number of messages in batch
• messaging.client.id = RocketMQ client ID

RocketMQ-Specific Attributes

• messaging.rocketmq.consumption_model = "clustering" (consumer spans)
• messaging.consumer.group.name = consumer group
• messaging.rocketmq.namespace = RocketMQ namespace
• messaging.rocketmq.message.type = "normal" | "fifo" | "delay"
• 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)

How It Works

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

1. Creates spans for send and receive operations
2. Propagates trace context via message properties (traceparent/tracestate)
3. Links consumer spans to producer spans for distributed tracing
4. 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 RocketMQ SemConv: messaging/rocketmq
• OpenTelemetry Messaging Attributes Registry: registry/attributes/messaging