From 20627ff2e1a2577820c18fe9c5498394168650c9 Mon Sep 17 00:00:00 2001 From: Rajkumar Rangaraj Date: Fri, 15 Nov 2024 15:13:14 -0800 Subject: [PATCH 1/3] Deserialize GrpcStatusDetailsHeader to Retrieve Grpc Retry Delay --- .../Grpc/GrpcStatusDeserializer.cs | 292 ++++++++++++++++ .../GrpcStatusDeserializerTests.cs | 319 ++++++++++++++++++ 2 files changed, 611 insertions(+) create mode 100644 src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs create mode 100644 test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs new file mode 100644 index 00000000000..ea1dd14098c --- /dev/null +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs @@ -0,0 +1,292 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +using System.Text; + +namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc; + +internal class GrpcStatusDeserializer +{ +#pragma warning disable SA1310 // Field names should not contain underscore + // Wire types in protocol buffers + private const int WIRETYPE_VARINT = 0; + private const int WIRETYPE_FIXED64 = 1; + private const int WIRETYPE_LENGTH_DELIMITED = 2; + private const int WIRETYPE_FIXED32 = 5; +#pragma warning restore SA1310 // Field names should not contain underscore + + internal static TimeSpan? TryGetGrpcRetryDelay(string? grpcStatusDetailsHeader) + { + try + { + var retryInfo = ExtractRetryInfo(grpcStatusDetailsHeader); + if (retryInfo?.RetryDelay != null) + { + return TimeSpan.FromSeconds(retryInfo.Value.RetryDelay.Value.Seconds) + + TimeSpan.FromTicks(retryInfo.Value.RetryDelay.Value.Nanos / 100); // Convert nanos to ticks + } + } + catch + { + // TODO: Log exception to event source. + return null; + } + + return null; + } + + // Marked as internal for test. + internal static Status? DeserializeStatus(string? grpcStatusDetailsBin) + { + if (string.IsNullOrWhiteSpace(grpcStatusDetailsBin)) + { + return null; + } + + var status = new Status(); + byte[] data = Convert.FromBase64String(grpcStatusDetailsBin); + using (var stream = new MemoryStream(data)) + { + while (stream.Position < stream.Length) + { + var tag = DecodeTag(stream); + var fieldNumber = tag >> 3; + var wireType = tag & 0x7; + + switch (fieldNumber) + { + case 1: // code + status.Code = DecodeInt32(stream); + break; + case 2: // message + status.Message = DecodeString(stream); + break; + case 3: // details + status.Details.Add(DecodeAny(stream)); + break; + default: + SkipField(stream, wireType); + break; + } + } + } + + return status; + } + + // Marked as internal for test. + internal static RetryInfo? ExtractRetryInfo(string? grpcStatusDetailsBin) + { + var status = DeserializeStatus(grpcStatusDetailsBin); + if (status == null) + { + return null; + } + + foreach (var detail in status.Value.Details) + { + if (detail.TypeUrl != null && detail.TypeUrl.EndsWith("/google.rpc.RetryInfo")) + { + return DeserializeRetryInfo(detail.Value!); + } + } + + return null; + } + + private static RetryInfo DeserializeRetryInfo(byte[] data) + { + RetryInfo retryInfo = default; + using (var stream = new MemoryStream(data)) + { + while (stream.Position < stream.Length) + { + var tag = DecodeTag(stream); + var fieldNumber = tag >> 3; + var wireType = tag & 0x7; + + switch (fieldNumber) + { + case 1: // retry_delay + retryInfo.RetryDelay = DecodeDuration(stream); + break; + default: + SkipField(stream, wireType); + break; + } + } + } + + return retryInfo; + } + + private static Duration DecodeDuration(Stream stream) + { + var length = DecodeVarint(stream); + var endPosition = stream.Position + length; + Duration duration = default; + + while (stream.Position < endPosition) + { + var tag = DecodeTag(stream); + var fieldNumber = tag >> 3; + var wireType = tag & 0x7; + + switch (fieldNumber) + { + case 1: // seconds + duration.Seconds = DecodeInt64(stream); + break; + case 2: // nanos + duration.Nanos = DecodeInt32(stream); + break; + default: + SkipField(stream, wireType); + break; + } + } + + return duration; + } + + private static Any DecodeAny(Stream stream) + { + var length = DecodeVarint(stream); + var endPosition = stream.Position + length; + Any any = default; + + while (stream.Position < endPosition) + { + var tag = DecodeTag(stream); + var fieldNumber = tag >> 3; + var wireType = tag & 0x7; + + switch (fieldNumber) + { + case 1: // type_url + any.TypeUrl = DecodeString(stream); + break; + case 2: // value + any.Value = DecodeBytes(stream); + break; + default: + SkipField(stream, wireType); + break; + } + } + + return any; + } + + private static uint DecodeTag(Stream stream) + { + return (uint)DecodeVarint(stream); + } + + private static long DecodeVarint(Stream stream) + { + long result = 0; + int shift = 0; + + while (true) + { + int b = stream.ReadByte(); + if (b == -1) + { + throw new EndOfStreamException(); + } + + result |= (long)(b & 0x7F) << shift; + if ((b & 0x80) == 0) + { + return result; + } + + shift += 7; + if (shift >= 64) + { + throw new InvalidDataException("Invalid varint"); + } + } + } + + private static int DecodeInt32(Stream stream) => (int)DecodeVarint(stream); + + private static long DecodeInt64(Stream stream) => DecodeVarint(stream); + + private static string DecodeString(Stream stream) + { + var bytes = DecodeBytes(stream); + return Encoding.UTF8.GetString(bytes); + } + + private static byte[] DecodeBytes(Stream stream) + { + var length = (int)DecodeVarint(stream); + var buffer = new byte[length]; + int read = stream.Read(buffer, 0, length); + if (read != length) + { + throw new EndOfStreamException(); + } + + return buffer; + } + + private static void SkipField(Stream stream, uint wireType) + { + switch (wireType) + { + case WIRETYPE_VARINT: + DecodeVarint(stream); + break; + case WIRETYPE_FIXED64: + stream.Position += 8; + break; + case WIRETYPE_LENGTH_DELIMITED: + var length = DecodeVarint(stream); + stream.Position += length; + break; + case WIRETYPE_FIXED32: + stream.Position += 4; + break; + default: + throw new InvalidDataException($"Unknown wire type: {wireType}"); + } + } + + internal struct Duration + { + public long Seconds { get; set; } + + public int Nanos { get; set; } + } + + internal struct RetryInfo + { + public Duration? RetryDelay { get; set; } + } + + internal struct Any + { + public string? TypeUrl { get; set; } + + public byte[]? Value { get; set; } + } + + internal struct Status + { + public Status() + { + this.Code = 0; + this.Message = null; + this.Details = []; + } + + public int Code { get; set; } + + public string? Message { get; set; } + + public List Details { get; set; } + } +} diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs new file mode 100644 index 00000000000..6df2b7c8d54 --- /dev/null +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs @@ -0,0 +1,319 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +using Google.Protobuf; +using Google.Protobuf.WellKnownTypes; +using OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc; +using Xunit; + +namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests.Implementation.ExportClient; + +public class GrpcStatusDeserializerTests +{ + [Fact] + public void DeserializeStatus_ValidBase64Input_ReturnsExpectedStatus() + { + var status = new Google.Rpc.Status + { + Code = 5, + Message = "Test error", + Details = + { + Any.Pack(new StringValue { Value = "Example detail" }), + }, + }; + + // Serialize the Status message and encode to base64 + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Use the GrpcStatusDeserializer to deserialize from the base64 input + var deserializedStatus = GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin); + + // Assertions to validate the deserialized Status object + Assert.NotNull(deserializedStatus); + Assert.Equal(status.Code, deserializedStatus.Value.Code); + Assert.Equal(status.Message, deserializedStatus.Value.Message); + Assert.Single(deserializedStatus.Value.Details); + Assert.Equal("type.googleapis.com/google.protobuf.StringValue", deserializedStatus.Value.Details[0].TypeUrl); + var stringValue = StringValue.Parser.ParseFrom(deserializedStatus.Value.Details[0].Value); + Assert.Equal("Example detail", stringValue.Value); + } + + [Fact] + public void DeserializeStatus_WithRetryInfo_ReturnsExpectedStatus() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 4, + Message = "Retry later", + Details = + { + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration { Seconds = 5 }, + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var retryInfo = GrpcStatusDeserializer.ExtractRetryInfo(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(retryInfo); + Assert.Equal(5, retryInfo.Value.RetryDelay?.Seconds); + } + + [Fact] + public void DeserializeStatus_EmptyStatus_ReturnsEmptyStatus() + { + // Arrange + var status = new Google.Rpc.Status(); + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var deserializedStatus = GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin); + + // Assert + Assert.Null(deserializedStatus); + } + + [Fact] + public void DeserializeStatus_MultipleDetails_ReturnsAllDetails() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 7, + Message = "Multiple details", + Details = + { + Any.Pack(new StringValue { Value = "First detail" }), + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration { Seconds = 10 }, + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var deserializedStatus = GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin); + var retryInfo = GrpcStatusDeserializer.ExtractRetryInfo(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(deserializedStatus); + Assert.Equal(status.Code, deserializedStatus.Value.Code); + Assert.Equal(status.Message, deserializedStatus.Value.Message); + Assert.Equal(2, deserializedStatus.Value.Details.Count); + + // Verify first detail (StringValue) + Assert.Equal("type.googleapis.com/google.protobuf.StringValue", deserializedStatus.Value.Details[0].TypeUrl); + var stringValue = StringValue.Parser.ParseFrom(deserializedStatus.Value.Details[0].Value); + Assert.Equal("First detail", stringValue.Value); + + // Verify second detail (RetryInfo) + Assert.Equal("type.googleapis.com/google.rpc.RetryInfo", deserializedStatus.Value.Details[1].TypeUrl); + Assert.NotNull(retryInfo); + Assert.Equal(10, retryInfo.Value.RetryDelay?.Seconds); + } + + [Fact] + public void DeserializeStatus_ComplexRetryInfo_ReturnsExpectedValues() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 4, + Message = "Complex retry scenario", + Details = + { + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration + { + Seconds = 5, + Nanos = 500000000, // 0.5 seconds + }, + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + byte[] data = Convert.FromBase64String(grpcStatusDetailsBin); + var retryInfo = GrpcStatusDeserializer.ExtractRetryInfo(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(retryInfo); + Assert.Equal(5, retryInfo.Value.RetryDelay?.Seconds); + Assert.Equal(500000000, retryInfo.Value.RetryDelay?.Nanos); + } + + [Fact] + public void ExtractRetryInfo_WithNoRetryInfoTypeUrl_ReturnsNull() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 3, + Message = "No retry info", + Details = { Any.Pack(new Google.Rpc.Status { Code = 5 }) }, // A different type packed + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var retryInfo = GrpcStatusDeserializer.ExtractRetryInfo(grpcStatusDetailsBin); + + // Assert + Assert.Null(retryInfo); + } + + [Fact] + public void DeserializeStatus_WithBoundaryCode_ReturnsExpectedStatus() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = int.MaxValue, + Message = "Boundary code test", + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var deserializedStatus = GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(deserializedStatus); + Assert.Equal(int.MaxValue, deserializedStatus.Value.Code); + Assert.Equal("Boundary code test", deserializedStatus.Value.Message); + } + + [Fact] + public void TryGetGrpcRetryDelay_NullOrEmptyInput_ReturnsNull() + { + Assert.Null(GrpcStatusDeserializer.TryGetGrpcRetryDelay(null)); + Assert.Null(GrpcStatusDeserializer.TryGetGrpcRetryDelay(string.Empty)); + Assert.Null(GrpcStatusDeserializer.TryGetGrpcRetryDelay(" ")); + } + + [Fact] + public void TryGetGrpcRetryDelay_InvalidBase64Input_ReturnsNull() + { + Assert.Null(GrpcStatusDeserializer.TryGetGrpcRetryDelay("invalid-base64")); + } + + [Fact] + public void TryGetGrpcRetryDelay_NoRetryInfo_ReturnsNull() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 3, + Message = "No retry info", + Details = { Any.Pack(new Google.Rpc.Status { Code = 5 }) }, // Non-RetryInfo type + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var result = GrpcStatusDeserializer.TryGetGrpcRetryDelay(grpcStatusDetailsBin); + + // Assert + Assert.Null(result); + } + + [Fact] + public void TryGetGrpcRetryDelay_BoundaryValuesForDuration_ReturnsNull() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 4, + Message = "Boundary test", + Details = + { + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration + { + Seconds = long.MaxValue, + Nanos = int.MaxValue, + }, + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var result = GrpcStatusDeserializer.TryGetGrpcRetryDelay(grpcStatusDetailsBin); + + // Assert + Assert.Null(result); + } + + [Fact] + public void TryGetGrpcRetryDelay_MultipleRetryInfos_UsesFirstRetryInfo() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 4, + Message = "Multiple RetryInfos", + Details = + { + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration { Seconds = 5 }, + }), + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration { Seconds = 10 }, + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var result = GrpcStatusDeserializer.TryGetGrpcRetryDelay(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(result); + Assert.Equal(TimeSpan.FromSeconds(5), result); + } + + [Fact] + public void TryGetGrpcRetryDelay_OnlyNanos_ReturnsExpected() + { + // Arrange + var status = new Google.Rpc.Status + { + Code = 4, + Message = "Only nanos", + Details = + { + Any.Pack(new Google.Rpc.RetryInfo + { + RetryDelay = new Duration { Nanos = 500000000 }, // 0.5 seconds + }), + }, + }; + + string grpcStatusDetailsBin = Convert.ToBase64String(status.ToByteArray()); + + // Act + var result = GrpcStatusDeserializer.TryGetGrpcRetryDelay(grpcStatusDetailsBin); + + // Assert + Assert.NotNull(result); + Assert.Equal(TimeSpan.FromSeconds(0.5), result); + } +} From 02610957ab70a89ab7e7823f26153ed964a16d30 Mon Sep 17 00:00:00 2001 From: Rajkumar Rangaraj Date: Fri, 15 Nov 2024 15:19:28 -0800 Subject: [PATCH 2/3] mark as static --- .../Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs index ea1dd14098c..fae5f364e7a 100644 --- a/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs +++ b/src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExportClient/Grpc/GrpcStatusDeserializer.cs @@ -5,7 +5,7 @@ namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.ExportClient.Grpc; -internal class GrpcStatusDeserializer +internal static class GrpcStatusDeserializer { #pragma warning disable SA1310 // Field names should not contain underscore // Wire types in protocol buffers From 54fdd4a2e8b8ef2f4f5ebc663db66a1363dbae77 Mon Sep 17 00:00:00 2001 From: Rajkumar Rangaraj Date: Fri, 15 Nov 2024 15:38:05 -0800 Subject: [PATCH 3/3] EndOfStreamException --- .../GrpcStatusDeserializerTests.cs | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs index 6df2b7c8d54..36e5eb521ae 100644 --- a/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs +++ b/test/OpenTelemetry.Exporter.OpenTelemetryProtocol.Tests/Implementation/ExportClient/GrpcStatusDeserializerTests.cs @@ -316,4 +316,24 @@ public void TryGetGrpcRetryDelay_OnlyNanos_ReturnsExpected() Assert.NotNull(result); Assert.Equal(TimeSpan.FromSeconds(0.5), result); } + + [Fact] + public void DeserializeStatus_TruncatedStream_ThrowsEndOfStreamException() + { + // Arrange: Create valid Base64 data and truncate it + var status = new Google.Rpc.Status + { + Code = 3, + Message = "Truncated stream test", + }; + + byte[] fullData = status.ToByteArray(); + byte[] truncatedData = fullData.Take(fullData.Length / 2).ToArray(); // Truncate the data + + string grpcStatusDetailsBin = Convert.ToBase64String(truncatedData); + + // Act & Assert: Attempt to deserialize and expect an EndOfStreamException + Assert.Throws(() => + GrpcStatusDeserializer.DeserializeStatus(grpcStatusDetailsBin)); + } }