TCP P: observability — metadata callbacks, OpenTelemetry tracing and ILogger

ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.Reflection;
 using ClickHouse.Driver.Compression;
+using Microsoft.Extensions.Logging.Abstractions;
 
 namespace ClickHouse.Driver.Tcp.Tests.Client;
 
@@ -34,6 +35,8 @@ public void Defaults_WhenNotOverridden_MatchNativeProtocolConventions()
             Assert.That(options.MaxConnectionLifetime, Is.EqualTo(TimeSpan.FromMinutes(30)));
             Assert.That(options.IdleTimeout, Is.EqualTo(TimeSpan.FromMinutes(5)));
             Assert.That(options.PoolReusePolicy, Is.EqualTo(ClickHouseTcpPoolReusePolicy.Lifo));
+            Assert.That(options.IncludeSqlInActivityTags, Is.False);
+            Assert.That(options.StatementMaxLength, Is.EqualTo(5), "a stub, so statement text has to be asked for");
         });
     }
 
@@ -470,6 +473,9 @@ public void WithOwnedCustomSettings_CopiesEveryPropertyAndSnapshotsTheSettings()
             IdleTimeout = TimeSpan.FromSeconds(7),
             SweepInterval = TimeSpan.FromSeconds(8),
             PoolReusePolicy = ClickHouseTcpPoolReusePolicy.Fifo,
+            LoggerFactory = NullLoggerFactory.Instance,
+            IncludeSqlInActivityTags = true,
+            StatementMaxLength = 42,
 
             // Zstd rather than Lz4 so this stays non-default whichever codec the default becomes.
             Compressor = ZstdCompressor.Default,

ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpProgressTests.cs

@@ -0,0 +1,44 @@
+namespace ClickHouse.Driver.Tcp.Tests.Client;
+
+// The accumulator exists because every consumer of OnProgress has to add the packets up, and a client that made
+// them discover that for themselves would have most of them reporting the last step as the total.
+[TestFixture]
+public class ClickHouseTcpProgressTests
+{
+    [Test]
+    public void OperatorPlus_TwoIncrements_SumsEveryCounter()
+    {
+        var first = new ClickHouseTcpProgress(rows: 1, bytes: 2, totalRows: 3, wroteRows: 4, wroteBytes: 5, elapsedNs: 6);
+        var second = new ClickHouseTcpProgress(rows: 10, bytes: 20, totalRows: 30, wroteRows: 40, wroteBytes: 50, elapsedNs: 60);
+
+        ClickHouseTcpProgress total = first + second;
+
+        Assert.Multiple(() =>
+        {
+            Assert.That(total.Rows, Is.EqualTo(11UL));
+            Assert.That(total.Bytes, Is.EqualTo(22UL));
+            Assert.That(total.TotalRows, Is.EqualTo(33UL));
+            Assert.That(total.WroteRows, Is.EqualTo(44UL));
+            Assert.That(total.WroteBytes, Is.EqualTo(55UL));
+            Assert.That(total.ElapsedNs, Is.EqualTo(66UL));
+        });
+    }
+
+    [Test]
+    public void Add_TwoIncrements_MatchesTheOperator()
+    {
+        var first = new ClickHouseTcpProgress(1, 2, 3, 4, 5, 6);
+        var second = new ClickHouseTcpProgress(10, 20, 30, 40, 50, 60);
+
+        Assert.That(ClickHouseTcpProgress.Add(first, second), Is.EqualTo(first + second));
+    }
+
+    [Test]
+    public void OperatorPlus_DefaultSeed_IsTheIdentity()
+    {
+        // So a caller can fold a sequence starting from default without a special first case.
+        var increment = new ClickHouseTcpProgress(1, 2, 3, 4, 5, 6);
+
+        Assert.That(default(ClickHouseTcpProgress) + increment, Is.EqualTo(increment));
+    }
+}

ClickHouse.Driver.Tcp.Tests/Client/ConnectionPoolLoggingTests.cs

