Report SESSION_ID and the driver configuration in the STARTUP options - stage 1

src/Cassandra.IntegrationTests/Core/ConnectionTests.cs

@@ -654,7 +654,7 @@ public void Wrong_Ip_Init_Throws_Exception()
                             config.ServerNameResolver,
                             null),
                     config,
-                    new StartupRequestFactory(config.StartupOptionsFactory),
+                    new StartupRequestFactory(config.StartupOptionsFactory, isControlConnection: false),
                     NullConnectionObserver.Instance))
             {
                 var ex = Assert.Throws<SocketException>(() => TaskHelper.WaitToComplete(connection.Open()));
@@ -668,7 +668,7 @@ public void Wrong_Ip_Init_Throws_Exception()
                         config.ServerNameResolver,
                         null),
                     config,
-                    new StartupRequestFactory(config.StartupOptionsFactory),
+                    new StartupRequestFactory(config.StartupOptionsFactory, isControlConnection: false),
                     NullConnectionObserver.Instance))
             {
                 Assert.Throws<SocketException>(() => TaskHelper.WaitToComplete(connection.Open()));
@@ -869,7 +869,7 @@ private Connection CreateConnection(ProtocolVersion protocolVersion, Configurati
                 new SerializerManager(protocolVersion).GetCurrentSerializer(),
                 new ConnectionEndPoint(new IPEndPoint(IPAddress.Parse(contactPoint ?? _testCluster.InitialContactPoint), 9042), config.ServerNameResolver, null),
                 config,
-                new StartupRequestFactory(config.StartupOptionsFactory),
+                new StartupRequestFactory(config.StartupOptionsFactory, isControlConnection: false),
                 NullConnectionObserver.Instance);
         }
 

src/Cassandra.IntegrationTests/Core/StartupOptionsTests.cs

