NCBC-3896: SDK Telemetry Collection in Server

Motivation
----------
Implement AppTelemetry reporting.

Changes
-------
- Added AppTelemetry options
- Added property AppTelemetryPath in NodesExt
- AppTelemetryCollector is added to the ServiceProvider from ClusterOptions
- An instance of AppTelemetryCollector is always present in the ServiceProvider, however it can be disabled which prevents it from collecting metrics.
- Adjusted Unit Tests mocks to support ServiceUriProvider and Services' Client changes
- Added AppTelemetry incrementing in Query, Analytics, Collection, User and Bucket Managers
- Histograms are not updated for Timeouts and Cancellations, only counter. This is due to the ambiguity of the behaviour in AppTelemetry/Orphan in RFCs
- Added Canonical and Alternate Hostname properties to NodeAdapter
- Added alt_node property to metrics
- The LightweightStopwatch is now provided by the AppTelemetryCollector, which will return null when disabled. This prevents creating a new LightweightStopWatch when it isn't needed.

Change-Id: I5d8d389eb23259531e078c83a64f038649187c69
Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/222591
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Emilien Bevierre <emilien.bevierre@couchbase.com>

Author: emilienbev

src/Couchbase/Analytics/AnalyticsClient.cs

@@ -1,10 +1,12 @@
 using System;
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Net;
 using System.Net.Http;
 using System.Text;
 using System.Threading.Tasks;
 using Couchbase.Core;
+using Couchbase.Core.Diagnostics.Metrics.AppTelemetry;
 using Couchbase.Core.Diagnostics.Tracing;
 using Couchbase.Core.Exceptions;
 using Couchbase.Core.Exceptions.Analytics;
@@ -29,6 +31,7 @@ internal class AnalyticsClient : HttpServiceBase, IAnalyticsClient
         private readonly ITypeSerializer _typeSerializer;
         private readonly ILogger<AnalyticsClient> _logger;
         private readonly IRequestTracer _tracer;
+        private readonly IAppTelemetryCollector _appTelemetryCollector;
         internal const string AnalyticsPriorityHeaderName = "Analytics-Priority";
 
         [RequiresUnreferencedCode(AnalyticsRequiresUnreferencedMembersWarning)]
@@ -38,13 +41,15 @@ public AnalyticsClient(
             IServiceUriProvider serviceUriProvider,
             ITypeSerializer typeSerializer,
             ILogger<AnalyticsClient> logger,
-            IRequestTracer tracer)
+            IRequestTracer tracer,
+            IAppTelemetryCollector appTelemetryCollector)
             : base(httpClientFactory)
         {
             _serviceUriProvider = serviceUriProvider ?? throw new ArgumentNullException(nameof(serviceUriProvider));
             _typeSerializer = typeSerializer ?? throw new ArgumentNullException(nameof(typeSerializer));
             _logger = logger ?? throw new ArgumentNullException(nameof(logger));
             _tracer = tracer;
+            _appTelemetryCollector = appTelemetryCollector;
         }
 
         /// <inheritdoc />
