Handle changes to MutableSettings and ExporterSettings without rebuilding (#7724)

## Summary of changes

- This is the big one
- Update services to dynamically update when mutable settings or
exporter settings change
- Stop rebuilding everything when there's manual/remote configuration

## Reason for change

This is the "endpoint" that we've been heading for - services only being
disposed/rebuilt at the end of the app, and otherwise only rebuilding
the _necessary_ parts. For example - we don't need to tear down all the
API factories when a customer changes a global tag via remote config;
they only need to change if the `ExporterSettings` change.

The hope is that overall this reduces the overhead of using
configuration in code and/or remote configuration, while also reducing
the number of issues due to managing disposal of services.

## Implementation details

Overall, this PR is kind of a pain. Moving from the "rebuild everything"
to "reconfigure each service" couldn't be done piecemeal, so this is the
one-shot PR. What's more, different services need different patterns
(though we can probably consolidate some of them, this has taken a _lot_
of work and I likely changed patterns unnecessarily in some places).

In general, there's a couple of patterns:

- CI Vis doesn't let you change settings at runtime, so it _never_ needs
to respond to changes. It always just uses the "initial" settings
- Debugger _today_ doesn't respond to changes at runtime (except its own
dynamic config), so for now we ignore Debugger too as it's not really a
regression. I hope we can fix this soon though.
- I've introduced the concept of `Managed*` versions of some services
- These services generally "wrap" the existing type, delegating access
to the underlying service, and handling settings changes
- Many services only care about a sub-set of mutable settings, so they
only update if they need to
- Somewhat annoyingly, setting updates occur on a background thread, so
we need to be careful about thread safety. Where necessary (most places)
I've made sure access to a now-mutable service is done using
`Volatile.Read()` (to ensure changes are visible) and are generally
cached to a local variable (as the underlying field may be updated in
the background).

## Test coverage

In the vast majority of places, this should be covered by existing tests

I plan to add some additional integration tests around reconfiguring and
a bunch of manual testing to make sure I'm confident.

## Other details

I strongly recommend reviewing commit-by-commit. They're generally
self-contained, and hopefully simple enough to understand one commit at
a time.

https://datadoghq.atlassian.net/browse/LANGPLAT-819

Part of a config stack

- #7522
- #7525
- #7530
- #7532
- #7543
- #7544
- #7721
- #7722
- #7695
- #7723
- #7724 👈
- #7796

This isn't the final PR in the stack, as there will be a bunch of
cleaning up to do, but it's the final "implementation" PR

Author: andrewlock

tracer/missing-nullability-files.csv

@@ -34,10 +34,8 @@ src/Datadog.Trace/Tracer.cs
 src/Datadog.Trace/TracerConstants.cs
 src/Datadog.Trace/TracerManager.cs
 src/Datadog.Trace/TracerManagerFactory.cs
-src/Datadog.Trace/Agent/AgentWriter.cs
 src/Datadog.Trace/Agent/Api.cs
 src/Datadog.Trace/Agent/ClientStatsPayload.cs
-src/Datadog.Trace/Agent/IAgentWriter.cs
 src/Datadog.Trace/Agent/IApi.cs
 src/Datadog.Trace/Agent/IApiRequest.cs
 src/Datadog.Trace/Agent/IApiRequestFactory.cs

tracer/src/Datadog.Trace.Tools.Runner/Utils.cs

