[Internal] Client Telemetry: Adds sampling logic for network level telemetry (#3750)

* sampling logic add

* throttlinfix

* fix tests

* fix tests

* remove dispose from sampler

* add 412

* draft push

* refactor code for sampling

* clean up

* start testing

* adding tests

* wip

* test fix

* fix test

* fix tests

* deleted extra file

* remove one toList

* add custome logic for sampling

* code refactor

* replace key

* remove commented code from test

* added more comments

* simpler sampler logic

* refactor code

Author: sourabh1007

Microsoft.Azure.Cosmos/src/Telemetry/ClientTelemetry.cs

@@ -36,7 +36,8 @@ internal class ClientTelemetry : IDisposable
         private readonly ClientTelemetryProperties clientTelemetryInfo;
         private readonly ClientTelemetryProcessor processor;
         private readonly DiagnosticsHandlerHelper diagnosticsHelper;
-
+        private readonly NetworkDataRecorder networkDataRecorder;
+        
         private readonly CancellationTokenSource cancellationTokenSource;
 
         private readonly GlobalEndpointManager globalEndpointManager;
@@ -49,11 +50,8 @@ internal class ClientTelemetry : IDisposable
         private ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram> cacheRefreshInfoMap 
             = new ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram>();
 
-        private ConcurrentDictionary<RequestInfo, LongConcurrentHistogram> requestInfoMap
-            = new ConcurrentDictionary<RequestInfo, LongConcurrentHistogram>();
-
         private int numberOfFailures = 0;
-
+        
         /// <summary>
         /// Only for Mocking in tests
         /// </summary>
@@ -112,6 +110,8 @@ internal ClientTelemetry(
             GlobalEndpointManager globalEndpointManager)
         {
             this.diagnosticsHelper = diagnosticsHelper ?? throw new ArgumentNullException(nameof(diagnosticsHelper));
+            this.globalEndpointManager = globalEndpointManager;
+            
             this.processor = new ClientTelemetryProcessor(httpClient, authorizationTokenProvider);
 
             this.clientTelemetryInfo = new ClientTelemetryProperties(
@@ -122,8 +122,9 @@ internal ClientTelemetry(
                 preferredRegions: preferredRegions,
                 aggregationIntervalInSec: (int)observingWindow.TotalSeconds);
 
+            this.networkDataRecorder = new NetworkDataRecorder();
+            
             this.cancellationTokenSource = new CancellationTokenSource();
-            this.globalEndpointManager = globalEndpointManager;
         }
 
         /// <summary>
@@ -179,18 +180,15 @@ private async Task EnrichAndSendAsync()
 
                     ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram> cacheRefreshInfoSnapshot
                         = Interlocked.Exchange(ref this.cacheRefreshInfoMap, new ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram>());
-
-                    ConcurrentDictionary<RequestInfo, LongConcurrentHistogram> requestInfoSnapshot
-                      = Interlocked.Exchange(ref this.requestInfoMap, new ConcurrentDictionary<RequestInfo, LongConcurrentHistogram>());
-
+                    
                     try
                     {
                         await this.processor
-                        .ProcessAndSendAsync(
+                            .ProcessAndSendAsync(
                                 clientTelemetryInfo: this.clientTelemetryInfo,
                                 operationInfoSnapshot: operationInfoSnapshot,
                                 cacheRefreshInfoSnapshot: cacheRefreshInfoSnapshot,
-                                requestInfoSnapshot: requestInfoSnapshot,
+                                requestInfoSnapshot: this.networkDataRecorder.GetRequests(),
                                 cancellationToken: this.cancellationTokenSource.Token);
 
                         this.numberOfFailures = 0;
@@ -296,7 +294,7 @@ internal void CollectOperationInfo(CosmosDiagnostics cosmosDiagnostics,
 
             // Record Network/Replica Information
             SummaryDiagnostics summaryDiagnostics = new SummaryDiagnostics(trace);
-            this.RecordRntbdResponses(containerId, databaseId, summaryDiagnostics.StoreResponseStatistics.Value);
+            this.networkDataRecorder.Record(summaryDiagnostics.StoreResponseStatistics.Value, databaseId, containerId);
 
             string regionsContacted = ClientTelemetryHelper.GetContactedRegions(cosmosDiagnostics.GetContactedRegions());
 
@@ -338,37 +336,6 @@ internal void CollectOperationInfo(CosmosDiagnostics cosmosDiagnostics,
             }
         }
 
-        /// <summary>
-        /// Records RNTBD calls statistics
-        /// </summary>
-        /// <param name="containerId"></param>
-        /// <param name="databaseId"></param>
-        /// <param name="storeResponseStatistics"></param>
-        private void RecordRntbdResponses(string containerId, string databaseId, List<StoreResponseStatistics> storeResponseStatistics)
-        {
-            foreach (StoreResponseStatistics storetatistics in storeResponseStatistics)
-            {
-                if (ClientTelemetryOptions.IsEligible((int)storetatistics.StoreResult.StatusCode, (int)storetatistics.StoreResult.SubStatusCode, storetatistics.RequestLatency))
-                {
-                    RequestInfo requestInfo = new RequestInfo()
-                    {
-                        DatabaseName = databaseId,
-                        ContainerName = containerId,
-                        Uri = storetatistics.StoreResult.StorePhysicalAddress.ToString(),
-                        StatusCode = (int)storetatistics.StoreResult.StatusCode,
-                        SubStatusCode = (int)storetatistics.StoreResult.SubStatusCode,
-                        Resource = storetatistics.RequestResourceType.ToString(),
-                        Operation = storetatistics.RequestOperationType.ToString(),
-                    };
-
-                    LongConcurrentHistogram latencyHist = this.requestInfoMap.GetOrAdd(requestInfo, x => new LongConcurrentHistogram(ClientTelemetryOptions.RequestLatencyMin,
-                                                            ClientTelemetryOptions.RequestLatencyMax,
-                                                            ClientTelemetryOptions.RequestLatencyPrecision));
-                    latencyHist.RecordValue(storetatistics.RequestLatency.Ticks);
-                }
-            }
-        }
-
         /// <summary>
         /// Dispose of cosmos client.It will get disposed with client so not making it thread safe.
         /// </summary>

Microsoft.Azure.Cosmos/src/Telemetry/ClientTelemetryOptions.cs

@@ -88,15 +88,18 @@ internal static class ClientTelemetryOptions
 
         internal static readonly ResourceType AllowedResourceTypes = ResourceType.Document;
         // Why 5 sec? As of now, if any network request is taking more than 5 millisecond sec, we will consider it slow request this value can be revisited in future
-        private static readonly TimeSpan NetworkLatencyThreshold = TimeSpan.FromMilliseconds(5);
+        internal static readonly TimeSpan NetworkLatencyThreshold = TimeSpan.FromMilliseconds(5);
+        internal static readonly int NetworkRequestsSampleSizeThreshold = 10;
+        
         internal static readonly JsonSerializerSettings JsonSerializerSettings = new JsonSerializerSettings 
         { 
             NullValueHandling = NullValueHandling.Ignore,
             MaxDepth = 64, // https://github.com/advisories/GHSA-5crp-9r3c-p9vr
         };
 
-        private static readonly List<int> ExcludedStatusCodes = new List<int> { 404, 409 };
+        internal static readonly List<int> ExcludedStatusCodes = new List<int> { 404, 409, 412 };
 
+        internal static readonly int NetworkTelemetrySampleSize = 200;
         internal static int PayloadSizeThreshold = 1024 * 1024 * 2; // 2MB
 
         private static Uri clientTelemetryEndpoint;
@@ -180,32 +183,5 @@ internal static string GetEnvironmentName()
             }
             return environmentName;
         }
-
-        /// <summary>
-        /// This method will return true if the request is failed with User or Server Exception and not excluded from telemetry.
-        /// This method will return true if the request latency is more than the threshold.
-        /// otherwise return false
-        /// </summary>
-        /// <param name="statusCode"></param>
-        /// <param name="subStatusCode"></param>
-        /// <param name="latencyInMs"></param>
-        /// <returns>true/false</returns>
-        internal static bool IsEligible(int statusCode, int subStatusCode, TimeSpan latencyInMs)
-        {
-            return
-                ClientTelemetryOptions.IsStatusCodeNotExcluded(statusCode, subStatusCode) && 
-                    (ClientTelemetryOptions.IsUserOrServerError(statusCode) || latencyInMs >= ClientTelemetryOptions.NetworkLatencyThreshold);
-        }
-
-        private static bool IsUserOrServerError(int statusCode)
-        {
-            return statusCode >= 400 && statusCode <= 599;
-        }
-
-        private static bool IsStatusCodeNotExcluded(int statusCode, int subStatusCode)
-        {
-            return !(ClientTelemetryOptions.ExcludedStatusCodes.Contains(statusCode) && subStatusCode == 0);
-        }
-
     }
 }