@@ -61,7 +66,11 @@ public async Task<IAnalyticsResult<T>> QueryAsync<T>(string statement, Analytics
                 .WithLocalAddress();
 
             // try get Analytics node
-            var analyticsUri = _serviceUriProvider.GetRandomAnalyticsUri();
+            var analyticsNode = _serviceUriProvider.GetRandomAnalyticsNode();
+            var analyticsUri = analyticsNode.AnalyticsUri;
+            var requestStopwatch = _appTelemetryCollector.StartNewLightweightStopwatch();
+            TimeSpan? operationElapsed;
+
             rootSpan.WithRemoteAddress(analyticsUri);
 
             _logger.LogDebug("Sending analytics query with a context id {contextId} to server {searchUri}",
@@ -91,8 +100,10 @@ public async Task<IAnalyticsResult<T>> QueryAsync<T>(string statement, Analytics
                     var httpClient = CreateHttpClient(options.TimeoutValue);
                     try
                     {
+                        requestStopwatch?.Restart();
                         var response = await httpClient.SendAsync(request, HttpClientFactory.DefaultCompletionOption, options.Token)
                             .ConfigureAwait(false);
+                        operationElapsed = requestStopwatch?.Elapsed;
                         dispatchSpan.Dispose();
 
                         var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
@@ -135,6 +146,14 @@ public async Task<IAnalyticsResult<T>> QueryAsync<T>(string statement, Analytics
                             CouchbaseException? ex = CreateExceptionForError(result, context, false);
                             if (ex != null) { throw ex; }
                         }
+
+                        _appTelemetryCollector.IncrementMetrics(
+                            operationElapsed,
+                            analyticsNode.NodesAdapter.CanonicalHostname,
+                            analyticsNode.NodesAdapter.AlternateHostname,
+                            analyticsNode.NodeUuid,
+                            AppTelemetryServiceType.Analytics,
+                            AppTelemetryCounterType.Total);
                     }
                     catch
                     {
@@ -148,9 +167,18 @@ public async Task<IAnalyticsResult<T>> QueryAsync<T>(string statement, Analytics
                 }
                 catch (OperationCanceledException e)
                 {
+                    operationElapsed = requestStopwatch?.Elapsed;
                     //treat as an orphaned response
                     rootSpan.LogOrphaned();
 
+                    _appTelemetryCollector.IncrementMetrics(
+                        operationElapsed,
+                        analyticsNode.NodesAdapter.CanonicalHostname,
+                        analyticsNode.NodesAdapter.AlternateHostname,
+                        analyticsNode.NodeUuid,
+                        AppTelemetryServiceType.Analytics,
+                        AppTelemetryCounterType.TimedOut);
+
                     var context = new AnalyticsErrorContext
                     {
                         ClientContextId = options.ClientContextIdValue,
@@ -174,9 +202,18 @@ public async Task<IAnalyticsResult<T>> QueryAsync<T>(string statement, Analytics
                 }
                 catch (HttpRequestException e)
                 {
+                    operationElapsed = requestStopwatch?.Elapsed;
                     //treat as an orphaned response
                     rootSpan.LogOrphaned();
 
+                    _appTelemetryCollector.IncrementMetrics(
+                        operationElapsed,
+                        analyticsNode.NodesAdapter.CanonicalHostname,
+                        analyticsNode.NodesAdapter.AlternateHostname,
+                        analyticsNode.NodeUuid,
+                        AppTelemetryServiceType.Analytics,
+                        AppTelemetryCounterType.Canceled);
+
                     var context = new AnalyticsErrorContext
                     {
                         ClientContextId = options.ClientContextIdValue,

src/Couchbase/CStringParams.cs

@@ -31,6 +31,18 @@ internal static class CStringParams
         public const string NetworkResolution = "network";
         public const string PreferredServerGroup = "preferred_server_group";
 
+        [InterfaceStability(Level.Volatile)]
+        public const string AppTelemetryEndpoint = "app_telemetry_endpoint";
+
+        [InterfaceStability(Level.Volatile)]
+        public const string AppTelemetryBackoff = "app_telemetry_backoff";
+
+        [InterfaceStability(Level.Volatile)]
+        public const string AppTelemetryPingInterval = "app_telemetry_ping_interval";
+
+        [InterfaceStability(Level.Volatile)]
+        public const string AppTelemetryPingTimeout = "app_telemetry_ping_timeout";
+
         [InterfaceStability(Level.Uncommitted)]
         public const string RandomSeedNodes = "random_seed_nodes";
     }

src/Couchbase/Cluster.cs

@@ -30,9 +30,11 @@
 using Microsoft.Extensions.Logging;
 using AnalyticsOptions = Couchbase.Analytics.AnalyticsOptions;
 using Couchbase.Core.RateLimiting;
+using Couchbase.Core.Diagnostics.Metrics.AppTelemetry;
 using Couchbase.Management.Eventing.Internal;
 using Couchbase.Search.Queries.Simple;
 using Couchbase.Search.Queries.Vector;
+using Couchbase.Utils;
 
 #nullable enable
 
@@ -59,6 +61,7 @@ public class Cluster : ICluster, IBootstrappable
         private readonly IRequestTracer _tracer;
         private readonly IRetryStrategy _retryStrategy;
         private readonly MeterForwarder? _meterForwarder;
+        private readonly IAppTelemetryCollector _appTelemetryCollector;
 
         // Internal is used to provide a seam for unit tests
         internal LazyService<IQueryClient> LazyQueryClient;
@@ -120,6 +123,10 @@ internal Cluster(ClusterOptions clusterOptions)
                 _meterForwarder = new MeterForwarder(meter);
             }
 
+            _appTelemetryCollector = _context.ServiceProvider.GetRequiredService<IAppTelemetryCollector>();
+            _appTelemetryCollector.ClusterContext = _context;
+            _appTelemetryCollector.Initialize();
+
             var bootstrapperFactory = _context.ServiceProvider.GetRequiredService<IBootstrapperFactory>();
             _bootstrapper = bootstrapperFactory.Create(clusterOptions.BootstrapPollInterval);
         }
@@ -439,6 +446,7 @@ public async Task<ISearchResult> SearchAsync(
         {
             options ??= new SearchOptions();
             options.TimeoutValue ??= _context.ClusterOptions.SearchTimeout;
+            options.Token = options.Token.FallbackToTimeout(options.TimeoutValue.Value)?.Token ?? options.Token;
 
             ThrowIfNotBootstrapped();
             // The RFC asks to wait until the GlobalConfig is ready.  That should be handled by ThrowIfNotBootstrapped.
@@ -621,6 +629,7 @@ public void Dispose()
                 _bootstrapper.Dispose();
                 _context.Dispose();
                 _meterForwarder?.Dispose();
+                _appTelemetryCollector.Dispose();
             }
         }
 

src/Couchbase/ClusterOptions.cs

@@ -22,6 +22,7 @@
 using Couchbase.Core.Logging;
 using Couchbase.Core.Retry;
 using Couchbase.Client.Transactions.Config;
+using Couchbase.Core.Diagnostics.Metrics.AppTelemetry;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
 
@@ -154,6 +155,22 @@ public string? ConnectionString
                     {
                         ConnectionStringValue.RandomizeSeedHosts = randomizeSeedNodes;
                     }
+                    if (ConnectionStringValue.TryGetParameter(CStringParams.AppTelemetryEndpoint, out string appTelemetryEndpoint))
+                    {
+                        AppTelemetry.Endpoint = new Uri(appTelemetryEndpoint);
+                    }
+                    if (ConnectionStringValue.TryGetParameter(CStringParams.AppTelemetryBackoff, out TimeSpan appTelemetryBackoff))
+                    {
+                        AppTelemetry.Backoff = appTelemetryBackoff;
+                    }
+                    if (ConnectionStringValue.TryGetParameter(CStringParams.AppTelemetryPingInterval, out TimeSpan appTelemetryPingInterval))
+                    {
+                        AppTelemetry.PingInterval = appTelemetryPingInterval;
+                    }
+                    if (ConnectionStringValue.TryGetParameter(CStringParams.AppTelemetryPingTimeout, out TimeSpan appTelemetryPingTimeout))
+                    {
+                        AppTelemetry.PingTimeout = appTelemetryPingTimeout;
+                    }
                     if (ConnectionStringValue.TryGetParameter(CStringParams.PreferredServerGroup, out string serverGroup))
                     {
                         PreferredServerGroup = serverGroup;
@@ -550,6 +567,74 @@ public ClusterOptions WithLoggingMeterOptions(Action<LoggingMeterOptions> config
             return WithLoggingMeterOptions(opts);
         }
 
+        /// <summary>
+        /// Configures the endpoint where AppTelemetry data should be sent to.
+        /// </summary>
+        /// <returns>
+        /// A <see cref="ClusterOptions"/> object for chaining.
+        /// </returns>
+        [InterfaceStability(Level.Volatile)]
+        public ClusterOptions WithAppTelemetryEndpoint(Uri endpoint)
+        {
+            AppTelemetry.Endpoint = endpoint;
+            AppTelemetry.Enabled = true;
+            return this;
+        }
+
+        /// <summary>
+        /// Configures the backoff for re-connecting to the AppTelemetry endpoint via WebSockets.
+        /// </summary>
+        /// <returns>
+        /// A <see cref="ClusterOptions"/> object for chaining.
+        /// </returns>
+        [InterfaceStability(Level.Volatile)]
+        public ClusterOptions WithAppTelemetryBackoff(TimeSpan backoff)
+        {
+            AppTelemetry.Backoff = backoff;
+            AppTelemetry.Enabled = true;
+            return this;
+        }
+
+        /// <summary>
+        /// Enables/Disables AppTelemetry.
+        /// </summary>
+        /// <returns>
+        /// A <see cref="ClusterOptions"/> object for chaining.
+        /// </returns>
+        [InterfaceStability(Level.Volatile)]
+        public ClusterOptions WithAppTelemetryEnabled(bool enabled)
+        {
+            AppTelemetry.Enabled = enabled;
+            return this;
+        }
+
+        /// <summary>
+        /// Configures the duration between consecutive PING commands sent to the server.
+        /// </summary>
+        /// <returns>
+        /// A <see cref="ClusterOptions"/> object for chaining.</returns>
+        [InterfaceStability(Level.Volatile)]
+        public ClusterOptions WithAppTelemetryPingInterval(TimeSpan interval)
+        {
+            AppTelemetry.PingInterval = interval;
+            AppTelemetry.Enabled = true;
+            return this;
+        }
+
+        /// <summary>
+        /// Configures the maximum timeout for the server to respond to a PING.
+        /// </summary>
+        /// <param name="timeout"></param>
+        /// <returns>
+        /// A <see cref="ClusterOptions"/> object for chaining.</returns>
+        [InterfaceStability(Level.Volatile)]
+        public ClusterOptions WithAppTelemetryPingTimeout(TimeSpan timeout)
+        {
+            AppTelemetry.PingTimeout = timeout;
+            AppTelemetry.Enabled = true;
+            return this;
+        }
+
         #endregion
 
         /// <summary>
@@ -843,6 +928,8 @@ public bool IgnoreRemoteCertificateNameMismatch
         /// <inheritdoc cref="ExperimentalOptions"/>
         public ExperimentalOptions Experiments { get; set; } = new();
 
+        public AppTelemetryOptions AppTelemetry { get; set; } = new();
+
         /// <summary>
         /// Provides a default implementation of <see cref="ClusterOptions"/>.
         /// </summary>