@@ -0,0 +1,144 @@
+//
+//      Copyright (C) ScyllaDB
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//
+
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+
+using Cassandra.IntegrationTests.SimulacronAPI.Models.Logs;
+using Cassandra.IntegrationTests.TestBase;
+using Cassandra.IntegrationTests.TestClusterManagement.Simulacron;
+using Cassandra.Requests;
+using Cassandra.Tests;
+
+using Newtonsoft.Json.Linq;
+
+using NUnit.Framework;
+using Assert = NUnit.Framework.Legacy.ClassicAssert;
+
+namespace Cassandra.IntegrationTests.Core
+{
+    /// <summary>
+    /// Verifies the SESSION_ID and DRIVER_CONFIG STARTUP options that the driver reports to the cluster
+    /// (see StartupOptionsFactory and DriverConfigReporter) over an actual connection.
+    /// </summary>
+    [TestFixture, Category(TestCategory.Short)]
+    public class StartupOptionsTests : TestGlobals
+    {
+        private SimulacronCluster _simulacronCluster;
+        private Cluster _cluster;
+        private Cluster _secondCluster;
+
+        [TearDown]
+        public void TearDown()
+        {
+            _cluster?.Dispose();
+            _cluster = null;
+            _secondCluster?.Dispose();
+            _secondCluster = null;
+            _simulacronCluster?.Dispose();
+            _simulacronCluster = null;
+        }
+
+        private Builder BuildClusterBuilder()
+        {
+            return ClusterBuilder()
+                   .AddContactPoint(_simulacronCluster.InitialContactPoint)
+                   .WithPoolingOptions(new PoolingOptions().SetCoreConnectionsPerHost(HostDistance.Local, 1));
+        }
+
+        [Test]
+        public async Task Should_ReportDriverConfig_OnlyOnTheControlConnection()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder().Build();
+
+            _cluster.Connect();
+
+            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
+            var startupMessages = startupLogs.Select(log => log.Frame.GetStartupMessage()).ToList();
+
+            Assert.GreaterOrEqual(startupMessages.Count, 2, "Expected at least the control connection and one pool connection");
+
+            var driverConfigMessages = startupMessages.Where(m => m.ContainsKey(DriverConfigReporter.DriverConfigOption)).ToList();
+            Assert.AreEqual(1, driverConfigMessages.Count, "Only the control connection should report the DRIVER_CONFIG option");
+
+            var report = JObject.Parse(driverConfigMessages.Single()[DriverConfigReporter.DriverConfigOption]);
+            Assert.AreEqual(DriverConfigReporter.SchemaVersion, report["version"].Value<int>());
+        }
+
+        [Test]
+        public async Task Should_NotReportDriverConfig_When_ReportingIsDisabled()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder().WithDriverConfigReporting(false).Build();
+
+            _cluster.Connect();
+
+            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
+            var startupMessages = startupLogs.Select(log => log.Frame.GetStartupMessage()).ToList();
+
+            Assert.GreaterOrEqual(startupMessages.Count, 2, "Expected at least the control connection and one pool connection");
+            Assert.IsTrue(
+                startupMessages.All(m => !m.ContainsKey(DriverConfigReporter.DriverConfigOption)),
+                "No connection should report the DRIVER_CONFIG option when reporting is disabled");
+        }
+
+        [Test]
+        public async Task Should_ReportTheSameSessionId_For_EveryConnectionOfTheSameCluster()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder().Build();
+
+            _cluster.Connect();
+
+            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
+            var sessionIds = startupLogs
+                              .Select(log => log.Frame.GetStartupMessage()[StartupOptionsFactory.SessionIdOption])
+                              .Distinct()
+                              .ToList();
+
+            Assert.AreEqual(1, sessionIds.Count, "Every connection of the same Cluster instance should report the same SESSION_ID");
+            Assert.IsTrue(Guid.TryParse(sessionIds.Single(), out _), "SESSION_ID should be a valid guid");
+        }
+
+        [Test]
+        public async Task Should_ReportDistinctSessionIds_For_DifferentClusterInstances()
+        {
+            _simulacronCluster = await SimulacronCluster.CreateNewAsync(1).ConfigureAwait(false);
+            _cluster = BuildClusterBuilder().Build();
+            _secondCluster = BuildClusterBuilder().Build();
+
+            _cluster.Connect();
+            _secondCluster.Connect();
+
+            var startupLogs = await _simulacronCluster.GetQueriesAsync(null, QueryType.Startup).ConfigureAwait(false);
+            var sessionIdsByClusterId = startupLogs
+                                        .Select(log => log.Frame.GetStartupMessage())
+                                        .GroupBy(m => m[StartupOptionsFactory.ClientIdOption], m => m[StartupOptionsFactory.SessionIdOption])
+                                        .ToDictionary(g => g.Key, g => g.Distinct().ToList());
+
+            Assert.AreEqual(2, sessionIdsByClusterId.Count, "Expected startup options from two distinct Cluster instances");
+            foreach (var sessionIds in sessionIdsByClusterId.Values)
+            {
+                Assert.AreEqual(1, sessionIds.Count, "Every connection of the same Cluster instance should report the same SESSION_ID");
+            }
+
+            var distinctSessionIds = sessionIdsByClusterId.Values.SelectMany(ids => ids).Distinct().ToList();
+            Assert.AreEqual(2, distinctSessionIds.Count, "Different Cluster instances should report distinct SESSION_IDs");
+        }
+    }
+}

src/Cassandra.IntegrationTests/SimulacronAPI/Models/Logs/Frame.cs

@@ -70,6 +70,11 @@ public BatchMessage GetBatchMessage()
             return GetTypedMessage<BatchMessage>();
         }
 