@@ -0,0 +1,220 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using ClickHouse.Driver.Tcp.Client;
+using ClickHouse.Driver.Tcp.Logging;
+using ClickHouse.Driver.Tcp.Tests.Utilities;
+using Microsoft.Extensions.Logging;
+
+namespace ClickHouse.Driver.Tcp.Tests.Client;
+
+// What the pool logs, over connections that need no server. Two of these events are reachable no other way: a
+// background top-up and a sweep both swallow their exceptions by design, so without a logger a failure in either
+// leaves no trace at all and no test could observe it.
+[TestFixture]
+public class ConnectionPoolLoggingTests
+{
+    private static readonly CancellationToken None = CancellationToken.None;
+
+    private CapturingLoggerFactory factory;
+
+    [SetUp]
+    public void CreateFactory() => factory = new CapturingLoggerFactory();
+
+    [TearDown]
+    public void DisposeFactory() => factory.Dispose();
+
+    private CapturingLogger Log => factory.Logger(ClickHouseTcpDiagnostics.PoolLogCategory);
+
+    private ClickHouseTcpClientOptions Options(
+        int maxPoolSize = 4,
+        int minPoolSize = 0,
+        TimeSpan? poolTimeout = null,
+        TimeSpan? maxConnectionLifetime = null,
+        TimeSpan? idleTimeout = null)
+        => new()
+        {
+            MaxPoolSize = maxPoolSize,
+            MinPoolSize = minPoolSize,
+            PoolTimeout = poolTimeout ?? TimeSpan.FromSeconds(30),
+            MaxConnectionLifetime = maxConnectionLifetime ?? TimeSpan.FromMinutes(30),
+            IdleTimeout = idleTimeout ?? TimeSpan.FromMinutes(5),
+            LoggerFactory = factory,
+        };
+
+    [Test]
+    public async Task RentAsync_NoLoggerFactory_AsksForNoLogger()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options() with { LoggerFactory = null }, connections, new ControlledTimeProvider());
+
+        await using IConnectionLease lease = await pool.RentAsync(None);
+
+        Assert.That(factory.Categories, Is.Empty, "no factory configured means nothing is created per pool");
+    }
+
+    [Test]
+    public async Task RentAsync_FirstRent_LogsThatItIsOpeningOne()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options(), connections, new ControlledTimeProvider());
+
+        await using IConnectionLease lease = await pool.RentAsync(None);
+
+        Assert.That(Log.WithEventId(3001), Is.Not.Empty, "no idle connection to reuse");
+    }
+
+    [Test]
+    public async Task RentAsync_AfterALeaseIsReturned_LogsTheReuse()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options(), connections, new ControlledTimeProvider());
+
+        await using (IConnectionLease first = await pool.RentAsync(None))
+        {
+        }
+
+        await using IConnectionLease second = await pool.RentAsync(None);
+
+        LogEntry reused = Log.WithEventId(3000).Single();
+        Assert.Multiple(() =>
+        {
+            Assert.That(reused.Level, Is.EqualTo(LogLevel.Trace), "a per-operation line belongs at Trace");
+
+            // The count is this checkout's, so the first reuse is the connection's second operation. Reading it
+            // before the checkout records itself would report the previous number.
+            Assert.That(reused.Message, Does.Contain("its 2 operation"));
+        });
+    }
+
+    [Test]
+    public async Task Return_ConnectionNoLongerUsable_LogsTheDiscard()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options(), connections, new ControlledTimeProvider());
+
+        await using (IConnectionLease lease = await pool.RentAsync(None))
+        {
+            // A terminated connection is what a failed or abandoned operation leaves behind, and the pool closes
+            // it on return rather than handing it to the next caller.
+            lease.Connection.Terminate();
+        }
+
+        Assert.That(Log.WithEventId(3002), Is.Not.Empty, "the discard is reported");
+    }
+
+    [Test]
+    public async Task Sweep_IdleConnectionPastItsLifetime_LogsWhatItRetired()
+    {
+        var time = new ControlledTimeProvider();
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options(maxConnectionLifetime: TimeSpan.FromMinutes(1)), connections, time);
+
+        await using (IConnectionLease lease = await pool.RentAsync(None))
+        {
+        }
+
+        time.Advance(TimeSpan.FromMinutes(2));
+        pool.Sweep();
+
+        LogEntry retired = Log.WithEventId(3003).Single();
+        Assert.That(retired.Message, Does.Contain("Retired 1"));
+    }
+
+    [Test]
+    public async Task Sweep_NothingExpired_LogsNothing()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(Options(), connections, new ControlledTimeProvider());
+
+        await using (IConnectionLease lease = await pool.RentAsync(None))
+        {
+        }
+
+        pool.Sweep();
+
+        Assert.That(Log.WithEventId(3003), Is.Empty, "a sweep that retires nothing is not worth a line");
+    }
+
+    [Test]
+    public async Task RentAsync_PoolExhausted_LogsAWarningBeforeThrowing()
+    {
+        var connections = new FakeConnectionFactory();
+        await using var pool = new ConnectionPool(
+            Options(maxPoolSize: 1, poolTimeout: TimeSpan.FromMilliseconds(50)),
+            connections,
+            new ControlledTimeProvider());
+
+        await using IConnectionLease held = await pool.RentAsync(None);
+
+        Assert.ThrowsAsync<TimeoutException>(async () => await pool.RentAsync(None));
+
+        LogEntry exhausted = Log.WithEventId(3004).Single();
+        Assert.Multiple(() =>
+        {
+            Assert.That(exhausted.Level, Is.EqualTo(LogLevel.Warning));
+            Assert.That(exhausted.Message, Does.Contain("PoolTimeout"));
+        });
+    }
+
+    [Test]
+    public async Task Sweep_BackgroundTopUpDialFails_LogsTheFailureNobodyElseSees()
+    {
+        var time = new ControlledTimeProvider();
+        var connections = new FakeConnectionFactory { FailNextWith = new InvalidOperationException("dial refused") };
+        await using var pool = new ConnectionPool(Options(minPoolSize: 1), connections, time);
+
+        pool.Sweep();
+        if (pool.LastRefill is not null)
+        {
+            await pool.LastRefill;
+        }
+
+        LogEntry failed = Log.WithEventId(3005).Single();
+        Assert.Multiple(() =>
+        {
+            Assert.That(failed.Level, Is.EqualTo(LogLevel.Warning));
+            Assert.That(failed.Exception, Is.TypeOf<InvalidOperationException>(), "the swallowed exception reaches the log");
+        });
+    }
+
+    [Test]
+    public async Task DisposeAsync_OpenPool_LogsTheDrain()
+    {
+        var connections = new FakeConnectionFactory();
+        var pool = new ConnectionPool(Options(), connections, new ControlledTimeProvider());
+
+        await using (IConnectionLease lease = await pool.RentAsync(None))
+        {
+        }
+
+        await pool.DisposeAsync();
+
+        LogEntry draining = Log.WithEventId(3007).Single();
+        Assert.That(draining.Message, Does.Contain("closing 1 idle"));
+    }
+
+    [Test]
+    public async Task RentAsync_ThrowingLogger_StillHandsOverTheConnection()
+    {
+        // The pool logs at the points a connection is between owners: taken out of the idle list but not yet
+        // leased, or out of the leased set but not yet closed. An exception from any of those calls would leave a
+        // socket with nobody left to close it, so a broken logger would read as the client leaking connections.
+        var connections = new FakeConnectionFactory();
+        var pool = new ConnectionPool(Options() with { LoggerFactory = new ThrowingLoggerFactory() }, connections, new ControlledTimeProvider());
+
+        await using (IConnectionLease first = await pool.RentAsync(None))
+        {
+        }
+
+        await using (IConnectionLease reused = await pool.RentAsync(None))
+        {
+            Assert.That(connections.CreateCount, Is.EqualTo(1), "the reuse path survived its log call");
+        }
+
+        await pool.DisposeAsync();
+
+        Assert.That(connections.Disposed, Is.True, "and teardown ran to the end");
+    }
+}

