From ddd8420aac3e4e96dfa4869ce668ea990a40f851 Mon Sep 17 00:00:00 2001 From: Ziya Suzen Date: Tue, 23 Jun 2026 12:59:59 +0100 Subject: [PATCH] core: exclude status frames from consumed metrics consumed.messages and received.bytes count only messages delivered to the application, per the OTel messaging convention. NATS status/control frames (no-responder 503s, JetStream heartbeats, flow-control, protocol notifications) are consumed internally and never reach the application, but the base receive path counted them. Skip them via the inline status code on the version line, which covers every subscription type through one chokepoint. --- src/NATS.Client.Core/NatsSubBase.cs | 22 +++++++++- .../StatusMsgTest.cs | 35 ++++++++++++++++ .../OpenTelemetryTest.cs | 42 +++++++++++++++++++ .../documentation/advanced/opentelemetry.md | 6 +++ 4 files changed, 104 insertions(+), 1 deletion(-) create mode 100644 tests/NATS.Client.CoreUnit.Tests/StatusMsgTest.cs diff --git a/src/NATS.Client.Core/NatsSubBase.cs b/src/NATS.Client.Core/NatsSubBase.cs index b8b655a13..733ed5171 100644 --- a/src/NATS.Client.Core/NatsSubBase.cs +++ b/src/NATS.Client.Core/NatsSubBase.cs @@ -374,7 +374,12 @@ public virtual async ValueTask ReceiveAsync(string subject, string? replyTo, Rea // the hand-off to the subscription channel. received.bytes follows the same // convention. Messages dropped by cancellation, channel closure, or serializer // exceptions are not counted. - if (Telemetry.ConsumedMessages.Enabled || Telemetry.ReceivedBytes.Enabled) + // + // NATS status/control frames (no-responder 503s and JetStream heartbeats, + // flow-control, and protocol notifications) are consumed internally and never + // reach the application, so they are excluded from both counters. They are + // identifiable by the inline status code on the "NATS/1.0 " header line. + if ((Telemetry.ConsumedMessages.Enabled || Telemetry.ReceivedBytes.Enabled) && !IsStatusMsg(headersBuffer)) { var tags = Telemetry.BuildMetricTags(Connection, Telemetry.Constants.OpRec); if (Telemetry.ConsumedMessages.Enabled) @@ -425,6 +430,21 @@ internal static bool IsHeader503(ReadOnlySequence? headersBuffer) => headersBuffer is { Length: >= 12 } && headersBuffer.Value.Slice(8, 4).ToSpan().SequenceEqual(NoRespondersHeaderSequence); + // A NATS status/control frame carries an inline status code on the version line + // ("NATS/1.0 "): a space then three ASCII digits at offset 8. Regular user + // headers have "\r\n" at that offset, so this never matches an application message. + internal static bool IsStatusMsg(ReadOnlySequence? headersBuffer) + { + if (headersBuffer is not { Length: >= 12 }) + return false; + + var code = headersBuffer.Value.Slice(8, 4).ToSpan(); + return code[0] == (byte)' ' + && code[1] is >= (byte)'0' and <= (byte)'9' + && code[2] is >= (byte)'0' and <= (byte)'9' + && code[3] is >= (byte)'0' and <= (byte)'9'; + } + internal void ClearException() => Interlocked.Exchange(ref _exception, null); /// diff --git a/tests/NATS.Client.CoreUnit.Tests/StatusMsgTest.cs b/tests/NATS.Client.CoreUnit.Tests/StatusMsgTest.cs new file mode 100644 index 000000000..5fd481501 --- /dev/null +++ b/tests/NATS.Client.CoreUnit.Tests/StatusMsgTest.cs @@ -0,0 +1,35 @@ +using System.Buffers; +using System.Text; + +namespace NATS.Client.Core.Tests; + +public class StatusMsgTest +{ + [Theory] + [InlineData("NATS/1.0 100 Idle Heartbeat\r\n\r\n")] // JetStream idle heartbeat + [InlineData("NATS/1.0 503\r\n\r\n")] // no responders + [InlineData("NATS/1.0 408 Request Timeout\r\n\r\n")] // pull request timeout + [InlineData("NATS/1.0 409 Message Size Exceeds MaxBytes\r\n\r\n")] // flow control + [InlineData("NATS/1.0 100 FlowControl Request\r\n\r\n")] + public void Status_frames_are_detected(string header) + { + NatsSubBase.IsStatusMsg(Seq(header)).Should().BeTrue(); + } + + [Theory] + [InlineData("NATS/1.0\r\nFoo: Bar\r\n\r\n")] // regular user headers + [InlineData("NATS/1.0\r\n\r\n")] // headers present, no user keys, no status code + public void User_headers_are_not_status_frames(string header) + { + NatsSubBase.IsStatusMsg(Seq(header)).Should().BeFalse(); + } + + [Fact] + public void Null_or_short_buffer_is_not_a_status_frame() + { + NatsSubBase.IsStatusMsg(null).Should().BeFalse(); + NatsSubBase.IsStatusMsg(Seq("NATS/1.0")).Should().BeFalse(); + } + + private static ReadOnlySequence Seq(string s) => new(Encoding.ASCII.GetBytes(s)); +} diff --git a/tests/NATS.Net.OpenTelemetry.Tests/OpenTelemetryTest.cs b/tests/NATS.Net.OpenTelemetry.Tests/OpenTelemetryTest.cs index a32f6bc11..97aa0fd84 100644 --- a/tests/NATS.Net.OpenTelemetry.Tests/OpenTelemetryTest.cs +++ b/tests/NATS.Net.OpenTelemetry.Tests/OpenTelemetryTest.cs @@ -442,6 +442,48 @@ public async Task Direct_request_reply_receive_activity_is_disposed() await reg; } + [Fact] + public async Task Consumed_counter_excludes_jetstream_control_messages() + { + using var meter = new MeterTracker(); + await using var server = await NatsServerProcess.StartAsync(); + await using var nats = new NatsConnection(new NatsOpts { Url = server.Url }); + + var js = new NatsJSContext(nats); + + var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30)); + + await js.CreateStreamAsync(new StreamConfig { Name = "ctrl-stream", Subjects = ["ctrl.>"] }, cts.Token); + await js.CreateOrUpdateConsumerAsync("ctrl-stream", new ConsumerConfig("ctrl-consumer"), cts.Token); + + for (var i = 0; i < 3; i++) + await js.PublishAsync("ctrl.subject", i, cancellationToken: cts.Token); + + var consumer = await js.GetConsumerAsync("ctrl-stream", "ctrl-consumer", cts.Token); + + // consumed.messages is connection-wide and already includes the PubAck and JS API + // replies from the setup above, so measure the delta across the fetch only. + long Consumed() => meter.LongMeasurements + .Where(m => m.Name == "messaging.client.consumed.messages") + .Sum(m => m.Value); + + var before = Consumed(); + + // MaxMsgs greater than the 3 available, with a short expiry: the server delivers the + // 3 data messages on the fetch subscription and then a status frame to end the batch. + // Only the 3 data messages were delivered to the application, so the delta must be 3. + var received = 0; + await foreach (var msg in consumer.FetchAsync( + new NatsJSFetchOpts { MaxMsgs = 10, Expires = TimeSpan.FromSeconds(2) }, + cancellationToken: cts.Token)) + { + received++; + } + + received.Should().Be(3); + (Consumed() - before).Should().Be(3); + } + private void AssertActivityData(string subject, IReadOnlyList activityList, string expectedHost, string expectedClientId) { var activities = activityList.ToArray(); diff --git a/tools/site_src/documentation/advanced/opentelemetry.md b/tools/site_src/documentation/advanced/opentelemetry.md index c062ca001..d88a2f14e 100644 --- a/tools/site_src/documentation/advanced/opentelemetry.md +++ b/tools/site_src/documentation/advanced/opentelemetry.md @@ -108,3 +108,9 @@ All instruments carry these tags: | `network.transport` | `tcp` | Transport protocol | `messaging.client.operation.duration` adds `error.type` (full exception type name) when the operation fails. + +`messaging.client.consumed.messages` and `nats.client.received.bytes` count only messages delivered to the +application, per the OTel definition ("messages delivered to the application"). NATS status and control +frames consumed internally by the client are excluded: no-responder `503` replies and JetStream heartbeats, +flow-control, and protocol notifications. The two counters stay consistent, so `received.bytes / consumed.messages` +reflects average delivered message size.