+        public Dictionary<string, string> GetStartupMessage()
+        {
+            return GetTypedMessage<StartupMessage>()?.Options;
+        }
+
         private T GetTypedMessage<T>()
         {
             return MessageJson == null ? default : JsonConvert.DeserializeObject<T>(MessageJson);

src/Cassandra.IntegrationTests/SimulacronAPI/Models/Logs/StartupMessage.cs

@@ -0,0 +1,27 @@
+//
+//      Copyright (C) ScyllaDB
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//
+
+using System.Collections.Generic;
+using Newtonsoft.Json;
+
+namespace Cassandra.IntegrationTests.SimulacronAPI.Models.Logs
+{
+    public class StartupMessage : BaseMessage
+    {
+        [JsonProperty("options")]
+        public Dictionary<string, string> Options { get; set; }
+    }
+}

src/Cassandra.Tests/BuilderTests.cs

@@ -311,5 +311,24 @@ public void Should_ReturnDefaultInsightsMonitoringEnabled_When_NotProvidedToBuil
             Assert.AreEqual(expected, cluster.Configuration.MonitorReportingOptions.MonitorReportingEnabled);
             Assert.AreEqual(MonitorReportingOptions.DefaultStatusEventDelayMilliseconds, cluster.Configuration.MonitorReportingOptions.StatusEventDelayMilliseconds);
         }
+
+        [Test]
+        public void Should_ReturnDefaultDriverConfigReportingEnabled_When_NotProvidedToBuilder()
+        {
+            var cluster = Cluster.Builder()
+                                 .AddContactPoint("192.168.1.10")
+                                 .Build();
+            Assert.AreEqual(Configuration.DefaultDriverConfigReportingEnabled, cluster.Configuration.DriverConfigReportingEnabled);
+        }
+
+        [Test]
+        public void Should_ReturnDriverConfigReportingDisabled_When_ProvidedToBuilder()
+        {
+            var cluster = Cluster.Builder()
+                                 .AddContactPoint("192.168.1.10")
+                                 .WithDriverConfigReporting(false)
+                                 .Build();
+            Assert.IsFalse(cluster.Configuration.DriverConfigReportingEnabled);
+        }
     }
 }

src/Cassandra.Tests/ConnectionTests.cs

@@ -47,7 +47,7 @@ private static Mock<Connection> GetConnectionMock(Configuration config = null, I
                 serializer?.GetCurrentSerializer() ?? new SerializerManager(ProtocolVersion.MaxSupported).GetCurrentSerializer(),
                 new ConnectionEndPoint(ConnectionTests.Address, config.ServerNameResolver, null),
                 config,
-                new StartupRequestFactory(config.StartupOptionsFactory),
+                new StartupRequestFactory(config.StartupOptionsFactory, isControlConnection: false),
                 NullConnectionObserver.Instance);
         }
 