@@ -417,9 +417,9 @@ public static async Task<AgentConfiguration> CheckAgentConnectionAsync(string ag
 
             var settings = new TracerSettings(configurationSource, new ConfigurationTelemetry(), new OverrideErrorLog());
 
-            Log.Debug("Creating DiscoveryService for: {AgentUri}", settings.Exporter.AgentUri);
-            var discoveryService = DiscoveryService.Create(
-                settings.Exporter,
+            Log.Debug("Creating DiscoveryService for: {AgentUri}", settings.Manager.InitialExporterSettings.AgentUri);
+            var discoveryService = DiscoveryService.CreateUnmanaged(
+                settings.Manager.InitialExporterSettings,
                 tcpTimeout: TimeSpan.FromSeconds(5),
                 initialRetryDelayMs: 200,
                 maxRetryDelayMs: 1000,
@@ -433,7 +433,7 @@ public static async Task<AgentConfiguration> CheckAgentConnectionAsync(string ag
             using (cts.Token.Register(
                        () =>
                        {
-                           WriteError($"Error connecting to the Datadog Agent at {settings.Exporter.AgentUri}.");
+                           WriteError($"Error connecting to the Datadog Agent at {settings.Manager.InitialExporterSettings.AgentUri}.");
                            tcs.TrySetResult(null);
                        }))
             {

tracer/src/Datadog.Trace/Agent/AgentWriter.cs

@@ -3,6 +3,8 @@
 // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
 // </copyright>
 
+#nullable enable
+
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
@@ -12,10 +14,8 @@
 using Datadog.Trace.Configuration;
 using Datadog.Trace.DogStatsd;
 using Datadog.Trace.Logging;
-using Datadog.Trace.Tagging;
 using Datadog.Trace.Telemetry;
 using Datadog.Trace.Telemetry.Metrics;
-using Datadog.Trace.Vendors.StatsdClient;
 
 namespace Datadog.Trace.Agent
 {
@@ -25,10 +25,10 @@ internal class AgentWriter : IAgentWriter
 
         private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor<AgentWriter>();
 
-        private static readonly ArraySegment<byte> EmptyPayload = new(new byte[] { 0x90 });
+        private static readonly ArraySegment<byte> EmptyPayload = new([0x90]);
 
         private readonly ConcurrentQueue<WorkItem> _pendingTraces = new ConcurrentQueue<WorkItem>();
-        private readonly IDogStatsd _statsd;
+        private readonly IStatsdManager _statsd;
         private readonly Task _flushTask;
         private readonly Task _serializationTask;
         private readonly TaskCompletionSource<bool> _processExit = new TaskCompletionSource<bool>();
@@ -43,7 +43,7 @@ internal class AgentWriter : IAgentWriter
         private readonly int _batchInterval;
         private readonly IKeepRateCalculator _traceKeepRateCalculator;
 
-        private readonly IStatsAggregator _statsAggregator;
+        private readonly IStatsAggregator? _statsAggregator;
 
         private readonly bool _apmTracingEnabled;
 
@@ -65,17 +65,28 @@ internal class AgentWriter : IAgentWriter
 
         private long _droppedTraces;
 
-        public AgentWriter(IApi api, IStatsAggregator statsAggregator, IDogStatsd statsd, TracerSettings settings)
-            : this(api, statsAggregator, statsd, maxBufferSize: settings.TraceBufferSize, batchInterval: settings.TraceBatchInterval, apmTracingEnabled: settings.ApmTracingEnabled)
+        private bool _traceMetricsEnabled;
+
+        public AgentWriter(IApi api, IStatsAggregator? statsAggregator, IStatsdManager statsd, TracerSettings settings)
+            : this(api, statsAggregator, statsd, maxBufferSize: settings.TraceBufferSize, batchInterval: settings.TraceBatchInterval, apmTracingEnabled: settings.ApmTracingEnabled, initialTracerMetricsEnabled: settings.Manager.InitialMutableSettings.TracerMetricsEnabled)
         {
+            settings.Manager.SubscribeToChanges(changes =>
+            {
+                if (changes.UpdatedMutable is { } mutable
+                 && mutable.TracerMetricsEnabled != changes.PreviousMutable.TracerMetricsEnabled)
+                {
+                    Volatile.Write(ref _traceMetricsEnabled, mutable.TracerMetricsEnabled);
+                    _statsd.SetRequired(StatsdConsumer.AgentWriter, mutable.TracerMetricsEnabled);
+                }
+            });
         }
 
-        public AgentWriter(IApi api, IStatsAggregator statsAggregator, IDogStatsd statsd, bool automaticFlush = true, int maxBufferSize = 1024 * 1024 * 10, int batchInterval = 100, bool apmTracingEnabled = true)
-        : this(api, statsAggregator, statsd, MovingAverageKeepRateCalculator.CreateDefaultKeepRateCalculator(), automaticFlush, maxBufferSize, batchInterval, apmTracingEnabled)
+        public AgentWriter(IApi api, IStatsAggregator? statsAggregator, IStatsdManager statsd, bool automaticFlush = true, int maxBufferSize = 1024 * 1024 * 10, int batchInterval = 100, bool apmTracingEnabled = true, bool initialTracerMetricsEnabled = false)
+        : this(api, statsAggregator, statsd, MovingAverageKeepRateCalculator.CreateDefaultKeepRateCalculator(), automaticFlush, maxBufferSize, batchInterval, apmTracingEnabled, initialTracerMetricsEnabled)
         {
         }
 
-        internal AgentWriter(IApi api, IStatsAggregator statsAggregator, IDogStatsd statsd, IKeepRateCalculator traceKeepRateCalculator, bool automaticFlush, int maxBufferSize, int batchInterval, bool apmTracingEnabled)
+        internal AgentWriter(IApi api, IStatsAggregator? statsAggregator, IStatsdManager statsd, IKeepRateCalculator traceKeepRateCalculator, bool automaticFlush, int maxBufferSize, int batchInterval, bool apmTracingEnabled, bool initialTracerMetricsEnabled)
         {
             _statsAggregator = statsAggregator;
 
@@ -92,18 +103,20 @@ internal AgentWriter(IApi api, IStatsAggregator statsAggregator, IDogStatsd stat
             _backBuffer = new SpanBuffer(maxBufferSize, formatterResolver);
             _activeBuffer = _frontBuffer;
 
+            _apmTracingEnabled = apmTracingEnabled;
+            _traceMetricsEnabled = initialTracerMetricsEnabled;
+            _statsd.SetRequired(StatsdConsumer.AgentWriter, initialTracerMetricsEnabled);
+
             _serializationTask = automaticFlush ? Task.Factory.StartNew(SerializeTracesLoop, TaskCreationOptions.LongRunning) : Task.CompletedTask;
             _serializationTask.ContinueWith(t => Log.Error(t.Exception, "Error in serialization task"), TaskContinuationOptions.OnlyOnFaulted);
 
             _flushTask = automaticFlush ? Task.Run(FlushBuffersTaskLoopAsync) : Task.CompletedTask;
             _flushTask.ContinueWith(t => Log.Error(t.Exception, "Error in flush task"), TaskContinuationOptions.OnlyOnFaulted);
 
             _backBufferFlushTask = _frontBufferFlushTask = Task.CompletedTask;
-
-            _apmTracingEnabled = apmTracingEnabled;
         }
 
-        internal event Action Flushed;
+        internal event Action? Flushed;
 
         internal SpanBuffer ActiveBuffer => _activeBuffer;
 
@@ -138,10 +151,14 @@ public void WriteTrace(ArraySegment<Span> trace)
                 }
             }
 
-            if (_statsd != null)
+            if (Volatile.Read(ref _traceMetricsEnabled))
             {
-                _statsd.Increment(TracerMetricNames.Queue.EnqueuedTraces);
-                _statsd.Increment(TracerMetricNames.Queue.EnqueuedSpans, trace.Count);
+                using var lease = _statsd.TryGetClientLease();
+                if (lease.Client is { } statsd)
+                {
+                    statsd.Increment(TracerMetricNames.Queue.EnqueuedTraces);
+                    statsd.Increment(TracerMetricNames.Queue.EnqueuedSpans, trace.Count);
+                }
             }
         }
 
@@ -244,7 +261,7 @@ private async Task FlushBuffersTaskLoopAsync()
             {
                 tasks[2] = Task.Delay(TimeSpan.FromSeconds(1));
                 await Task.WhenAny(tasks).ConfigureAwait(false);
-                tasks[2] = null;
+                tasks[2] = null!;
 
                 if (_forceFlush.Task.IsCompleted)
                 {
@@ -314,10 +331,14 @@ async Task InternalBufferFlush()
 
                 try
                 {
-                    if (_statsd != null)
+                    if (Volatile.Read(ref _traceMetricsEnabled))
                     {
-                        _statsd.Increment(TracerMetricNames.Queue.DequeuedTraces, buffer.TraceCount);
-                        _statsd.Increment(TracerMetricNames.Queue.DequeuedSpans, buffer.SpanCount);
+                        using var lease = _statsd.TryGetClientLease();
+                        if (lease.Client is { } statsd)
+                        {
+                            statsd.Increment(TracerMetricNames.Queue.DequeuedTraces, buffer.TraceCount);
+                            statsd.Increment(TracerMetricNames.Queue.DequeuedSpans, buffer.SpanCount);
+                        }
                     }
 
                     var droppedTraces = Interlocked.Exchange(ref _droppedTraces, 0);
@@ -336,7 +357,7 @@ async Task InternalBufferFlush()
                         {
                             droppedP0Traces = Interlocked.Exchange(ref _droppedP0Traces, 0);
                             droppedP0Spans = Interlocked.Exchange(ref _droppedP0Spans, 0);
-                            Log.Debug<int, int, long, long>("Flushing {Spans} spans across {Traces} traces. CanComputeStats is enabled with {DroppedP0Traces} droppedP0Traces and {DroppedP0Spans} droppedP0Spans", buffer.SpanCount, buffer.TraceCount, droppedP0Traces, droppedP0Spans);
+                            Log.Debug("Flushing {Spans} spans across {Traces} traces. CanComputeStats is enabled with {DroppedP0Traces} droppedP0Traces and {DroppedP0Spans} droppedP0Spans", buffer.SpanCount, buffer.TraceCount, droppedP0Traces, droppedP0Spans);
                             // Metrics for unsampled traces/spans already recorded
                         }
                         else
@@ -377,7 +398,7 @@ async Task InternalBufferFlush()
         private void SerializeTrace(ArraySegment<Span> spans)
         {
             // Declaring as inline method because only safe to invoke in the context of SerializeTrace
-            SpanBuffer SwapBuffers()
+            SpanBuffer? SwapBuffers()
             {
                 if (_activeBuffer == _frontBuffer)
                 {
@@ -512,10 +533,14 @@ private void DropTrace(ArraySegment<Span> spans)
             TelemetryFactory.Metrics.RecordCountSpanDropped(MetricTags.DropReason.OverfullBuffer, spans.Count);
             TelemetryFactory.Metrics.RecordCountTraceChunkDropped(MetricTags.DropReason.OverfullBuffer);
 
-            if (_statsd != null)
+            if (Volatile.Read(ref _traceMetricsEnabled))
             {
-                _statsd.Increment(TracerMetricNames.Queue.DroppedTraces);
-                _statsd.Increment(TracerMetricNames.Queue.DroppedSpans, spans.Count);
+                using var lease = _statsd.TryGetClientLease();
+                if (lease.Client is { } statsd)
+                {
+                    statsd.Increment(TracerMetricNames.Queue.DroppedTraces);
+                    statsd.Increment(TracerMetricNames.Queue.DroppedSpans, spans.Count);
+                }
             }
         }
 
@@ -578,7 +603,7 @@ private void SerializeTracesLoop()
         private readonly struct WorkItem
         {
             public readonly ArraySegment<Span> Trace;
-            public readonly Action Callback;
+            public readonly Action? Callback;
 
             public WorkItem(ArraySegment<Span> trace)
             {

tracer/src/Datadog.Trace/Agent/Api.cs

@@ -5,8 +5,10 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Net.Sockets;
+using System.Threading;
 using System.Threading.Tasks;
 using Datadog.Trace.Agent.Transports;
 using Datadog.Trace.DogStatsd;
@@ -32,7 +34,7 @@ internal class Api : IApi
 
         private readonly IDatadogLogger _log;
         private readonly IApiRequestFactory _apiRequestFactory;
-        private readonly IDogStatsd _statsd;
+        private readonly IStatsdManager _statsd;
         private readonly string _containerId;
         private readonly string _entityId;
         private readonly Uri _tracesEndpoint;
@@ -43,12 +45,14 @@ internal class Api : IApi
         private readonly SendCallback<SendTracesState> _sendTraces;
         private string _cachedResponse;
         private string _agentVersion;
+        private bool _healthMetricsEnabled;
 
         public Api(
             IApiRequestFactory apiRequestFactory,
-            IDogStatsd statsd,
+            IStatsdManager statsd,
             Action<Dictionary<string, float>> updateSampleRates,
             bool partialFlushEnabled,
+            bool healthMetricsEnabled,
             IDatadogLogger log = null)
         {
             // optionally injecting a log instance in here for testing purposes
@@ -58,10 +62,12 @@ public Api(
             _sendTraces = SendTracesAsyncImpl;
             _updateSampleRates = updateSampleRates;
             _statsd = statsd;
+            ToggleTracerHealthMetrics(healthMetricsEnabled);
             _containerId = ContainerMetadata.GetContainerId();
             _entityId = ContainerMetadata.GetEntityId();
             _apiRequestFactory = apiRequestFactory;
             _partialFlushEnabled = partialFlushEnabled;
+            _healthMetricsEnabled = healthMetricsEnabled;
             _tracesEndpoint = _apiRequestFactory.GetEndpoint(TracesPath);
             _log.Debug("Using traces endpoint {TracesEndpoint}", _tracesEndpoint.ToString());
             _statsEndpoint = _apiRequestFactory.GetEndpoint(StatsPath);
@@ -77,6 +83,13 @@ private enum SendResult
             Failed_DontRetry,
         }
 
+        [MemberNotNull(nameof(_statsd))]
+        public void ToggleTracerHealthMetrics(bool enabled)
+        {
+            Volatile.Write(ref _healthMetricsEnabled, enabled);
+            _statsd.SetRequired(StatsdConsumer.TraceApi, enabled);
+        }
+
         public Task<bool> SendStatsAsync(StatsBuffer stats, long bucketDuration)
         {
             _log.Debug("Sending stats to the Datadog Agent.");
@@ -299,10 +312,13 @@ private async Task<SendResult> SendTracesAsyncImpl(IApiRequest request, bool fin
 
             try
             {
+                var healthMetricsEnabled = Volatile.Read(ref _healthMetricsEnabled);
+                using var lease = healthMetricsEnabled ? _statsd.TryGetClientLease() : default;
+                var healthStats = healthMetricsEnabled ? lease.Client : null;
                 try
                 {
                     TelemetryFactory.Metrics.RecordCountTraceApiRequests();
-                    _statsd?.Increment(TracerMetricNames.Api.Requests);
+                    healthStats?.Increment(TracerMetricNames.Api.Requests);
                     response = await request.PostAsync(traces, MimeTypes.MsgPack).ConfigureAwait(false);
                 }
                 catch (Exception ex)
@@ -311,17 +327,17 @@ private async Task<SendResult> SendTracesAsyncImpl(IApiRequest request, bool fin
                     // (which are handled below)
                     var tag = ex is TimeoutException ? MetricTags.ApiError.Timeout : MetricTags.ApiError.NetworkError;
                     TelemetryFactory.Metrics.RecordCountTraceApiErrors(tag);
-                    _statsd?.Increment(TracerMetricNames.Api.Errors);
+                    healthStats?.Increment(TracerMetricNames.Api.Errors);
                     throw;
                 }
 
-                if (_statsd != null)
+                if (healthStats != null)
                 {
                     // don't bother creating the tags array if trace metrics are disabled
                     string[] tags = { $"status:{response.StatusCode}" };
 
                     // count every response, grouped by status code
-                    _statsd?.Increment(TracerMetricNames.Api.Responses, tags: tags);
+                    healthStats.Increment(TracerMetricNames.Api.Responses, tags: tags);
                 }
 
                 TelemetryFactory.Metrics.RecordCountTraceApiResponses(response.GetTelemetryStatusCodeMetricTag());

tracer/src/Datadog.Trace/Agent/ClientStatsPayload.cs

@@ -4,21 +4,29 @@
 // </copyright>
 
 using System.Threading;
+using Datadog.Trace.Configuration;
 
 namespace Datadog.Trace.Agent
 {
-    internal class ClientStatsPayload
+    internal class ClientStatsPayload(MutableSettings settings)
     {
+        private AppSettings _settings = CreateSettings(settings);
         private long _sequence;
 
-        public string HostName { get; set; }
+        public string HostName { get; init; }
 
-        public string Environment { get; set; }
+        public AppSettings Details => _settings;
 
-        public string Version { get; set; }
-
-        public string ProcessTags { get; set; }
+        public string ProcessTags { get; init; }
 
         public long GetSequenceNumber() => Interlocked.Increment(ref _sequence);
+
+        public void UpdateDetails(MutableSettings settings)
+            => Interlocked.Exchange(ref _settings, CreateSettings(settings));
+
+        private static AppSettings CreateSettings(MutableSettings settings)
+            => new(settings.Environment, settings.ServiceVersion);
+
+        internal record AppSettings(string Environment, string Version);
     }
 }