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.