Open Telemetry : Adds implementation for network level Metrics

Directory.Build.props

@@ -3,7 +3,7 @@
 		<ClientOfficialVersion>3.46.0</ClientOfficialVersion>
 		<ClientPreviewVersion>3.47.0</ClientPreviewVersion>
 		<ClientPreviewSuffixVersion>preview.0</ClientPreviewSuffixVersion>
-		<DirectVersion>3.37.1</DirectVersion>
+		<DirectVersion>3.37.2</DirectVersion>
 		<FaultInjectionVersion>1.0.0</FaultInjectionVersion>
 		<FaultInjectionSuffixVersion>beta.0</FaultInjectionSuffixVersion>
 		<EncryptionOfficialVersion>2.0.4</EncryptionOfficialVersion>

Microsoft.Azure.Cosmos/src/DocumentClient.cs

@@ -955,7 +955,8 @@ internal virtual void Initialize(Uri serviceEndpoint,
             if (this.cosmosClientTelemetryOptions.IsClientMetricsEnabled)
             {
                 CosmosDbOperationMeter.Initialize();
-
+                CosmosDbNetworkMeter.Initialize();
+
                 CosmosDbOperationMeter.AddInstanceCount(this.ServiceEndpoint);
             }
 

Microsoft.Azure.Cosmos/src/Resource/ClientContextCore.cs

@@ -13,7 +13,6 @@ namespace Microsoft.Azure.Cosmos
     using System.Text;
     using System.Threading;
     using System.Threading.Tasks;
-    using global::Azure;
     using Microsoft.Azure.Cosmos.Handlers;
     using Microsoft.Azure.Cosmos.Resource.CosmosExceptions;
     using Microsoft.Azure.Cosmos.Routing;
@@ -498,10 +497,13 @@ private async Task<TResult> RunWithDiagnosticsHelperAsync<TResult>(
             RequestOptions requestOptions,
             ResourceType? resourceType = null)
         {
+            bool isOtelCompatibleOperation = openTelemetry != null && this.ShouldRecordTelemetry();
+            Uri gatewayEndpoint = this.client.Endpoint;
+
             Func<string> getOperationName = () =>
             {
                 // If opentelemetry is not enabled then return null operation name, so that no activity is created.
-                if (openTelemetry == null)
+                if (!isOtelCompatibleOperation)
                 {
                     return null;
                 }
@@ -513,81 +515,117 @@ private async Task<TResult> RunWithDiagnosticsHelperAsync<TResult>(
                 return openTelemetry.Item1;
             };
 
-            using (OpenTelemetryCoreRecorder recorder =
+            using (OpenTelemetryCoreRecorder recorder = isOtelCompatibleOperation ? 
                                 OpenTelemetryRecorderFactory.CreateRecorder(
                                     getOperationName: getOperationName,
                                     containerName: containerName,
                                     databaseName: databaseName,
                                     operationType: operationType,
                                     requestOptions: requestOptions,
                                     trace: trace,
-                                    clientContext: this.isDisposed ? null : this))
+                                    clientContext: this.isDisposed ? null : this) : default)
             using (new ActivityScope(Guid.NewGuid()))
             {
                 try
                 {
                     TResult result = await task(trace).ConfigureAwait(false);
                     // Checks if OpenTelemetry is configured for this operation and either Trace or Metrics are enabled by customer
-                    if (openTelemetry != null 
-                        && (!this.ClientOptions.CosmosClientTelemetryOptions.DisableDistributedTracing || this.ClientOptions.CosmosClientTelemetryOptions.IsClientMetricsEnabled))
+                    if (isOtelCompatibleOperation)
                     {
                         // Extracts and records telemetry data from the result of the operation.
-                        OpenTelemetryAttributes response = openTelemetry?.Item2(result);
-
-                        // Records the telemetry attributes for Distributed Tracing (if enabled)
-                        recorder.Record(response);
-
-                        // Records metrics such as request units, latency, and item count for the operation.
-                        CosmosDbOperationMeter.RecordTelemetry(getOperationName: getOperationName,
-                                                             accountName: this.client.Endpoint,
-                                                             containerName: containerName,
-                                                             databaseName: databaseName,
-                                                             attributes: response);
+                        OpenTelemetryAttributes otelAttributes = openTelemetry?.Item2(result);
+
+                        // Records the telemetry attributes for Distributed Tracing (if enabled) and Metrics
+                        recorder.Record(otelAttributes);
+                        RecordMetrics(getOperationName,
+                            this.client.Endpoint,
+                            containerName,
+                            databaseName,
+                            attributes: otelAttributes);
                     }
-                    return result;
-                }
-                catch (OperationCanceledException oe) when (!(oe is CosmosOperationCanceledException))
-                {
-                    CosmosOperationCanceledException operationCancelledException = new CosmosOperationCanceledException(oe, trace);
-                    recorder.MarkFailed(operationCancelledException);
-
-                    throw operationCancelledException;
-                }
-                catch (ObjectDisposedException objectDisposed) when (!(objectDisposed is CosmosObjectDisposedException))
-                {
-                    CosmosObjectDisposedException objectDisposedException = new CosmosObjectDisposedException(
-                        objectDisposed,
-                        this.client,
-                        trace);
-                    recorder.MarkFailed(objectDisposedException);
 
-                    throw objectDisposedException;
+                    return result;
                 }
-                catch (NullReferenceException nullRefException) when (!(nullRefException is CosmosNullReferenceException))
+                catch (Exception ex) when (TryTransformException(ex, trace, this.client, out Exception cosmosException))
                 {
-                    CosmosNullReferenceException nullException = new CosmosNullReferenceException(
-                        nullRefException,
-                        trace);
-                    recorder.MarkFailed(nullException);
+                    if (isOtelCompatibleOperation)
+                    {
+                        recorder.MarkFailed(cosmosException);
+                        RecordMetrics(getOperationName,
+                            gatewayEndpoint,
+                            containerName,
+                            databaseName,
+                            cosmosException: cosmosException);
+                    }
 
-                    throw nullException;
+                    throw cosmosException; // Rethrow after recording telemetry
                 }
                 catch (Exception ex)
                 {
-                    recorder.MarkFailed(ex);
-                    if (openTelemetry != null && ex is CosmosException cosmosException)
-                    {
-                        // Records telemetry data related to the exception.
-                        CosmosDbOperationMeter.RecordTelemetry(getOperationName: getOperationName,
-                                                             accountName: this.client.Endpoint,
-                                                             containerName: containerName,
-                                                             databaseName: databaseName,
-                                                             ex: cosmosException);
-                    }
-
-                    throw;
+                    // Fallback handling for exceptions not covered by the 'when' filter
+                    recorder.MarkFailed(ex); // Record the exception using the telemetry recorder
+
+                    // Optionally rethrow or handle the exception gracefully
+                    throw; // Re-throwing to ensure the caller is aware of the unhandled exception
                 }
+
+            }
+        }
+
+        // Checks if telemetry is enabled
+        private bool ShouldRecordTelemetry()
+        {
+            CosmosClientTelemetryOptions telemetryOptions = this.clientOptions.CosmosClientTelemetryOptions;
+            return !telemetryOptions.DisableDistributedTracing || telemetryOptions.IsClientMetricsEnabled;
+        }
+
+        // Handles exceptions and records telemetry
+        private static bool TryTransformException(
+            Exception ex,
+            ITrace trace,
+            CosmosClient cosmosClient,
+            out Exception cosmosException)
+        {
+            cosmosException = ex switch
+            {
+                OperationCanceledException oe when oe is not CosmosOperationCanceledException =>
+                    new CosmosOperationCanceledException(oe, trace),
+                ObjectDisposedException od when od is not CosmosObjectDisposedException =>
+                    new CosmosObjectDisposedException(od, cosmosClient, trace),
+                NullReferenceException nr when nr is not CosmosNullReferenceException =>
+                    new CosmosNullReferenceException(nr, trace),
+                Exception ce when ce is CosmosException => ex,
+                _ => null
+            };
+
+            if (cosmosException is null)
+            {
+                return false;
             }
+            return true;
+        }
+
+        private static void RecordMetrics(Func<string> getOperationName,
+            Uri accountName,
+            string containerName,
+            string databaseName,
+            OpenTelemetryAttributes attributes = null,
+            Exception cosmosException = null)
+        {
+            // Records telemetry data
+            CosmosDbOperationMeter.RecordTelemetry(getOperationName: getOperationName,
+                                                 accountName: accountName,
+                                                 containerName: containerName,
+                                                 databaseName: databaseName,
+                                                 attributes: attributes,
+                                                 ex: cosmosException);
+
+            CosmosDbNetworkMeter.RecordTelemetry(getOperationName: getOperationName,
+                                                 accountName: accountName,
+                                                 containerName: containerName,
+                                                 databaseName: databaseName,
+                                                 attributes: attributes,
+                                                 ex: cosmosException);
         }
 
         private async Task<ResponseMessage> ProcessResourceOperationAsBulkStreamAsync(

Microsoft.Azure.Cosmos/src/Telemetry/Models/NetworkMetricData.cs

@@ -0,0 +1,47 @@
+//------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+//------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.Telemetry.Models
+{
+    internal class NetworkMetricData
+    {
+        // Constructor
+        public NetworkMetricData(
+            double latency,
+            long? requestBodySize,
+            long? responseBodySize,
+            double backendLatency,
+            double? channelAcquisitionLatency,
+            double? transitTimeLatency,
+            double? receivedLatency)
+        {
+            this.Latency = latency;
+            this.RequestBodySize = requestBodySize;
+            this.ResponseBodySize = responseBodySize;
+            this.BackendLatency = backendLatency;
+            this.ChannelAcquisitionLatency = channelAcquisitionLatency;
+            this.TransitTimeLatency = transitTimeLatency;
+            this.ReceivedLatency = receivedLatency;
+        }
+
+        // Constructor
+        public NetworkMetricData(
+            double latency,
+            long? requestBodySize,
+            long? responseBodySize)
+        {
+            this.Latency = latency;
+            this.RequestBodySize = requestBodySize;
+            this.ResponseBodySize = responseBodySize;
+        }
+
+        public double Latency { get; }
+        public long? RequestBodySize { get; }
+        public long? ResponseBodySize { get; }
+        public double BackendLatency { get; }
+        public double? ChannelAcquisitionLatency { get; }
+        public double? TransitTimeLatency { get; }
+        public double? ReceivedLatency { get; }
+    }
+}

Microsoft.Azure.Cosmos/src/Telemetry/Models/OperationMetricData.cs

@@ -0,0 +1,19 @@
+//------------------------------------------------------------
+// Copyright (c) Microsoft Corporation.  All rights reserved.
+//------------------------------------------------------------
+
+namespace Microsoft.Azure.Cosmos.Telemetry.Models
+{
+    internal class OperationMetricData
+    {
+        public OperationMetricData(string itemCount, double? requestCharge)
+        {
+            this.ItemCount = itemCount;
+            this.RequestCharge = requestCharge;
+        }
+
+        public string ItemCount { get; }
+
+        public double? RequestCharge { get; }
+    }
+}

Microsoft.Azure.Cosmos/src/Telemetry/OpenTelemetry/AppInsightClassicAttributeKeys.cs

@@ -7,6 +7,7 @@ namespace Microsoft.Azure.Cosmos.Telemetry
     using System;
     using System.Collections.Generic;
     using global::Azure.Core;
+    using Microsoft.Azure.Cosmos.Tracing.TraceData;
 
     internal sealed class AppInsightClassicAttributeKeys : IActivityAttributePopulator
     {
@@ -142,7 +143,10 @@ public void PopulateAttributes(DiagnosticScope scope, Exception exception)
             }
         }
 
-        public void PopulateAttributes(DiagnosticScope scope, QueryTextMode? queryTextMode, string operationType, OpenTelemetryAttributes response)
+        public void PopulateAttributes(DiagnosticScope scope, 
+            QueryTextMode? queryTextMode, 
+            string operationType, 
+            OpenTelemetryAttributes response)
         {
             scope.AddAttribute(AppInsightClassicAttributeKeys.OperationType, operationType);
             if (response != null)
@@ -162,17 +166,50 @@ public void PopulateAttributes(DiagnosticScope scope, QueryTextMode? queryTextMo
             }
         }
 
-        public KeyValuePair<string, object>[] PopulateOperationMeterDimensions(string operationName, string containerName, string databaseName, Uri accountName, OpenTelemetryAttributes attributes, CosmosException ex)
+        public KeyValuePair<string, object>[] PopulateNetworkMeterDimensions(string operationName, 
+            Uri accountName, 
+            string containerName, 
+            string databaseName, 
+            OpenTelemetryAttributes attributes, 
+            Exception ex, 
+            ClientSideRequestStatisticsTraceDatum.StoreResponseStatistics tcpStats = null, 
+            ClientSideRequestStatisticsTraceDatum.HttpResponseStatistics? httpStats = null)
         {
             return new KeyValuePair<string, object>[]
             {
                 new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ContainerName, containerName),
                 new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.DbName, databaseName),
-                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ServerAddress, accountName.Host),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ServerAddress, accountName?.Host),
                 new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.DbOperation, operationName),
