fix: Make OTLP metrics snapshot testing more stable (#7374)

## Summary of changes
Makes the OTLP metrics snapshots more stable by adding ordering, doing
some deduplication (that wasn't reproduced locally), and consolidating
data series by their identifying Resources/Scopes

## Reason for change
We had to mark this test as Flaky, but we really want to make sure it's
stable so we can correctly do regression tests against it.

## Implementation details
Processes the OTLP metrics data after receiving it from the mock tracer
agent. This includes:
- Grouping multiple series of ScopeMetrics into one series if their
Resources match (this is expected and asserted to be true)
- Keeping only one metric data point from a Metric with the same
Resource & InstrumentationScope & Metric Name (should fix the issue with
multiple exports of the same asynchronous counter)
- Scrubs the `telemetry.sdk.version` attribute with a dummy value so we
can successfully test against multiple package versions

## Test coverage
This is the test coverage.

## Other details

Author: zacharycmontoya

tracer/test/Datadog.Trace.ClrProfiler.IntegrationTests/OpenTelemetrySdkTests.cs

@@ -202,7 +202,6 @@ public async Task IntegrationDisabled(string packageVersion)
         [SkippableTheory]
         [Trait("Category", "EndToEnd")]
         [Trait("RunOnWindows", "True")]
-        [Flaky("The received payload is super flaky, sometimes we receive multiple payloads", maxRetries: 3)]
         [MemberData(nameof(PackageVersions.OpenTelemetry), MemberType = typeof(PackageVersions))]
         public async Task SubmitsOtlpMetrics(string packageVersion)
         {
@@ -235,22 +234,66 @@ public async Task SubmitsOtlpMetrics(string packageVersion)
 
                 metricRequests.Should().NotBeEmpty("Expected OTLP metric requests were not received.");
 
-                // TODO: We only expect one metric request in the snapshot, but we could actually have multiple payloads, some of which
-                // could contain no metrics, some of which could contain _duplicate_ metrics. Flake central.
-                // To fix it we need to deserialize the data into a "real" payload, and aggregate the metrics as appropriate,
-                // to make sure that we're actually sending the "correct" results. For example, gauge values should not change,
-                // count values should be 0 in subsequent payloads etc.
-                var snapshotPayload = metricRequests
-                    .Select(r => r.DeserializedData)
-                    .ToList();
+                // Group the scope metrics by the resource metrics and schema URL (should only be one unique combination)
+                var resourceMetricByResource = metricRequests
+                                        .SelectMany(r => r.MetricsData.ResourceMetrics)
+                                        .GroupBy(r => new Tuple<global::OpenTelemetry.Proto.Resource.V1.Resource, string>(r.Resource, r.SchemaUrl))
+                                        .Should()
+                                        .ContainSingle()
+                                        .Subject;
+
+                // Group the individual metrics by scope metric and schema URL (should only be one unique combination since we're only using one ActivitySource)
+                // This may result in multiple entries for metrics that are repeated multiple times before the test exits
+                var scopeMetricsByResource = resourceMetricByResource
+                                        .SelectMany(r => r.ScopeMetrics)
+                                        .GroupBy(r => new Tuple<global::OpenTelemetry.Proto.Common.V1.InstrumentationScope, string>(r.Scope, r.SchemaUrl))
+                                        .OrderBy(group => group.Key.Item1.Name);
+
+                var scopeMetrics = new List<object>();
+                foreach (var scopeMetricByResource in scopeMetricsByResource)
+                {
+                    var metrics = scopeMetricByResource
+                                    .SelectMany(r => r.Metrics)
+                                    .GroupBy(r => r.Name)
+                                    .OrderBy(group => group.Key)
+                                    .Select(group => group.First())
+                                    .ToList();
+
+                    scopeMetrics.Add(new
+                    {
+                        Scope = scopeMetricByResource.Key.Item1,
+                        Metrics = metrics,
+                        SchemaUrl = scopeMetricByResource.Key.Item2
+                    });
+                }
+
+                // Filter out the telemetry resource name, if any
+                foreach (var attribute in resourceMetricByResource.Key.Item1.Attributes)
+                {
+                    if (attribute.Key.Equals("telemetry.sdk.version"))
+                    {
+                        attribute.Value.StringValue = "sdk-version";
+                    }
+                }
+
+                // Although there's only one resource, let's still emit snapshot data in the expected array format
+                var resourceMetrics = new object[]
+                {
+                    new
+                    {
+                        Resource = resourceMetricByResource.Key.Item1,
+                        ScopeMetrics = scopeMetrics,
+                        SchemaUrl = resourceMetricByResource.Key.Item2,
+                    }
+                };
 
                 var settings = VerifyHelper.GetSpanVerifierSettings();
                 settings.AddRegexScrubber(_timeUnixNanoRegexMetrics, @"TimeUnixNano"": <DateTimeOffset.Now>");
 
                 var suffix = GetSuffix(packageVersion);
                 var fileName = $"{nameof(OpenTelemetrySdkTests)}.SubmitsOtlpMetrics{suffix}{snapshotName}";
 
-                await Verifier.Verify(snapshotPayload, settings)
+                await Verifier.Verify(resourceMetrics, settings)
                               .UseFileName(fileName)
                               .DisableRequireUniquePrefix();
             }

tracer/test/Datadog.Trace.TestHelpers/MockTracerAgent.cs

@@ -976,34 +976,28 @@ private void HandlePotentialOtlpData(MockHttpRequest request)
                 h => h.Value.ToList(),
                 StringComparer.OrdinalIgnoreCase);
 
-            object deserializedData = null;
+            MetricsData metricsData = null;
 
             if (request.PathAndQuery.StartsWith("/v1/metrics") && contentType.Contains("protobuf"))
             {
                 try
                 {
-                    var metricsData = MetricsData.Parser.ParseFrom(body);
-                    var resource = metricsData.ResourceMetrics;
-
-                    if (resource is not null)
-                    {
-                        deserializedData = resource;
-                    }
+                    metricsData = MetricsData.Parser.ParseFrom(body);
                 }
                 catch (Exception ex)
                 {
                     Output?.WriteLine($"[OTLP] Failed to deserialize metrics data: {ex.Message}");
                 }
             }
 
-            if (deserializedData is not null)
+            if (metricsData is not null)
             {
                 OtlpRequests.Enqueue(new MockOtlpRequest(
                                          request.PathAndQuery,
                                          headersDict,
                                          body,
                                          contentType,
-                                         deserializedData));
+                                         metricsData));
             }
         }
 
@@ -1704,13 +1698,13 @@ private async Task HandleUdsTraces()
 
         public class MockOtlpRequest
         {
-            public MockOtlpRequest(string pathAndQuery, Dictionary<string, List<string>> headers, byte[] body, string contentType, object deserializedData)
+            public MockOtlpRequest(string pathAndQuery, Dictionary<string, List<string>> headers, byte[] body, string contentType, MetricsData metricsData)
             {
                 PathAndQuery = pathAndQuery;
                 Headers = headers;
                 Body = body;
                 ContentType = contentType;
-                DeserializedData = deserializedData;
+                MetricsData = metricsData;
             }
 
             public string PathAndQuery { get; }
@@ -1724,7 +1718,7 @@ public MockOtlpRequest(string pathAndQuery, Dictionary<string, List<string>> hea
             /// <summary>
             /// Gets the deserialized protobuf data (if available)
             /// </summary>
-            public object DeserializedData { get; }
+            public MetricsData MetricsData { get; }
         }
     }
 }