@@ -352,7 +352,7 @@ private static byte[] GetResultBuffer(short streamId, ProtocolVersion version =
                 return new byte[]
                 {
                     //header
-                    header, 0, bytes[0], bytes[1], ResultResponse.OpCode, 0, 0, 0, 4, 
+                    header, 0, bytes[0], bytes[1], ResultResponse.OpCode, 0, 0, 0, 4,
                     //body
                     0, 0, 0, 1
                 };
@@ -361,7 +361,7 @@ private static byte[] GetResultBuffer(short streamId, ProtocolVersion version =
             return new byte[]
             {
                 //header
-                header, 0, (byte)streamId, ResultResponse.OpCode, 0, 0, 0, 4, 
+                header, 0, (byte)streamId, ResultResponse.OpCode, 0, 0, 0, 4,
                 //body
                 0, 0, 0, 1
             };

src/Cassandra.Tests/HostConnectionPoolTests.cs

@@ -79,7 +79,7 @@ private static IConnection CreateConnection(byte lastIpByte = 1, Configuration c
                 new SerializerManager(ProtocolVersion.MaxSupported).GetCurrentSerializer(),
                 new ConnectionEndPoint(GetIpEndPoint(lastIpByte), config.ServerNameResolver, null),
                 config,
-                new StartupRequestFactory(config.StartupOptionsFactory),
+                new StartupRequestFactory(config.StartupOptionsFactory, isControlConnection: false),
                 NullConnectionObserver.Instance);
         }
 

src/Cassandra.Tests/Requests/DriverConfigReporterTests.cs

@@ -0,0 +1,112 @@
+//
+//      Copyright (C) ScyllaDB
+//
+//   Licensed under the Apache License, Version 2.0 (the "License");
+//   you may not use this file except in compliance with the License.
+//   You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//   Unless required by applicable law or agreed to in writing, software
+//   distributed under the License is distributed on an "AS IS" BASIS,
+//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//   See the License for the specific language governing permissions and
+//   limitations under the License.
+//
+
+using System;
+using System.Collections.Generic;
+using System.Text;
+using Cassandra.Requests;
+using NUnit.Framework;
+using Assert = NUnit.Framework.Legacy.ClassicAssert;
+
+namespace Cassandra.Tests.Requests
+{
+    [TestFixture]
+    public class DriverConfigReporterTests
+    {
+        [Test]
+        public void Should_ReportSchemaVersion_When_ReportingIsEnabled()
+        {
+            var options = new Dictionary<string, string>();
+
+            new DriverConfigReporter().AddStartupOptions(options);
+
+            Assert.AreEqual("{\"version\":" + DriverConfigReporter.SchemaVersion + "}", options[DriverConfigReporter.DriverConfigOption]);
+        }
+
+        [Test]
+        public void Should_NotReportAnything_When_ReporterIsNull()
+        {
+            var factory = new StartupOptionsFactory(Guid.NewGuid(), null, null, null);
+
+            var options = factory.CreateStartupOptions(new ProtocolOptions(), null, true);
+
+            Assert.IsFalse(options.ContainsKey(DriverConfigReporter.DriverConfigOption));
+        }
+
+        [Test]
+        public void Should_ReportAConfigThatFitsInAFrame()
+        {
+            var options = new Dictionary<string, string>();
+
+            new DriverConfigReporter().AddStartupOptions(options);
+
+            // Tripwire for when actual config groups land: if the report ever grew past the limit, it would be
+            // dropped by AddStartupOptions and this Assert.IsTrue would fail with a clear message, instead of
+            // the indexer below throwing an unrelated KeyNotFoundException. Enforcement of the limit itself is
+            // covered by Should_NotReportAnything_When_ReportExceedsTheLengthLimit.
+            Assert.IsTrue(options.ContainsKey(DriverConfigReporter.DriverConfigOption), "The report was dropped, it must have exceeded the length limit.");
+
+            // The limit is enforced on the encoded length, so the assertion has to measure bytes as well.
+            Assert.LessOrEqual(
+                Encoding.UTF8.GetByteCount(options[DriverConfigReporter.DriverConfigOption]),
+                DriverConfigReporter.MaxDriverConfigLength);
+        }
+
+        [Test]
+        public void Should_NotReportAnything_When_ReportExceedsTheLengthLimit()
+        {
+            var options = new Dictionary<string, string>();
+            var oversizedReport = new string('a', DriverConfigReporter.MaxDriverConfigLength + 1);
+
+            new OversizedDriverConfigReporter(oversizedReport).AddStartupOptions(options);
+
+            Assert.IsFalse(options.ContainsKey(DriverConfigReporter.DriverConfigOption));
+        }
+
+        [Test]
+        public void Should_NotReportAnything_When_BuildingTheReportThrows()
+        {
+            var options = new Dictionary<string, string>();
+
+            new ThrowingDriverConfigReporter().AddStartupOptions(options);
+
+            Assert.IsFalse(options.ContainsKey(DriverConfigReporter.DriverConfigOption));
+        }
+
+        private class OversizedDriverConfigReporter : DriverConfigReporter
+        {
+            private readonly string _report;
+
+            public OversizedDriverConfigReporter(string report)
+            {
+                _report = report;
+            }
+
+            protected override string BuildReport()
+            {
+                return _report;
+            }
+        }
+
+        private class ThrowingDriverConfigReporter : DriverConfigReporter
+        {
+            protected override string BuildReport()
+            {
+                throw new InvalidOperationException("Simulated failure while building the report.");
+            }
+        }
+    }
+}