ClickHouse.Driver.Tcp.Tests/Client/ConnectionPoolTests.cs

@@ -36,6 +36,47 @@ private static ClickHouseTcpClientOptions Options(
             PoolReusePolicy = reusePolicy,
         };
 
+    [Test]
+    public async Task Sweep_PoolBuiltUnderAnAmbientActivity_DialsTheFloorWithoutInheritingIt()
+    {
+        // A real timer, not the controlled clock: this is about the execution context a timer captures when it is
+        // built, and ControlledTimeProvider hands back an inert timer that never fires. A pool is often built
+        // inside a traced request, and the sweep it starts outlives that request by the life of the client, so a
+        // background dial that inherited the request's span would hang every later connect off a finished trace.
+        var factory = new FakeConnectionFactory();
+        var dialed = new TaskCompletionSource<Activity>(TaskCreationOptions.RunContinuationsAsynchronously);
+        factory.BeforeCreate = _ =>
+        {
+            dialed.TrySetResult(Activity.Current);
+            return Task.CompletedTask;
+        };
+
+        ClickHouseTcpClientOptions options = Options(maxPoolSize: 2, minPoolSize: 1) with
+        {
+            SweepInterval = TimeSpan.FromMilliseconds(20),
+        };
+
+        var source = new ActivitySource("ConnectionPoolTests.Ambient");
+        using ActivityListener listener = new()
+        {
+            ShouldListenTo = candidate => candidate.Name == source.Name,
+            Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllData,
+        };
+        ActivitySource.AddActivityListener(listener);
+
+        ConnectionPool pool;
+        using (source.StartActivity("ambient"))
+        {
+            pool = new ConnectionPool(options, factory, TimeProvider.System);
+        }
+
+        await using (pool)
+        {
+            Activity ambient = await dialed.Task.WaitAsync(TimeSpan.FromSeconds(30), None);
+            Assert.That(ambient, Is.Null, "the background dial runs with no ambient span to parent its connect to");
+        }
+    }
+
     [Test]
     public async Task RentAsync_FirstRent_OpensOneConnection()
     {