-                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.StatusCode, (int)(attributes?.StatusCode ?? ex?.StatusCode)),
-                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.SubStatusCode, attributes?.SubStatusCode ?? ex?.SubStatusCode),
-                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.Region, string.Join(",", attributes.Diagnostics.GetContactedRegions()))
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.StatusCode, CosmosDbMeterUtil.GetStatusCode(attributes, ex)),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.SubStatusCode, CosmosDbMeterUtil.GetSubStatusCode(attributes, ex))
+            };
+        }
+
+        public KeyValuePair<string, object>[] PopulateOperationMeterDimensions(string operationName, 
+            string containerName, 
+            string databaseName, 
+            Uri accountName, 
+            OpenTelemetryAttributes attributes, 
+            Exception ex)
+        {
+            return new KeyValuePair<string, object>[]
+            {
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ContainerName, containerName),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.DbName, databaseName),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ServerAddress, accountName?.Host),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.DbOperation, operationName),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.StatusCode, CosmosDbMeterUtil.GetStatusCode(attributes, ex)),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.SubStatusCode, CosmosDbMeterUtil.GetSubStatusCode(attributes, ex)),
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.Region,  CosmosDbMeterUtil.GetRegions(attributes?.Diagnostics))
+            };
+        }
+
+        public KeyValuePair<string, object>[] PopulateInstanceCountDimensions(Uri accountEndpoint)
+        {
+            return new[]
+            {
+                new KeyValuePair<string, object>(AppInsightClassicAttributeKeys.ServerAddress, accountEndpoint.Host)
             };
         }
     }