Microsoft.Azure.Cosmos/src/Telemetry/ClientTelemetryPayloadWriter.cs

@@ -21,7 +21,7 @@ public static async Task SerializedPayloadChunksAsync(
             ClientTelemetryProperties properties,
             ConcurrentDictionary<OperationInfo, (LongConcurrentHistogram latency, LongConcurrentHistogram requestcharge)> operationInfoSnapshot,
             ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram> cacheRefreshInfoSnapshot,
-            ConcurrentDictionary<RequestInfo, LongConcurrentHistogram> requestInfoSnapshot,
+            IReadOnlyList<RequestInfo> sampledRequestInfo,
             Func<string, Task> callback)
         {
             if (properties == null)
@@ -46,11 +46,12 @@ public static async Task SerializedPayloadChunksAsync(
                     OperationInfo payloadForRequestCharge = payloadForLatency.Copy();
                     payloadForRequestCharge.MetricInfo = new MetricInfo(ClientTelemetryOptions.RequestChargeName, ClientTelemetryOptions.RequestChargeUnit);
                     payloadForRequestCharge.SetAggregators(entry.Value.requestcharge, ClientTelemetryOptions.HistogramPrecisionFactor);
-
+                    
                     string latencyMetrics = JsonConvert.SerializeObject(payloadForLatency);
                     string requestChargeMetrics = JsonConvert.SerializeObject(payloadForRequestCharge);
-                    
-                    if (lengthNow + latencyMetrics.Length + requestChargeMetrics.Length > ClientTelemetryOptions.PayloadSizeThreshold)
+
+                    int thisSectionLength = latencyMetrics.Length + requestChargeMetrics.Length;
+                    if (lengthNow + thisSectionLength > ClientTelemetryOptions.PayloadSizeThreshold)
                     {
                         writer.WriteEndArray();
                         writer.WriteEndObject();
@@ -98,21 +99,16 @@ public static async Task SerializedPayloadChunksAsync(
 
             }
 
-            if (requestInfoSnapshot?.Any() == true)
+            if (sampledRequestInfo?.Any() == true)
             {
                 writer.WritePropertyName("requestInfo");
                 writer.WriteStartArray();
 
-                foreach (KeyValuePair<RequestInfo, LongConcurrentHistogram> entry in requestInfoSnapshot)
+                foreach (RequestInfo entry in sampledRequestInfo)
                 {
                     long lengthNow = stringBuilder.Length;
-                    
-                    MetricInfo metricInfo = new MetricInfo(ClientTelemetryOptions.RequestLatencyName, ClientTelemetryOptions.RequestLatencyUnit);
-                    metricInfo.SetAggregators(entry.Value, ClientTelemetryOptions.TicksToMsFactor);
-
-                    RequestInfo payloadForLatency = entry.Key;
-                    payloadForLatency.Metrics.Add(metricInfo);
-                    string latencyMetrics = JsonConvert.SerializeObject(payloadForLatency);
+                  
+                    string latencyMetrics = JsonConvert.SerializeObject(entry);
 
                     if (lengthNow + latencyMetrics.Length > ClientTelemetryOptions.PayloadSizeThreshold)
                     {

Microsoft.Azure.Cosmos/src/Telemetry/ClientTelemetryProcessor.cs

@@ -6,6 +6,7 @@ namespace Microsoft.Azure.Cosmos.Telemetry
 {
     using System;
     using System.Collections.Concurrent;
+    using System.Collections.Generic;
     using System.Net.Http;
     using System.Text;
     using System.Threading;
@@ -22,7 +23,7 @@ internal class ClientTelemetryProcessor
 
         private readonly AuthorizationTokenProvider tokenProvider;
         private readonly CosmosHttpClient httpClient;
-
+            
         internal ClientTelemetryProcessor(CosmosHttpClient httpClient, AuthorizationTokenProvider tokenProvider)
         {
             this.httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
@@ -32,17 +33,12 @@ internal ClientTelemetryProcessor(CosmosHttpClient httpClient, AuthorizationToke
         /// <summary>
         /// It will create Task to process and send client telemetry payload to Client Telemetry Service.
         /// </summary>
-        /// <param name="clientTelemetryInfo"></param>
-        /// <param name="operationInfoSnapshot"></param>
-        /// <param name="cacheRefreshInfoSnapshot"></param>
-        /// <param name="requestInfoSnapshot"></param>
-        /// <param name="cancellationToken"></param>
         /// <returns>Task</returns>
         internal async Task ProcessAndSendAsync(
             ClientTelemetryProperties clientTelemetryInfo, 
-            ConcurrentDictionary<OperationInfo, (LongConcurrentHistogram latency, LongConcurrentHistogram requestcharge)> operationInfoSnapshot,
+            ConcurrentDictionary<OperationInfo, (LongConcurrentHistogram latency, LongConcurrentHistogram requestcharget)> operationInfoSnapshot,
             ConcurrentDictionary<CacheRefreshInfo, LongConcurrentHistogram> cacheRefreshInfoSnapshot,
-            ConcurrentDictionary<RequestInfo, LongConcurrentHistogram> requestInfoSnapshot,
+            IReadOnlyList<RequestInfo> requestInfoSnapshot,
             CancellationToken cancellationToken)
         {
             try
@@ -51,7 +47,7 @@ await ClientTelemetryPayloadWriter.SerializedPayloadChunksAsync(
                     properties: clientTelemetryInfo,
                     operationInfoSnapshot: operationInfoSnapshot,
                     cacheRefreshInfoSnapshot: cacheRefreshInfoSnapshot,
-                    requestInfoSnapshot: requestInfoSnapshot,
+                    sampledRequestInfo: requestInfoSnapshot,
                     callback: async (payload) => await this.SendAsync(clientTelemetryInfo.GlobalDatabaseAccountName, payload, cancellationToken));
             }
             catch (Exception ex)

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

@@ -37,8 +37,14 @@ internal sealed class RequestInfo
 
         public override int GetHashCode()
         {
-            int hash = 3;
+            int hash = this.GetHashCodeForSampler();
             hash = (hash * 7) ^ (this.Uri == null ? 0 : this.Uri.GetHashCode());
+            return hash;
+        }
+
+        public int GetHashCodeForSampler()
+        {
+            int hash = 3;
             hash = (hash * 7) ^ (this.DatabaseName == null ? 0 : this.DatabaseName.GetHashCode());
             hash = (hash * 7) ^ (this.ContainerName == null ? 0 : this.ContainerName.GetHashCode());
             hash = (hash * 7) ^ (this.Operation == null ? 0 : this.Operation.GetHashCode());
@@ -47,7 +53,7 @@ public override int GetHashCode()
             hash = (hash * 7) ^ (this.SubStatusCode.GetHashCode());
             return hash;
         }
-        
+
         public override bool Equals(object obj)
         {
             bool isequal = obj is RequestInfo payload &&
@@ -62,5 +68,26 @@ public override bool Equals(object obj)
             return isequal;
         }
 
+        public double GetP99Latency()
+        {
+            foreach (MetricInfo metric in this.Metrics)
+            {
+                if (metric.MetricsName.Equals(ClientTelemetryOptions.RequestLatencyName, StringComparison.OrdinalIgnoreCase))
+                {
+                    return metric.Percentiles[ClientTelemetryOptions.Percentile99];
+                }
+            }
+            return Double.MinValue; // least prioity for request info w/o latency info
+        }
+
+        public double GetSampleCount()
+        {
+            return (double)this.Metrics[0].Count;
+        }
+
+        public override string ToString()
+        {
+            return "Latency : " + this.GetP99Latency() + ", SampleCount : " + this.GetSampleCount(); 
+        }
     }
 }