From b1ae364dd6260a6ef29a996b31290bc4d0196244 Mon Sep 17 00:00:00 2001 From: Alex Williams-Ferreira Date: Fri, 16 Oct 2026 16:03:28 -0700 Subject: [PATCH 1/4] Migrate Kafka workload source, tests and profile to open source Preserve the internal workload behavior and metric schema while adapting namespaces and public test helpers. Add dependency and schema coverage and public workload documentation. Package publication, live parity and coordinated internal cleanup remain release gates. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../TestResources/Results_KafkaConsumer.txt | 2 + .../TestResources/Results_KafkaProducer.txt | 5 + .../Kafka/KafkaClientProfileTests.cs | 173 ++++++ .../Kafka/KafkaProfileTests.cs | 118 ++++ .../Kafka/KafkaServerProfileTests.cs | 207 +++++++ .../Kafka/KafkaConsumerResultExample.txt | 2 + .../Kafka/KafkaProducerResultExample.txt | 5 + .../Kafka/KafkaClientExecutorTests.cs | 313 +++++++++++ .../Kafka/KafkaConsumerMetricsParserTests.cs | 67 +++ .../Kafka/KafkaExecutorTests.cs | 106 ++++ .../Kafka/KafkaProducerMetricsParserTests.cs | 68 +++ .../Kafka/KafkaServerExecutorTests.cs | 170 ++++++ .../Kafka/KafkaClientExecutor.cs | 403 ++++++++++++++ .../Kafka/KafkaConsumerMetricsParser.cs | 132 +++++ .../Kafka/KafkaExecutor.cs | 298 ++++++++++ .../Kafka/KafkaProducerMetricsParser.cs | 144 +++++ .../Kafka/KafkaServerExecutor.cs | 509 ++++++++++++++++++ .../profiles/PERF-KAFKA.json | 169 ++++++ .../docs/workloads/kafka/kafka-profiles.md | 116 ++++ website/docs/workloads/kafka/kafka.md | 133 +++++ 20 files changed, 3140 insertions(+) create mode 100644 src/VirtualClient/TestResources/Results_KafkaConsumer.txt create mode 100644 src/VirtualClient/TestResources/Results_KafkaProducer.txt create mode 100644 src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaClientProfileTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaServerProfileTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaConsumerResultExample.txt create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaProducerResultExample.txt create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaClientExecutorTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaConsumerMetricsParserTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaExecutorTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaProducerMetricsParserTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs create mode 100644 src/VirtualClient/VirtualClient.Actions/Kafka/KafkaClientExecutor.cs create mode 100644 src/VirtualClient/VirtualClient.Actions/Kafka/KafkaConsumerMetricsParser.cs create mode 100644 src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs create mode 100644 src/VirtualClient/VirtualClient.Actions/Kafka/KafkaProducerMetricsParser.cs create mode 100644 src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs create mode 100644 src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json create mode 100644 website/docs/workloads/kafka/kafka-profiles.md create mode 100644 website/docs/workloads/kafka/kafka.md diff --git a/src/VirtualClient/TestResources/Results_KafkaConsumer.txt b/src/VirtualClient/TestResources/Results_KafkaConsumer.txt new file mode 100644 index 0000000000..e5ea6578d8 --- /dev/null +++ b/src/VirtualClient/TestResources/Results_KafkaConsumer.txt @@ -0,0 +1,2 @@ +start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec +2023-10-31 14:19:58:815, 2023-10-31 14:20:05:591, 476.8396, 70.3718, 5000026, 737902.3022, 703, 6073, 78.5180, 823320.5994 \ No newline at end of file diff --git a/src/VirtualClient/TestResources/Results_KafkaProducer.txt b/src/VirtualClient/TestResources/Results_KafkaProducer.txt new file mode 100644 index 0000000000..4e38b86439 --- /dev/null +++ b/src/VirtualClient/TestResources/Results_KafkaProducer.txt @@ -0,0 +1,5 @@ +150295 records sent, 24406.5 records/sec (2.33 MB/sec), 85.4 ms avg latency, 1253.0 ms max latency. +877118 records sent, 175388.5 records/sec (16.73 MB/sec), 2425.3 ms avg latency, 3687.0 ms max latency. +1141892 records sent, 228378.4 records/sec (21.78 MB/sec), 2114.2 ms avg latency, 2572.0 ms max latency. +1689507 records sent, 337429.0 records/sec (32.18 MB/sec), 1251.2 ms avg latency, 2509.0 ms max latency. +5000000 records sent, 175358.608354 records/sec (16.72 MB/sec), 1333.88 ms avg latency, 3687.00 ms max latency, 1514 ms 50th, 3147 ms 95th, 3480 ms 99th, 3657 ms 99.9th. \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaClientProfileTests.cs b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaClientProfileTests.cs new file mode 100644 index 0000000000..c3e262077a --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaClientProfileTests.cs @@ -0,0 +1,173 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Runtime.InteropServices; + using System.Threading; + using System.Threading.Tasks; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Common; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Functional")] + public class KafkaClientProfileTests + { + private DependencyFixture mockFixture; + private string clientAgentId; + private string serverAgentId; + + public void SetupFixture(PlatformID platformID, Architecture architecture) + { + this.mockFixture = new DependencyFixture(); + this.clientAgentId = $"{Environment.MachineName}-Client"; + this.serverAgentId = $"{Environment.MachineName}-Server"; + + ComponentTypeCache.Instance.LoadComponentTypes(MockFixture.TestAssemblyDirectory); + + this.mockFixture.Setup(platformID, architecture, this.clientAgentId).SetupLayout( + new ClientInstance(this.clientAgentId, "1.2.3.4", "Client"), + new ClientInstance(this.serverAgentId, "1.2.3.5", "Server")); + + this.SetupDefaults(platformID); + } + + [Test] + [TestCase(PlatformID.Win32NT, Architecture.X64)] + [TestCase(PlatformID.Win32NT, Architecture.Arm64)] + [TestCase(PlatformID.Unix, Architecture.X64)] + [TestCase(PlatformID.Unix, Architecture.Arm64)] + public void KafkaWorkloadProfileActionsWillNotBeExecutedIfTheClientWorkloadPackageDoesNotExist(PlatformID platformID, Architecture architecture) + { + string profile = "PERF-KAFKA.json"; + this.SetupFixture(platformID, architecture); + this.mockFixture.PackageManager.Clear(); + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + DependencyException error = Assert.ThrowsAsync(() => executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None)); + Assert.AreEqual(ErrorReason.WorkloadDependencyMissing, error.Reason); + } + } + + [Test] + [TestCase(PlatformID.Win32NT, Architecture.X64)] + [TestCase(PlatformID.Win32NT, Architecture.Arm64)] + public async Task KafkaWorkloadProfileExecutesTheWorkloadAsExpectedForClientOnWindows(PlatformID platformID, Architecture architecture) + { + string profile = "PERF-KAFKA.json"; + this.SetupFixture(platformID, architecture); + List expectedCommands = new List(); + + string kafkaTopicScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "windows", "kafka-topics.bat"); + string kafkaProducerScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "windows", "kafka-producer-perf-test.bat"); + string kafkaConsumerScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "windows", "kafka-consumer-perf-test.bat"); + + // Command to create topic + expectedCommands.Add($"cmd /c {kafkaTopicScriptPath} --create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server 1.2.3.5:9092"); + // Command to produce records + expectedCommands.Add($"cmd /c {kafkaProducerScriptPath} --topic sync-test-rep-one --num-records 5000000 --record-size 100 --throughput -1 --producer-props acks=1 bootstrap.servers=1.2.3.5:9092 buffer.memory=67108864 batch.size=8196"); + // Command to consume records + expectedCommands.Add($"cmd /c {kafkaConsumerScriptPath} --topic sync-test-rep-one --messages 5000000 --bootstrap-server 1.2.3.5:9092"); + + this.mockFixture.ProcessManager.OnCreateProcess = (command, arguments, workingDir) => + { + IProcessProxy process = this.mockFixture.CreateProcess(command, arguments, workingDir); + if (arguments.Contains("--messages", StringComparison.OrdinalIgnoreCase)) + { + process.StandardOutput.Append(MockFixture.ReadTestResourcesFile("Results_KafkaConsumer.txt")); + } + else if (arguments.Contains("--producer-props", StringComparison.OrdinalIgnoreCase)) + { + process.StandardOutput.Append(MockFixture.ReadTestResourcesFile("Results_KafkaProducer.txt")); + } + + return process; + }; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None) + .ConfigureAwait(false); + WorkloadAssert.CommandsExecuted(this.mockFixture, expectedCommands.ToArray()); + } + } + + [Test] + [TestCase(PlatformID.Unix, Architecture.X64)] + [TestCase(PlatformID.Unix, Architecture.Arm64)] + public async Task KafkaWorkloadProfileExecutesTheWorkloadAsExpectedForClientOnLinux(PlatformID platformID, Architecture architecture) + { + string profile = "PERF-KAFKA.json"; + this.SetupFixture(platformID, architecture); + List expectedCommands = new List(); + + string kafkaPackagePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka"); + string jdkPackagePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "javadevelopmentkit"); + string kafkaTopicScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "kafka-topics.sh"); + string kafkaProducerScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "kafka-producer-perf-test.sh"); + string kafkaConsumerScriptPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "kafka-consumer-perf-test.sh"); + + // Command to give permissions to directory + expectedCommands.Add($"sudo chmod -R 2777 \"{kafkaPackagePath}\""); + // Command to create topic + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaTopicScriptPath} --create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server 1.2.3.5:9092"); + // Command to produce records + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaProducerScriptPath} --topic sync-test-rep-one --num-records 5000000 --record-size 100 --throughput -1 --producer-props acks=1 bootstrap.servers=1.2.3.5:9092 buffer.memory=67108864 batch.size=8196"); + // Command to consume records + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaConsumerScriptPath} --topic sync-test-rep-one --messages 5000000 --bootstrap-server 1.2.3.5:9092"); + + this.mockFixture.ProcessManager.OnCreateProcess = (command, arguments, workingDir) => + { + IProcessProxy process = this.mockFixture.CreateProcess(command, arguments, workingDir); + if (arguments.Contains("--messages", StringComparison.OrdinalIgnoreCase)) + { + process.StandardOutput.Append(MockFixture.ReadTestResourcesFile("Results_KafkaConsumer.txt")); + } + else if (arguments.Contains("--producer-props", StringComparison.OrdinalIgnoreCase)) + { + process.StandardOutput.Append(MockFixture.ReadTestResourcesFile("Results_KafkaProducer.txt")); + } + + return process; + }; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None) + .ConfigureAwait(false); + WorkloadAssert.CommandsExecuted(this.mockFixture, expectedCommands.ToArray()); + } + } + + private void SetupDefaults(PlatformID platformID) + { + Dictionary specifics = new Dictionary() + { + { PackageMetadata.ExecutablePath, "java.exe" } + }; + + if (platformID == PlatformID.Win32NT) + { + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-topics.bat"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-producer-perf-test.bat"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-consumer-perf-test.bat"); + this.mockFixture.SetupPackage("javadevelopmentkit", specifics, expectedFiles: @"runtimes/win-x64/bin/java.exe"); + } + else + { + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-topics.sh"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-producer-perf-test.sh"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-consumer-perf-test.sh"); + this.mockFixture.SetupPackage("javadevelopmentkit", specifics, expectedFiles: @"runtimes/linux-x64/bin/java"); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs new file mode 100644 index 0000000000..2020d151ac --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs @@ -0,0 +1,118 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Linq; + using System.Runtime.InteropServices; + using System.Threading; + using System.Threading.Tasks; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Contracts; + using VirtualClient.Contracts.Validation; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Functional")] + public class KafkaProfileTests + { + private DependencyFixture mockFixture; + + [OneTimeSetUp] + public void SetupFixture() + { + this.mockFixture = new DependencyFixture(); + + this.mockFixture.Setup(PlatformID.Unix, Architecture.X64, "ClientAgent").SetupLayout( + new ClientInstance("ClientAgent", "1.2.3.4", "Client"), + new ClientInstance("ServerAgent", "1.2.3.5", "Server")); + + ComponentTypeCache.Instance.LoadComponentTypes(MockFixture.TestAssemblyDirectory); + } + + [Test] + [TestCase(PlatformID.Unix, Architecture.X64)] + [TestCase(PlatformID.Unix, Architecture.Arm64)] + [TestCase(PlatformID.Win32NT, Architecture.X64)] + [TestCase(PlatformID.Win32NT, Architecture.Arm64)] + public async Task KafkaWorkloadProfileInstallsTheExpectedDependencies(PlatformID platform, Architecture architecture) + { + this.mockFixture.Setup(platform, architecture, "ClientAgent").SetupLayout( + new ClientInstance("ClientAgent", "1.2.3.4", "Client"), + new ClientInstance("ServerAgent", "1.2.3.5", "Server")); + this.mockFixture.SetupFile(this.mockFixture.Combine( + this.mockFixture.PackagesDirectory, + "microsoft-jdk-21.0.1", + PlatformSpecifics.GetPlatformArchitectureName(platform, architecture), + "bin", + platform == PlatformID.Win32NT ? "java.exe" : "java")); + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor( + "PERF-KAFKA.json", this.mockFixture.Dependencies, dependenciesOnly: true)) + { + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None); + + WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "kafka"); + WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "javadevelopmentkit"); + } + } + + [Test] + public void KafkaWorkloadProfileMeetsTheSchemaRequirements() + { + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor( + "PERF-KAFKA.json", this.mockFixture.Dependencies)) + { + ValidationResult result = SchemaRules.Instance.Validate(executor.Profile); + + Assert.IsTrue(result.IsValid, string.Join(Environment.NewLine, result.ValidationErrors)); + } + } + + [Test] + [TestCase("PERF-KAFKA.json")] + public void KafkaWorkloadProfileParametersAreInlinedCorrectly(string profile) + { + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + WorkloadAssert.ParameterReferencesInlined(executor.Profile); + } + } + + [Test] + [TestCase("PERF-KAFKA.json")] + public void KafkaWorkloadProfileUsesTheWmicFreePackageOnWindows(string profile) + { + this.mockFixture.Setup(PlatformID.Win32NT, Architecture.X64, "ClientAgent").SetupLayout( + new ClientInstance("ClientAgent", "1.2.3.4", "Client"), + new ClientInstance("ServerAgent", "1.2.3.5", "Server")); + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + ExecutionProfileElement kafkaPackage = executor.Profile.Dependencies.Single( + dependency => dependency.Parameters.TryGetValue("PackageName", out IConvertible packageName) + && packageName.ToString() == "kafka"); + + Assert.AreEqual("kafka_2.13-3.8.0-2.zip", kafkaPackage.Parameters["BlobName"].ToString()); + } + } + + [Test] + [TestCase("PERF-KAFKA.json")] + public void KafkaWorkloadProfileActionsWillNotBeExecutedIfTheDependencyPackagesDoesNotExist(string profile) + { + // We ensure the workload package does not exist. + this.mockFixture.PackageManager.Clear(); + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + DependencyException error = Assert.ThrowsAsync(() => executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None)); + Assert.AreEqual(ErrorReason.WorkloadDependencyMissing, error.Reason); + Assert.IsFalse(this.mockFixture.ProcessManager.Commands.Any()); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaServerProfileTests.cs b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaServerProfileTests.cs new file mode 100644 index 0000000000..e5870c6498 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaServerProfileTests.cs @@ -0,0 +1,207 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Net; + using System.Runtime.InteropServices; + using System.Threading; + using System.Threading.Tasks; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Common; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Functional")] + public class KafkaServerProfileTests + { + private DependencyFixture mockFixture; + + public void SetupFixture(PlatformID platformID, Architecture architecture) + { + this.mockFixture = new DependencyFixture(); + this.mockFixture + .Setup(platformID, architecture, "Server01") + .SetupLayout( + new ClientInstance("Client01", "1.2.3.4", "Client"), + new ClientInstance("Server01", "1.2.3.5", "Server")); + + ComponentTypeCache.Instance.LoadComponentTypes(MockFixture.TestAssemblyDirectory); + this.SetupDefaults(platformID); + this.mockFixture.SetupPackage("kafka", expectedFiles: "config/kraft/server.properties"); + } + + [Test] + [TestCase(PlatformID.Win32NT, Architecture.X64)] + [TestCase(PlatformID.Win32NT, Architecture.Arm64)] + public async Task KafkaWorkloadProfileInstallsTheExpectedServerDependenciesOnWindows(PlatformID platformID, Architecture architecture) + { + this.SetupFixture(platformID, architecture); + string profile = "PERF-KAFKA.json"; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None) + .ConfigureAwait(false); + + // Workload dependency package expectations + WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "kafka"); + } + } + + [Test] + [TestCase(PlatformID.Unix, Architecture.X64)] + [TestCase(PlatformID.Unix, Architecture.Arm64)] + public async Task KafkaWorkloadProfileInstallsTheExpectedServerDependenciesOnLinux(PlatformID platformID, Architecture architecture) + { + this.SetupFixture(platformID, architecture); + string profile = "PERF-KAFKA.json"; + + this.mockFixture.ProcessManager.OnCreateProcess = (command, arguments, workingDir) => + { + IProcessProxy process = this.mockFixture.CreateProcess(command, arguments, workingDir); + + if (arguments.Contains("format")) + { + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-1"); + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-2"); + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-3"); + } + + return process; + }; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None) + .ConfigureAwait(false); + + // Workload dependency package expectations + WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "kafka"); + } + } + + [Test] + [TestCase(PlatformID.Win32NT, Architecture.X64)] + [TestCase(PlatformID.Win32NT, Architecture.Arm64)] + public async Task KafkaWorkloadProfileExecutesTheWorkloadAsExpectedForServerOnWindows(PlatformID platformID, Architecture architecture) + { + string profile = "PERF-KAFKA.json"; + this.SetupFixture(platformID, architecture); + List expectedCommands = new List(); + + string kafkaStartBatFilePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "windows", "kafka-server-start.bat"); + string kafkaStorageBatFilePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "windows", "kafka-storage.bat"); + string kafkaKraftDirectoryPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "config", "kraft"); + string serveFilePath = this.mockFixture.Combine(kafkaKraftDirectoryPath, $"server-1.properties"); + + // Command to create clusterId + expectedCommands.Add($"cmd /c {kafkaStorageBatFilePath} random-uuid"); + // Command to format server instance file + expectedCommands.Add($"cmd /c {kafkaStorageBatFilePath} format -t -c {serveFilePath}"); + // Start server instance + expectedCommands.Add($"cmd /c {kafkaStartBatFilePath} {serveFilePath}"); + + // Setup the expectations for the workload + // - Workload package is installed and exists. + // - Workload binaries/executables exist on the file system. + // - Expected processes are executed. + IPAddress.TryParse("1.2.3.5", out IPAddress ipAddress); + + this.mockFixture.ProcessManager.OnCreateProcess = (command, arguments, workingDir) => + { + IProcessProxy process = this.mockFixture.CreateProcess(command, arguments, workingDir); + return process; + }; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None); + WorkloadAssert.CommandsExecuted(this.mockFixture, expectedCommands.ToArray()); + } + } + + [Test] + [TestCase(PlatformID.Unix, Architecture.X64)] + [TestCase(PlatformID.Unix, Architecture.Arm64)] + public async Task KafkaWorkloadProfileExecutesTheWorkloadAsExpectedForServerOnLinux(PlatformID platformID, Architecture architecture) + { + string profile = "PERF-KAFKA.json"; + this.SetupFixture(platformID, architecture); + List expectedCommands = new List(); + + string kafkaPackagePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka"); + string jdkPackagePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "javadevelopmentkit"); + string kafkaStartBatFilePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "kafka-server-start.sh"); + string kafkaStorageBatFilePath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "bin", "kafka-storage.sh"); + string kafkaKraftDirectoryPath = this.mockFixture.Combine(this.mockFixture.PackagesDirectory, "kafka", "config", "kraft"); + string serverFilePath = this.mockFixture.Combine(kafkaKraftDirectoryPath, $"server-1.properties"); + + // Command to give permissions to directory + expectedCommands.Add($"sudo chmod -R 2777 \"{kafkaPackagePath}\""); + // Command to create clusterId + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaStorageBatFilePath} random-uuid"); + // Command to format server instance file + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaStorageBatFilePath} format -t -c {serverFilePath}"); + // Start server instance + expectedCommands.Add($"sudo env JAVA_HOME={jdkPackagePath} bash {kafkaStartBatFilePath} {serverFilePath}"); + + // Setup the expectations for the workload + // - Workload package is installed and exists. + // - Workload binaries/executables exist on the file system. + // - Expected processes are executed. + IPAddress.TryParse("1.2.3.5", out IPAddress ipAddress); + + this.mockFixture.ProcessManager.OnCreateProcess = (command, arguments, workingDir) => + { + IProcessProxy process = this.mockFixture.CreateProcess(command, arguments, workingDir); + + if (arguments.Contains("format")) + { + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-1"); + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-2"); + this.mockFixture.SetupDirectory("/tmp/kraft-combined-logs-3"); + } + + return process; + }; + + using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies)) + { + executor.ExecuteDependencies = false; + await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None); + WorkloadAssert.CommandsExecuted(this.mockFixture, expectedCommands.ToArray()); + } + } + + private void SetupDefaults(PlatformID platformID) + { + Dictionary specifics = new Dictionary() + { + { PackageMetadata.ExecutablePath, "java.exe" } + }; + + if (platformID == PlatformID.Win32NT) + { + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-server-start.bat"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-server-stop.bat"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/windows/kafka-storage.bat"); + this.mockFixture.SetupPackage("javadevelopmentkit", specifics, expectedFiles: @"runtimes/win-x64/bin/java.exe"); + } + else + { + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/kafka-server-start.sh"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/kafka-server-stop.sh"); + this.mockFixture.SetupPackage("kafka", expectedFiles: "bin/kafka-storage.sh"); + this.mockFixture.SetupPackage("javadevelopmentkit", specifics, expectedFiles: @"runtimes/linux-x64/bin/java"); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaConsumerResultExample.txt b/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaConsumerResultExample.txt new file mode 100644 index 0000000000..e5ea6578d8 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaConsumerResultExample.txt @@ -0,0 +1,2 @@ +start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec +2023-10-31 14:19:58:815, 2023-10-31 14:20:05:591, 476.8396, 70.3718, 5000026, 737902.3022, 703, 6073, 78.5180, 823320.5994 \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaProducerResultExample.txt b/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaProducerResultExample.txt new file mode 100644 index 0000000000..4e38b86439 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Examples/Kafka/KafkaProducerResultExample.txt @@ -0,0 +1,5 @@ +150295 records sent, 24406.5 records/sec (2.33 MB/sec), 85.4 ms avg latency, 1253.0 ms max latency. +877118 records sent, 175388.5 records/sec (16.73 MB/sec), 2425.3 ms avg latency, 3687.0 ms max latency. +1141892 records sent, 228378.4 records/sec (21.78 MB/sec), 2114.2 ms avg latency, 2572.0 ms max latency. +1689507 records sent, 337429.0 records/sec (32.18 MB/sec), 1251.2 ms avg latency, 2509.0 ms max latency. +5000000 records sent, 175358.608354 records/sec (16.72 MB/sec), 1333.88 ms avg latency, 3687.00 ms max latency, 1514 ms 50th, 3147 ms 95th, 3480 ms 99th, 3657 ms 99.9th. \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaClientExecutorTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaClientExecutorTests.cs new file mode 100644 index 0000000000..497f7a9615 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaClientExecutorTests.cs @@ -0,0 +1,313 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Net.Http; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Moq; + using NUnit.Framework; + using Polly; + using VirtualClient; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Unit")] + public class KafkaClientExecutorTests + { + private MockFixture fixture; + private DependencyPath mockKafkaPackage; + private DependencyPath mockJdkPackage; + private string consumerResults; + private string producerResults; + + public void SetupTests(PlatformID platformID) + { + this.fixture = new MockFixture(); + this.fixture.Setup(platformID); + this.mockKafkaPackage = new DependencyPath("kafka", this.fixture.GetPackagePath("kafka")); + this.mockJdkPackage = new DependencyPath("javadevelopmentkit", this.fixture.GetPackagePath("javadevelopmentkit")); + + this.fixture.Parameters = new Dictionary() + { + ["Scenario"] = "Kafka_Client_Scenario", + ["Port"] = 9092, + ["PackageName"] = this.mockKafkaPackage.Name, + ["CommandLine"] = "--create --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server {0}:{Port}", + ["CommandType"] = KafkaCommandType.CreateTopic + }; + + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(this.mockKafkaPackage); + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(this.mockJdkPackage.Name, It.IsAny())) + .ReturnsAsync(this.mockJdkPackage); + + this.fixture.FileSystem.Setup(fe => fe.File.Exists(It.IsAny())).Returns(true); + this.fixture.FileSystem.Setup(fe => fe.Directory.Exists(It.IsAny())).Returns(true); + this.consumerResults = MockFixture.ReadTestExamplesFile("Kafka", "KafkaConsumerResultExample.txt"); + this.producerResults = MockFixture.ReadTestExamplesFile("Kafka", "KafkaProducerResultExample.txt"); + + this.fixture.ApiClient.Setup(client => client.GetHeartbeatAsync(It.IsAny(), It.IsAny>())) + .ReturnsAsync(this.fixture.CreateHttpResponse(System.Net.HttpStatusCode.OK)); + + this.fixture.ApiClient.Setup(client => client.GetServerOnlineStatusAsync(It.IsAny(), It.IsAny>())) + .ReturnsAsync(this.fixture.CreateHttpResponse(System.Net.HttpStatusCode.OK)); + } + + [Test] + [TestCase(PlatformID.Win32NT)] + [TestCase(PlatformID.Unix)] + public async Task KafkaClientExecutorConfirmsTheExpectedPackagesOnInitialization(PlatformID platformID) + { + this.SetupTests(platformID); + using (var component = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + await component.InitializeAsync(EventContext.None, CancellationToken.None); + this.fixture.PackageManager.Verify(mgr => mgr.GetPackageAsync(this.mockKafkaPackage.Name, It.IsAny())); + } + } + + [Test] + [TestCase(PlatformID.Win32NT, KafkaCommandType.CreateTopic, "kafka-topics.bat")] + [TestCase(PlatformID.Win32NT, KafkaCommandType.ProducerTest, "kafka-producer-perf-test.bat")] + [TestCase(PlatformID.Win32NT, KafkaCommandType.ConsumerTest, "kafka-consumer-perf-test.bat")] + [TestCase(PlatformID.Unix, KafkaCommandType.CreateTopic, "kafka-topics.sh")] + [TestCase(PlatformID.Unix, KafkaCommandType.ProducerTest, "kafka-producer-perf-test.sh")] + [TestCase(PlatformID.Unix, KafkaCommandType.ConsumerTest, "kafka-consumer-perf-test.sh")] + public async Task KafkaClientExecutorConfirmsTheExpectedKafkaStartScript(PlatformID platformID, KafkaCommandType kafkaCommandType, string commandScript) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = kafkaCommandType.ToString(); + using (var component = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + Assert.IsEmpty(component.TestKafkaCommandScriptPath); + await component.InitializeAsync(EventContext.None, CancellationToken.None); + Assert.IsTrue(component.TestKafkaCommandScriptPath.Contains(commandScript)); + } + } + + [Test] + [TestCase(PlatformID.Win32NT)] + public async Task KafkaClientExecutorExecutesWorkloadForSetupOnWindows(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.CreateTopic; + this.fixture.Parameters["CommandLine"] = "--create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server {0}:{Port}"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "windows", "kafka-topics.bat"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + List expectedCommands = new List() + { + // Command to create topic + $"cmd /c {KafkaCommandScriptPath} --create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server 1.2.3.5:9092" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Unix)] + public async Task KafkaClientExecutorExecutesWorkloadForSetupOnLinux(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.CreateTopic; + this.fixture.Parameters["CommandLine"] = "--create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server {0}:{Port}"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaPackagePath = this.mockKafkaPackage.Path; + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "kafka-topics.sh"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + List expectedCommands = new List() + { + // Command to create topic + $"sudo chmod -R 2777 \"{KafkaPackagePath}\"", + $"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {KafkaCommandScriptPath} --create --if-not-exists --topic sync-test-rep-one --partitions 6 --replication-factor 1 --bootstrap-server 1.2.3.5:9092" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Win32NT)] + public async Task KafkaClientExecutorExecutesWorkloadForProducerOnWindows(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.ProducerTest; + this.fixture.Parameters["RecordSize"] = 100; + this.fixture.Parameters["RecordCount"] = 5000000; + this.fixture.Parameters["CommandLine"] = "--topic sync-test-rep-one --num-records {RecordCount} --record-size {RecordSize} --throughput -1 --producer-props acks=1 bootstrap.servers={0}:{Port} buffer.memory=67108864 batch.size=8196"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "windows", "kafka-producer-perf-test.bat"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + + List expectedCommands = new List() + { + // Command to produce records + $"cmd /c {KafkaCommandScriptPath} --topic sync-test-rep-one --num-records 5000000 --record-size 100 --throughput -1 --producer-props acks=1 bootstrap.servers=1.2.3.5:9092 buffer.memory=67108864 batch.size=8196" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + this.fixture.Process.StandardOutput.Append(this.producerResults); + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Unix)] + public async Task KafkaClientExecutorExecutesWorkloadForProducerOnLinux(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.ProducerTest; + this.fixture.Parameters["RecordSize"] = 100; + this.fixture.Parameters["RecordCount"] = 5000000; + this.fixture.Parameters["CommandLine"] = "--topic sync-test-rep-one --num-records {RecordCount} --record-size {RecordSize} --throughput -1 --producer-props acks=1 bootstrap.servers={0}:{Port} buffer.memory=67108864 batch.size=8196"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "kafka-producer-perf-test.sh"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + + List expectedCommands = new List() + { + // Command to produce records + $"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {KafkaCommandScriptPath} --topic sync-test-rep-one --num-records 5000000 --record-size 100 --throughput -1 --producer-props acks=1 bootstrap.servers=1.2.3.5:9092 buffer.memory=67108864 batch.size=8196" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + this.fixture.Process.StandardOutput.Append(this.producerResults); + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Win32NT)] + public async Task KafkaClientExecutorExecutesWorkloadForConsumerOnWindows(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.ConsumerTest; + this.fixture.Parameters["Messages"] = 5000000; + this.fixture.Parameters["CommandLine"] = "--topic sync-test-rep-one --messages {Messages} --bootstrap-server {0}:{Port}"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "windows", "kafka-consumer-perf-test.bat"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + + List expectedCommands = new List() + { + // Command to consume records + $"cmd /c {KafkaCommandScriptPath} --topic sync-test-rep-one --messages 5000000 --bootstrap-server 1.2.3.5:9092" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + this.fixture.Process.StandardOutput.Append(this.consumerResults); + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Unix)] + public async Task KafkaClientExecutorExecutesWorkloadForConsumerOnLinux(PlatformID platformID) + { + this.SetupTests(platformID); + this.fixture.Parameters["CommandType"] = KafkaCommandType.ConsumerTest; + this.fixture.Parameters["Messages"] = 5000000; + this.fixture.Parameters["CommandLine"] = "--topic sync-test-rep-one --messages {Messages} --bootstrap-server {0}:{Port}"; + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + + string KafkaCommandScriptPath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "kafka-consumer-perf-test.sh"); + + using (var executor = new TestKafkaClientExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + + List expectedCommands = new List() + { + // Command to consume records + $"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {KafkaCommandScriptPath} --topic sync-test-rep-one --messages 5000000 --bootstrap-server 1.2.3.5:9092" + }; + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + this.fixture.Process.StandardOutput.Append(this.consumerResults); + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + private class TestKafkaClientExecutor : KafkaClientExecutor + { + public string TestKafkaCommandScriptPath + { + get + { + return this.KafkaCommandScriptPath ?? string.Empty; + } + set { } + } + + public TestKafkaClientExecutor(IServiceCollection services, IDictionary parameters = null) + : base(services, parameters) + { + } + + public new Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + return base.InitializeAsync(telemetryContext, cancellationToken); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaConsumerMetricsParserTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaConsumerMetricsParserTests.cs new file mode 100644 index 0000000000..3d861a6976 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaConsumerMetricsParserTests.cs @@ -0,0 +1,67 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System.Collections.Generic; + using System.Linq; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Unit")] + public class KafkaConsumerParserUnitTests + { + private string rawText; + private KafkaConsumerMetricsParser testParser; + + [SetUp] + public void Setup() + { + this.rawText = MockFixture.ReadTestExamplesFile("Kafka", "KafkaConsumerResultExample.txt"); + this.testParser = new KafkaConsumerMetricsParser(this.rawText); + } + + [Test] + public void KafkaConsumerMetricsParserParsesTheExpectedMetricsFromResultsCorrectly() + { + IList metrics = this.testParser.Parse(); + + Assert.AreEqual(7, metrics.Count); + + MetricAssert.Exists(metrics, "data_consumed_in_megabyte", 476.8396, MetricUnit.Megabytes); + MetricAssert.Exists(metrics, "megabyte_per_sec_throughput", 70.3718, MetricUnit.MegabytesPerSecond); + MetricAssert.Exists(metrics, "data_consumed_in_nMsg", 5000026, MetricUnit.Operations); + MetricAssert.Exists(metrics, "nMsg_per_sec_throughput", 737902.3022, MetricUnit.OperationsPerSec); + MetricAssert.Exists(metrics, "fetch_time_in_millisec", 6073, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "fetch_megabyte_per_sec", 78.5180, MetricUnit.MegabytesPerSecond); + MetricAssert.Exists(metrics, "fetch_nMsg_per_sec", 823320.5994, MetricUnit.OperationsPerSec); + } + + [Test] + public void KafkaConsumerMetricsParserAssociatesTheCorrectRelativityToTheMetrics() + { + IList metrics = this.testParser.Parse(); + + if (metrics.Count != 7) + { + Assert.Inconclusive(); + } + + Assert.IsTrue(metrics.Where(m => m.Name.EndsWith("data_consumed_in_megabyte")).All(m => m.Relativity == MetricRelativity.HigherIsBetter)); + Assert.IsTrue(metrics.Where(m => m.Name.Contains("fetch_time_in_millisec")).All(m => m.Relativity == MetricRelativity.LowerIsBetter)); + } + + [Test] + public void KafkaConsumerMetricsParserThrowIfInvalidOutputFormat() + { + this.rawText = MockFixture.ReadTestExamplesFile("Kafka", "KafkaConsumerResultExample.txt").Substring(0, 10); + this.testParser = new KafkaConsumerMetricsParser(this.rawText); + + SchemaException exception = Assert.Throws(() => this.testParser.Parse()); + Assert.IsTrue(exception.Message.StartsWith("Invalid/unpexpected format.")); + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaExecutorTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaExecutorTests.cs new file mode 100644 index 0000000000..62567b3249 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaExecutorTests.cs @@ -0,0 +1,106 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Moq; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Unit")] + public class KafkaExecutorTests + { + private MockFixture fixture; + private DependencyPath mockKafkaPackage; + private DependencyPath mockJdkPackage; + + public void SetupDefaults(PlatformID platformID) + { + this.fixture = new MockFixture(); + this.fixture.Setup(platformID); + this.mockKafkaPackage = new DependencyPath("kafka", this.fixture.GetPackagePath("kafka")); + this.mockJdkPackage = new DependencyPath("javadevelopmentkit", this.fixture.GetPackagePath("javadevelopmentkit")); + + this.fixture.Parameters = new Dictionary + { + ["Scenario"] = "Kafka_Scenario", + ["PackageName"] = this.mockKafkaPackage.Name, + ["ServerInstances"] = 1, + ["Port"] = 9092 + }; + + this.fixture.FileSystem.Setup(fe => fe.Directory.Exists(It.IsAny())).Returns(true); + this.fixture.ProcessManager.OnCreateProcess = (cmd, args, wd) => this.fixture.Process; + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(this.mockKafkaPackage.Name, It.IsAny())) + .ReturnsAsync(this.mockKafkaPackage); + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(this.mockJdkPackage.Name, It.IsAny())) + .ReturnsAsync(this.mockJdkPackage); + } + + [Test] + [TestCase(PlatformID.Win32NT, "cmd")] + [TestCase(PlatformID.Unix, "bash")] + public async Task KafkaExecutorThrowsOnUnsupportedDistroAsync(PlatformID platformID, string command) + { + this.SetupDefaults(platformID); + + using (var executor = new TestKafkaExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + await executor.InitializeAsync(EventContext.None, CancellationToken.None); + Assert.IsTrue(executor.TestPlatformSpecificCommandType.Contains(command)); + } + } + + [Test] + public void KafkaExecutorValidationIdentifiesTheMaxClientInstanceCountParameter() + { + this.SetupDefaults(PlatformID.Unix); + this.fixture.Parameters["Partitions"] = 1; + this.fixture.Parameters["MaxClientInstances"] = 2; + + using (var executor = new TestKafkaExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + ArgumentException exception = Assert.Throws(() => executor.ValidateParameters()); + + StringAssert.Contains(nameof(executor.MaxClientInstanceCount), exception.Message); + StringAssert.Contains(nameof(executor.Partitions), exception.Message); + } + } + + private class TestKafkaExecutor : KafkaExecutor + { + public TestKafkaExecutor(IServiceCollection dependencies, IDictionary parameters = null) + : base(dependencies, parameters) + { + } + + public string TestPlatformSpecificCommandType + { + get + { + return this.PlatformSpecificCommandType ?? string.Empty; + } + set { } + } + + public new Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + return base.InitializeAsync(telemetryContext, cancellationToken); + } + + public void ValidateParameters() + { + base.Validate(); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaProducerMetricsParserTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaProducerMetricsParserTests.cs new file mode 100644 index 0000000000..55a7c9b13e --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaProducerMetricsParserTests.cs @@ -0,0 +1,68 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System.Collections.Generic; + using System.Linq; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Unit")] + public class KafkaProducerParserUnitTests + { + private string rawText; + private KafkaProducerMetricsParser testParser; + + [SetUp] + public void Setup() + { + this.rawText = MockFixture.ReadTestExamplesFile("Kafka", "KafkaProducerResultExample.txt"); + this.testParser = new KafkaProducerMetricsParser(this.rawText); + } + + [Test] + public void KafkaProducerMetricsParserParsesTheExpectedMetricsFromResultsCorrectly() + { + IList metrics = this.testParser.Parse(); + + Assert.AreEqual(8, metrics.Count); + + MetricAssert.Exists(metrics, "Total_Records_Sent", 5000000, MetricUnit.Operations); + MetricAssert.Exists(metrics, "Records_Per_Sec", 175358.608354, MetricUnit.OperationsPerSec); + MetricAssert.Exists(metrics, "Latency-Avg", 1333.88, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "Latency-Max", 3687.00, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "Latency-P50", 1514, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "Latency-P95", 3147, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "Latency-P99", 3480, MetricUnit.Milliseconds); + MetricAssert.Exists(metrics, "Latency-P99.9", 3657, MetricUnit.Milliseconds); + } + + [Test] + public void KafkaProducerMetricsParserAssociatesTheCorrectRelativityToTheMetrics() + { + IList metrics = this.testParser.Parse(); + + if (metrics.Count != 8) + { + Assert.Inconclusive(); + } + + Assert.IsTrue(metrics.Where(m => m.Name.EndsWith("Records_Per_Sec")).All(m => m.Relativity == MetricRelativity.HigherIsBetter)); + Assert.IsTrue(metrics.Where(m => m.Name.Contains("Latency-Avg")).All(m => m.Relativity == MetricRelativity.LowerIsBetter)); + } + + [Test] + public void KafkaProducerMetricsParserThrowIfInvalidOutputFormat() + { + this.rawText = MockFixture.ReadTestExamplesFile("Kafka", "KafkaProducerResultExample.txt").Substring(0, 10); + this.testParser = new KafkaProducerMetricsParser(this.rawText); + + SchemaException exception = Assert.Throws(() => this.testParser.Parse()); + Assert.IsTrue(exception.Message.StartsWith("Invalid/unpexpected format.")); + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs new file mode 100644 index 0000000000..667bd0dce1 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs @@ -0,0 +1,170 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Moq; + using NUnit.Framework; + using VirtualClient; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + using VirtualClient.TestExtensions; + + [TestFixture] + [Category("Unit")] + public class KafkaServerExecutorTests + { + private MockFixture fixture; + private DependencyPath mockKafkaPackage; + private DependencyPath mockJdkPackage; + + public void SetupTests(PlatformID platformID) + { + this.fixture = new MockFixture(); + this.fixture.Setup(platformID); + this.mockKafkaPackage = new DependencyPath("kafka", this.fixture.GetPackagePath("kafka")); + this.mockJdkPackage = new DependencyPath("javadevelopmentkit", this.fixture.GetPackagePath("javadevelopmentkit")); + + this.fixture.Parameters = new Dictionary() + { + ["Scenario"] = "Kafka_Scenario", + ["PackageName"] = this.mockKafkaPackage.Name, + ["ServerInstances"] = 1, + ["Port"] = 9092 + }; + + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(this.mockKafkaPackage.Name, It.IsAny())) + .ReturnsAsync(this.mockKafkaPackage); + this.fixture.PackageManager.Setup(mgr => mgr.GetPackageAsync(this.mockJdkPackage.Name, It.IsAny())) + .ReturnsAsync(this.mockJdkPackage); + this.fixture.FileSystem.Setup(fe => fe.File.Exists(It.IsAny())).Returns(true); + this.fixture.FileSystem.Setup(fe => fe.Directory.Exists(It.IsAny())).Returns(true); + } + + [Test] + [TestCase(PlatformID.Win32NT, 1)] + [TestCase(PlatformID.Win32NT, 2)] + [TestCase(PlatformID.Win32NT, 3)] + public async Task KafkaServerExecutorExecutesWorkloadAsExpectedForMultipleInstancesOnWindows(PlatformID platformID, int serverInstanceCount) + { + this.SetupTests(platformID); + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + this.fixture.FileSystem.Setup(fe => fe.File.ReadAllTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("node.id"); + this.fixture.FileSystem.Setup(fe => fe.File.WriteAllTextAsync(It.IsAny(), It.IsAny(), It.IsAny())); + + string kafkaStartBatFilePath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "windows", "kafka-server-start.bat"); + string kafkaStorageBatFilePath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "windows", "kafka-storage.bat"); + string kafkaKraftDirectoryPath = this.fixture.Combine(this.mockKafkaPackage.Path, "config", "kraft"); + + using (var executor = new TestKafkaServerExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + executor.Parameters[nameof(executor.ServerInstanceCount)] = serverInstanceCount; + + List expectedCommands = new List() + { + // Command to create clusterId + $"cmd /c {kafkaStorageBatFilePath} random-uuid" + }; + + for (int server = 0; server < serverInstanceCount; server++) + { + string serverFilePath = this.fixture.Combine(kafkaKraftDirectoryPath, $"server-{server+1}.properties"); + // Command to format server instance file + expectedCommands.Add($"cmd /c {kafkaStorageBatFilePath} format -t -c {serverFilePath}"); + // Start server instance + expectedCommands.Add($"cmd /c {kafkaStartBatFilePath} {serverFilePath}"); + } + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + [Test] + [TestCase(PlatformID.Unix, 1)] + [TestCase(PlatformID.Unix, 2)] + [TestCase(PlatformID.Unix, 3)] + public async Task KafkaServerExecutorExecutesWorkloadAsExpectedForMultipleInstancesOnLinux(PlatformID platformID, int serverInstanceCount) + { + this.SetupTests(platformID); + this.fixture.FileSystem.Setup(fe => fe.Directory.Delete(It.IsAny())); + this.fixture.FileSystem.Setup(fe => fe.File.ReadAllTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("node.id"); + this.fixture.FileSystem.Setup(fe => fe.File.WriteAllTextAsync(It.IsAny(), It.IsAny(), It.IsAny())); + + string kafkaStartBatFilePath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "kafka-server-start.sh"); + string kafkaStorageBatFilePath = this.fixture.Combine(this.mockKafkaPackage.Path, "bin", "kafka-storage.sh"); + string kafkaKraftDirectoryPath = this.fixture.Combine(this.mockKafkaPackage.Path, "config", "kraft"); + + using (var executor = new TestKafkaServerExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + executor.Parameters[nameof(executor.ServerInstanceCount)] = serverInstanceCount; + + List expectedCommands = new List() + { + // Command to create clusterId + $"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {kafkaStorageBatFilePath} random-uuid" + }; + + for (int server = 0; server < serverInstanceCount; server++) + { + string serverFilePath = this.fixture.Combine(kafkaKraftDirectoryPath, $"server-{server + 1}.properties"); + // Command to format server instance file + expectedCommands.Add($"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {kafkaStorageBatFilePath} format -t -c {serverFilePath}"); + // Start server instance + expectedCommands.Add($"sudo env JAVA_HOME={this.mockJdkPackage.Path} bash {kafkaStartBatFilePath} {serverFilePath}"); + } + + this.fixture.ProcessManager.OnCreateProcess = (exe, arguments, workingDirectory) => + { + expectedCommands.Remove($"{exe} {arguments}"); + return this.fixture.Process; + }; + + await executor.ExecuteAsync(CancellationToken.None); + Assert.IsEmpty(expectedCommands); + } + } + + private class TestKafkaServerExecutor : KafkaServerExecutor + { + public string TestKafkaStartScriptPath + { + get + { + return this.KafkaStartScriptPath ?? string.Empty; + } + set { } + } + public string TestKafkaPackagePath + { + get + { + return this.KafkaPackagePath ?? string.Empty; + } + set { } + } + public TestKafkaServerExecutor(IServiceCollection services, IDictionary parameters = null) + : base(services, parameters) + { + } + + public new Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + return base.InitializeAsync(telemetryContext, cancellationToken); + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaClientExecutor.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaClientExecutor.cs new file mode 100644 index 0000000000..17db36ef79 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaClientExecutor.cs @@ -0,0 +1,403 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Net; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Microsoft.Extensions.Logging; + using Polly; + using VirtualClient; + using VirtualClient.Common; + using VirtualClient.Common.Extensions; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + using VirtualClient.Contracts.Metadata; + using VirtualClient.Logging; + + /// + /// Kafka Command Type. + /// + public enum KafkaCommandType + { + /// + /// Create kafka topics. + /// + CreateTopic, + + /// + /// Run producer script. + /// + ProducerTest, + + /// + /// Run consumer script. + /// + ConsumerTest + } + + /// + /// Kafka client executor. + /// + [SupportedPlatforms("linux-arm64,linux-x64,win-arm64,win-x64")] + public class KafkaClientExecutor : KafkaExecutor + { + private readonly object lockObject = new object(); + + /// + /// Initializes a new instance of the class. + /// + /// Provides all of the required dependencies to the Virtual Client component. + /// An enumeration of key-value pairs that can control the execution of the component./param> + public KafkaClientExecutor(IServiceCollection dependencies, IDictionary parameters = null) + : base(dependencies, parameters) + { + this.ClientFlowRetryPolicy = Policy.Handle(exc => !(exc is OperationCanceledException)) + .WaitAndRetryAsync(3, (retries) => TimeSpan.FromSeconds(retries * 2)); + + this.ClientRetryPolicy = Policy.Handle(exc => !(exc is OperationCanceledException)) + .WaitAndRetryAsync(3, (retries) => TimeSpan.FromSeconds(retries)); + + this.PollingTimeout = TimeSpan.FromMinutes(40); + } + + /// + /// Parameter defines the kafka command line to execute. + /// + public string CommandLine { get; set; } + + /// + /// Parameter defines true/false whether the action is meant to warm up the server. + /// + public bool WarmUp + { + get + { + return this.Parameters.GetValue(nameof(this.WarmUp), false); + } + } + + /// + /// Parameter defines the number of client instances/copies to run. + /// + public int ClientInstanceCount + { + get + { + return this.Parameters.GetValue(new string[] { nameof(this.ClientInstanceCount), "ClientInstances" }, 3); + } + } + + /// + /// The retry policy to apply to the client-side execution workflow. + /// + protected IAsyncPolicy ClientFlowRetryPolicy { get; set; } + + /// + /// The retry policy to apply to each Kafka benchmark instance when trying to startup + /// against a target server. + /// + protected IAsyncPolicy ClientRetryPolicy { get; set; } + + /// + /// The timespan at which the client will poll the server for responses before + /// timing out. + /// + protected TimeSpan PollingTimeout { get; set; } + + /// + /// Path to Kafka topic executable. + /// + protected string KafkaTopicScriptPath { get; set; } + + /// + /// Path to Kafka producer performance test executable. + /// + protected string KafkProducerPerfScriptPath { get; set; } + + /// + /// Path to Kafka consumer performance test executable. + /// + protected string KafkaConsumerPerfScriptPath { get; set; } + + /// + /// Path to Kafka command executable. + /// + protected string KafkaCommandScriptPath { get; set; } + + /// + /// True/false whether the Producer test has ran once + /// + protected bool IsServerWarmedUp { get; set; } + + /// + /// Parameter defines the kafka command type to decide which exe to run. + /// + protected KafkaCommandType CommandType + { + get + { + return this.Parameters.GetEnumValue(nameof(this.CommandType)); + } + } + + /// + /// Validates the component definition for requirements. + /// + protected override void Validate() + { + base.Validate(); + if (!Enum.IsDefined(typeof(KafkaCommandType), this.CommandType)) + { + throw new ArgumentException($"Parameter CommandType should be one of ${KafkaCommandType.CreateTopic}, ${KafkaCommandType.ProducerTest} or ${KafkaCommandType.ConsumerTest}"); + } + } + + /// + /// Initializes the environment and dependencies for client of kafka Benchmark workload. + /// + /// Provides context information that will be captured with telemetry events. + /// A token that can be used to cancel the operation. + /// + protected override async Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + await base.InitializeAsync(telemetryContext, cancellationToken).ConfigureAwait(false); + + switch (this.Platform) + { + case PlatformID.Win32NT: + this.KafkaTopicScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-topics.bat"); + this.KafkProducerPerfScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-producer-perf-test.bat"); + this.KafkaConsumerPerfScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-consumer-perf-test.bat"); + break; + + case PlatformID.Unix: + this.KafkaTopicScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-topics.sh"); + this.KafkProducerPerfScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-producer-perf-test.sh"); + this.KafkaConsumerPerfScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-consumer-perf-test.sh"); + break; + } + + this.CommandLine = this.Parameters.GetValue(nameof(this.CommandLine)); + + switch (this.CommandType) + { + case KafkaCommandType.CreateTopic: + this.KafkaCommandScriptPath = this.KafkaTopicScriptPath; + break; + case KafkaCommandType.ProducerTest: + this.KafkaCommandScriptPath = this.KafkProducerPerfScriptPath; + break; + case KafkaCommandType.ConsumerTest: + this.KafkaCommandScriptPath = this.KafkaConsumerPerfScriptPath; + break; + } + + switch (this.Platform) + { + case PlatformID.Win32NT: + await this.SystemManagement.MakeFileExecutableAsync(this.KafkaCommandScriptPath, this.Platform, cancellationToken); + break; + + case PlatformID.Unix: + await this.SystemManagement.MakeFilesExecutableAsync(this.KafkaPackagePath, this.Platform, cancellationToken); + break; + } + } + + /// + /// Executes the workload. + /// + /// Provides context information that will be captured with telemetry events. + /// A token that can be used to cancel the operation. + protected override async Task ExecuteAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + if (!this.WarmUp || !this.IsServerWarmedUp) + { + IPAddress ipAddress; + List clientWorkloadTasks = new List(); + int replicationFactor = (this.Platform == PlatformID.Win32NT || this.ServerInstanceCount <= 3) ? 1 : (this.ServerInstanceCount > 5 ? 4 : this.ServerInstanceCount - 2); + this.CommandLine = this.CommandLine.Replace("$ReplicationFactor", replicationFactor.ToString()); + + if (this.IsMultiRoleLayout()) + { + IEnumerable targetServers = this.GetLayoutClientInstances(ClientRole.Server); + + foreach (ClientInstance server in targetServers) + { + this.CommandLine = string.Format(this.CommandLine, server.IPAddress); + clientWorkloadTasks.Add(this.ClientFlowRetryPolicy.ExecuteAsync(async () => + { + if (!cancellationToken.IsCancellationRequested) + { + IApiClient serverApiClient = this.ApiClientManager.GetOrCreateApiClient(server.Name, server); + // 1) Confirm server is online. + // =========================================================================== + this.Logger.LogTraceMessage("Synchronization: Poll server API for heartbeat..."); + await serverApiClient.PollForHeartbeatAsync(this.PollingTimeout, cancellationToken) + .ConfigureAwait(false); + + // 2) Confirm the server-side application (e.g. web server) is online. + // =========================================================================== + this.Logger.LogTraceMessage("Synchronization: Poll server for online signal..."); + await serverApiClient.PollForServerOnlineAsync(TimeSpan.FromMinutes(10), cancellationToken) + .ConfigureAwait(false); + + this.Logger.LogTraceMessage("Synchronization: Server online signal confirmed..."); + this.Logger.LogTraceMessage("Synchronization: Start client workload..."); + ConsoleLogger.Default.LogMessage("StartingClientWorkload...", telemetryContext); + + // 3) Execute the client workload. + // =========================================================================== + ipAddress = IPAddress.Parse(server.IPAddress); + await this.ExecuteWorkloadsAsync(ipAddress, telemetryContext, cancellationToken) + .ConfigureAwait(false); + } + })); + } + } + else + { + ipAddress = IPAddress.Loopback; + this.CommandLine = string.Format(this.CommandLine, "localhost"); + clientWorkloadTasks.Add(this.ClientFlowRetryPolicy.ExecuteAsync(async () => + { + if (!cancellationToken.IsCancellationRequested) + { + await this.ExecuteWorkloadsAsync(ipAddress, telemetryContext, cancellationToken).ConfigureAwait(false); + } + })); + } + + await Task.WhenAll(clientWorkloadTasks); + + if (this.WarmUp) + { + this.IsServerWarmedUp = true; + } + } + } + + private Task ExecuteWorkloadsAsync(IPAddress serverIPAddress, EventContext telemetryContext, CancellationToken cancellationToken) + { + EventContext relatedContext = telemetryContext.Clone() + .AddContext("serverIPAddress", serverIPAddress.ToString()); + + return this.Logger.LogMessageAsync($"{this.TypeName}.ExecuteWorkloads", relatedContext.Clone(), async () => + { + using (BackgroundOperations profiling = BackgroundOperations.BeginProfiling(this, cancellationToken)) + { + List commands = new List(); + relatedContext.AddContext("command", this.PlatformSpecificCommandType); + relatedContext.AddContext("commandArguments", commands); + List workloadProcesses = new List(); + for (int i = 0; i < this.ClientInstanceCount; i++) + { + string commandArguments = this.GetPlatformFormattedCommandArguement(this.KafkaCommandScriptPath, this.CommandLine); + commands.Add($"{this.PlatformSpecificCommandType} {commandArguments}"); + workloadProcesses.Add(this.ExecuteWorkloadAsync(this.PlatformSpecificCommandType, commandArguments, this.KafkaPackagePath, relatedContext, cancellationToken)); + } + + await Task.WhenAll(workloadProcesses); + } + }); + } + + private async Task ExecuteWorkloadAsync(string command, string commandArguments, string workingDirectory, EventContext telemetryContext, CancellationToken cancellationToken) + { + try + { + await (this.ClientRetryPolicy ?? Policy.NoOpAsync()).ExecuteAsync(async () => + { + try + { + DateTime startTime = DateTime.UtcNow; + using (IProcessProxy process = await this.RunProcessAsync(command, commandArguments, workingDirectory, telemetryContext, cancellationToken, runElevated: true)) + { + if (!cancellationToken.IsCancellationRequested) + { + await this.LogProcessDetailsAsync(process, telemetryContext, "Kafka", logToFile: true); + process.ThrowIfWorkloadFailed(); + if (this.CommandType == KafkaCommandType.ConsumerTest) + { + string output = process.StandardOutput.ToString(); + telemetryContext.AddContext(this.CommandType.ToString(), output); + this.CaptureMetrics(output, process.FullCommand(), startTime, DateTime.UtcNow, telemetryContext, cancellationToken); + } + } + } + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.WorkloadStartError", + LogLevel.Warning, + telemetryContext.Clone().AddError(exc)); + throw; + } + }); + } + catch (OperationCanceledException) + { + // Expected whenever certain operations (e.g. Task.Delay) are cancelled. + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.ExecuteWorkloadError", + LogLevel.Error, + telemetryContext.Clone().AddError(exc)); + + throw; + } + } + + private void CaptureMetrics(string output, string commandArguments, DateTime startTime, DateTime endTime, EventContext telemetryContext, CancellationToken cancellationToken) + { + if (!cancellationToken.IsCancellationRequested) + { + try + { + this.MetadataContract.AddForScenario( + "Kafka-Benchmark", + commandArguments, + toolVersion: null); + + this.MetadataContract.Apply(telemetryContext); + + // The Kafka workloads run multi-threaded. The lock is meant to ensure we do not have + // race conditions that affect the parsing of the results. + lock (this.lockObject) + { + MetricsParser kafkaMetricsParser = new KafkaProducerMetricsParser(output); + if (this.CommandType == KafkaCommandType.ConsumerTest) + { + kafkaMetricsParser = new KafkaConsumerMetricsParser(output); + } + + IList workloadMetrics = kafkaMetricsParser.Parse(); + + this.Logger.LogMetrics( + "Kafka-Benchmark", + scenarioName: this.Scenario, + startTime, + endTime, + workloadMetrics, + null, + commandArguments, + this.Tags, + telemetryContext); + } + } + catch (SchemaException exc) + { + throw new WorkloadResultsException($"Failed to parse workload results.", exc, ErrorReason.WorkloadResultsParsingFailed); + } + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaConsumerMetricsParser.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaConsumerMetricsParser.cs new file mode 100644 index 0000000000..77a8f0afca --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaConsumerMetricsParser.cs @@ -0,0 +1,132 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Data; + using System.Linq; + using VirtualClient; + using VirtualClient.Contracts; + + /// + /// Consumer parser for Kafka. + /// + public class KafkaConsumerMetricsParser : MetricsParser + { + /// + /// Initializes a new instance of the class. + /// + /// + public KafkaConsumerMetricsParser(string rawText) + : base(rawText) + { + } + + /// + /// Kafka result. + /// + public DataTable KafkaResult { get; set; } + + /// + public override IList Parse() + { + List metrics = new List(); + this.AddMetricsFromCsv(metrics); + + if (metrics.Count <= 0) + { + throw new SchemaException( + $"Invalid/unpexpected format. The Kafka benchmark workload consumer results were not in the expected format or did not contain valid measurements."); + } + + return metrics; + } + + private void AddMetricsFromCsv(List metrics) + { + string consumerResult = this.RawText.Split(Environment.NewLine, StringSplitOptions.RemoveEmptyEntries)?.LastOrDefault(); + + if (!string.IsNullOrEmpty(consumerResult)) + { + var values = consumerResult.Split(',', StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries); + + if (values.Length > 9) + { + // The metric values are read in the order at which they exist within the Kafka Consumer + // benchmark CSV output. + if (double.TryParse(values[2], out double dataConsumedInMb)) + { + metrics.Add(new Metric( + "data_consumed_in_megabyte", + dataConsumedInMb, + MetricUnit.Megabytes, + relativity: MetricRelativity.HigherIsBetter, + description: "Total data consumed in MB.")); + } + + if (double.TryParse(values[3], out double mbPerSec)) + { + metrics.Add(new Metric( + "megabyte_per_sec_throughput", + mbPerSec, + MetricUnit.MegabytesPerSecond, + relativity: MetricRelativity.HigherIsBetter, + description: "Megabytes per second/ Throughput")); + } + + if (double.TryParse(values[4], out double dataConsumedInNMsg)) + { + metrics.Add(new Metric( + "data_consumed_in_nMsg", + dataConsumedInNMsg, + MetricUnit.Operations, + relativity: MetricRelativity.HigherIsBetter, + description: "Total data consumed in number of messages.")); + } + + if (double.TryParse(values[5], out double nMsgPerSec)) + { + metrics.Add(new Metric( + "nMsg_per_sec_throughput", + nMsgPerSec, + MetricUnit.OperationsPerSec, + relativity: MetricRelativity.HigherIsBetter, + description: "Number of messages per second/ Throughput")); + } + + if (double.TryParse(values[7], out double fetchTimeInMilliSec)) + { + metrics.Add(new Metric( + "fetch_time_in_millisec", + fetchTimeInMilliSec, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "Fetch time per milli seconds")); + } + + if (double.TryParse(values[8], out double fetchMbPerSec)) + { + metrics.Add(new Metric( + "fetch_megabyte_per_sec", + fetchMbPerSec, + MetricUnit.MegabytesPerSecond, + relativity: MetricRelativity.HigherIsBetter, + description: "Fetch megabytes per second/ Throughput")); + } + + if (double.TryParse(values[9], out double fetchNMsgPerSec)) + { + metrics.Add(new Metric( + "fetch_nMsg_per_sec", + fetchNMsgPerSec, + MetricUnit.OperationsPerSec, + relativity: MetricRelativity.HigherIsBetter, + description: "Fetch number of messages per second/ Throughput")); + } + } + } + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs new file mode 100644 index 0000000000..bdcfb80199 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs @@ -0,0 +1,298 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Net; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using VirtualClient; + using VirtualClient.Common; + using VirtualClient.Common.Extensions; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + + /// + /// The Kafka client action + /// + [SupportedPlatforms("win-x64,win-arm64,linux-arm64,linux-x64")] + public abstract class KafkaExecutor : VirtualClientComponent + { + /// + /// Constructor + /// + /// Provides required dependencies to the component. + /// Parameters defined in the profile or supplied on the command line. + public KafkaExecutor(IServiceCollection dependencies, IDictionary parameters) + : base(dependencies, parameters) + { + this.ApiClientManager = dependencies.GetService(); + this.SystemManagement = this.Dependencies.GetService(); + } + + /// + /// Parameter defines the number of server instances/copies to run. + /// + public int ServerInstanceCount + { + get + { + return this.Parameters.GetValue(new string[] { nameof(this.ServerInstanceCount), "ServerInstances" }, 3); + } + } + + /// + /// Parameter defines the number of client instances/copies to run. + /// + public int MaxClientInstanceCount + { + get + { + return this.Parameters.GetValue(new string[] { nameof(this.MaxClientInstanceCount), "MaxClientInstances" }, 3); + } + } + + /// + /// Parameter defines the number of client instances/copies to run. + /// + public int Partitions + { + get + { + return this.Parameters.GetValue(nameof(this.Partitions), 6); + } + } + + /// + /// Java Development Kit package name. + /// + public string JdkPackageName + { + get + { + return this.Parameters.GetValue(nameof(this.JdkPackageName), "javadevelopmentkit"); + } + } + + /// + /// Provides the ability to create API clients for interacting with local as well as remote instances + /// of the Virtual Client API service. + /// + protected IApiClientManager ApiClientManager { get; } + + /// + /// Client used to communicate with the hosted instance of the + /// Virtual Client API at server side. + /// + protected IApiClient ServerApiClient { get; set; } + + /// + /// Server IpAddress on which Kafka Server runs. + /// + protected string ServerIpAddress { get; set; } + + /// + /// Cancellation Token Source for Server. + /// + protected CancellationTokenSource ServerCancellationSource { get; set; } + + /// + /// An interface that can be used to communicate with the underlying system. + /// + protected ISystemManagement SystemManagement { get; set; } + + /// + /// Path to Kafka server package. + /// + protected string KafkaPackagePath { get; set; } + + /// + /// Path to the Java Development Kit package. + /// + protected string JDKPackagePath { get; set; } + + /// + /// Command type to be used based on platform. + /// + protected string PlatformSpecificCommandType { get; set; } + + /// + /// Validates the component definition for requirements. + /// + protected override void Validate() + { + base.Validate(); + if (this.MaxClientInstanceCount > this.Partitions) + { + throw new ArgumentException( + $"Parameter {nameof(this.MaxClientInstanceCount)} {this.MaxClientInstanceCount} should be <= parameter {nameof(this.Partitions)} {this.Partitions}"); + } + } + + /// + /// Executes the workload. + /// + /// Provides context information that will be captured with telemetry events. + /// A token that can be used to cancel the operation. + protected override Task ExecuteAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + throw new NotImplementedException(); + } + + /// + /// Initializes the environment and dependencies for running the Kafka workload. + /// + protected override async Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + await this.EvaluateParametersAsync(cancellationToken); + + if (this.IsMultiRoleLayout()) + { + ClientInstance clientInstance = this.GetLayoutClientInstance(); + string layoutIPAddress = clientInstance.IPAddress; + + this.ThrowIfLayoutClientIPAddressNotFound(layoutIPAddress); + this.ThrowIfRoleNotSupported(clientInstance.Role); + } + + this.InitializeApiClients(); + + DependencyPath kafkaPackage = await this.GetPackageAsync(this.PackageName, cancellationToken); + this.KafkaPackagePath = kafkaPackage.Path; + + DependencyPath jdkPackage = await this.GetPackageAsync(this.JdkPackageName, cancellationToken); + this.JDKPackagePath = jdkPackage.Path; + this.SetEnvironmentVariable(EnvironmentVariable.JAVA_HOME, this.JDKPackagePath, EnvironmentVariableTarget.Process); + + switch (this.Platform) + { + case PlatformID.Win32NT: + this.PlatformSpecificCommandType = "cmd"; + break; + + case PlatformID.Unix: + this.PlatformSpecificCommandType = "bash"; + break; + + default: + throw new WorkloadException( + $"The kafka workload is not supported on the current platform/architecture " + + $"{PlatformSpecifics.GetPlatformArchitectureName(this.Platform, this.CpuArchitecture)}." + + ErrorReason.PlatformNotSupported); + } + } + + /// + /// Initializes API client. + /// + protected void InitializeApiClients() + { + IApiClientManager clientManager = this.Dependencies.GetService(); + bool isSingleVM = !this.IsMultiRoleLayout(); + + if (isSingleVM) + { + this.ServerApiClient = clientManager.GetOrCreateApiClient(IPAddress.Loopback.ToString(), IPAddress.Loopback); + } + else + { + ClientInstance serverInstance = this.GetLayoutClientInstances(ClientRole.Server).First(); + IPAddress.TryParse(serverInstance.IPAddress, out IPAddress serverIPAddress); + this.ServerIpAddress = serverIPAddress.ToString(); + + this.ServerApiClient = clientManager.GetOrCreateApiClient(serverIPAddress.ToString(), serverIPAddress); + this.RegisterToSendExitNotifications($"{this.TypeName}.ExitNotification", this.ServerApiClient); + } + } + + /// + /// Returns platform specific formatted command string + /// + protected string GetPlatformFormattedCommandArguement(string scriptPath, string scriptArgs) + { + string scriptCommand = $"{scriptPath} {scriptArgs}"; + string commandArgs = this.Platform == PlatformID.Win32NT ? $"/c {scriptCommand}" : scriptCommand; + + return commandArgs; + } + + /// + /// Executes a command within an isolated process. + /// + /// The command to execute within the process. + /// The arguments to supply to the command. + /// The working directory from which the command should be executed. + /// Provides context information to include with telemetry events. + /// A token that can be used to cancel the process execution. + /// True to run the process with elevated privileges. Default = false + /// The username to use for executing the command. Note that this is applied ONLY for Unix/Linux scenarios. + /// Optional delegate/action allows the user to configure the process after creation but before execution. + /// The process that executed the command. + protected async Task RunProcessAsync( + string command, + string commandArguments, + string workingDirectory, + EventContext telemetryContext, + CancellationToken cancellationToken, + bool runElevated = false, + string username = null, + Action beforeExecution = null) + { + command.ThrowIfNullOrWhiteSpace(nameof(command)); + telemetryContext.ThrowIfNull(nameof(telemetryContext)); + + if (!string.IsNullOrWhiteSpace(username)) + { + if (this.Platform != PlatformID.Unix) + { + throw new NotSupportedException($"The application of a username is not supported on '{this.Platform}' platform/architecture systems."); + } + + if (!runElevated) + { + throw new NotSupportedException($"The application of a username is not supported unless running elevated. Use the '{nameof(runElevated)}' parameter."); + } + } + + EventContext relatedContext = telemetryContext.Clone() + .AddContext(nameof(command), command) + .AddContext(nameof(commandArguments), commandArguments) + .AddContext(nameof(workingDirectory), workingDirectory) + .AddContext(nameof(runElevated), runElevated); + + IProcessProxy process = null; + if (!cancellationToken.IsCancellationRequested) + { + ProcessManager processManager = this.Dependencies.GetService(); + + if (!runElevated) + { + process = processManager.CreateProcess(command, commandArguments, workingDirectory); + } + else if (this.Platform == PlatformID.Unix && !string.IsNullOrEmpty(this.JDKPackagePath)) + { + // sudo strips environment variables by default on Linux (env_reset policy). + // Using 'env JAVA_HOME=...' as the elevated command ensures the variable is + // set in the process where Kafka scripts execute. + process = processManager.CreateElevatedProcess(this.Platform, "env", $"JAVA_HOME={this.JDKPackagePath} {command} {commandArguments}", workingDirectory, username); + } + else + { + process = processManager.CreateElevatedProcess(this.Platform, command, commandArguments, workingDirectory, username); + } + + this.Logger.LogTraceMessage($"Executing: {command} {SensitiveData.ObscureSecrets(commandArguments)}".Trim(), relatedContext); + + beforeExecution?.Invoke(process); + await process.StartAndWaitAsync(cancellationToken) + .ConfigureAwait(false); + } + + return process; + } + } +} \ No newline at end of file diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaProducerMetricsParser.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaProducerMetricsParser.cs new file mode 100644 index 0000000000..9e2f906f42 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaProducerMetricsParser.cs @@ -0,0 +1,144 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.Data; + using System.Linq; + using System.Text.RegularExpressions; + using VirtualClient; + using VirtualClient.Contracts; + + /// + /// Producer parser for Kafka. + /// + public class KafkaProducerMetricsParser : MetricsParser + { + /// + /// Initializes a new instance of the class. + /// + /// + public KafkaProducerMetricsParser(string rawText) + : base(rawText) + { + } + + /// + /// Kafka result. + /// + public DataTable KafkaResult { get; set; } + + /// + public override IList Parse() + { + List metrics = new List(); + this.AddMetricsFromCsv(metrics); + + if (metrics.Count <= 0) + { + throw new SchemaException( + $"Invalid/unpexpected format. The Kafka benchmark workload producer results were not in the expected format or did not contain valid measurements."); + } + + return metrics; + } + + private void AddMetricsFromCsv(List metrics) + { + string producerResult = this.RawText.Split(Environment.NewLine, StringSplitOptions.RemoveEmptyEntries)?.LastOrDefault(); + if (!string.IsNullOrEmpty(producerResult)) + { + var values = producerResult.Split(',', StringSplitOptions.TrimEntries | StringSplitOptions.RemoveEmptyEntries) + .Select(item => Regex.Replace(item, "\\s(.*)", string.Empty)) + .ToList(); + + if (values.Count > 7) + { + // The metric values are read in the order at which they exist within the Kafka Producer + // benchmark CSV output. + if (double.TryParse(values[0], out double totalRecordsSent)) + { + metrics.Add(new Metric( + "Total_Records_Sent", + totalRecordsSent, + MetricUnit.Operations, + relativity: MetricRelativity.HigherIsBetter, + description: "Total records sent.")); + } + + if (double.TryParse(values[1], out double recordsPerSec)) + { + metrics.Add(new Metric( + "Records_Per_Sec", + recordsPerSec, + MetricUnit.OperationsPerSec, + relativity: MetricRelativity.HigherIsBetter, + description: "Records sent per second.")); + } + + if (double.TryParse(values[2], out double latencyAvg)) + { + metrics.Add(new Metric( + "Latency-Avg", + latencyAvg, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "Average latency for requests/operations during the period of time.")); + } + + if (double.TryParse(values[3], out double latencyMax)) + { + metrics.Add(new Metric( + "Latency-Max", + latencyMax, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "Maximum latency for requests/operations during the period of time.")); + } + + if (double.TryParse(values[4], out double latencyP50)) + { + metrics.Add(new Metric( + "Latency-P50", + latencyP50, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "The latency for 50% of all requests was at or under this value.")); + } + + if (double.TryParse(values[5], out double latencyP95)) + { + metrics.Add(new Metric( + "Latency-P95", + latencyP95, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "The latency for 95% of all requests was at or under this value.")); + } + + if (double.TryParse(values[6], out double latencyP99)) + { + metrics.Add(new Metric( + "Latency-P99", + latencyP99, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "The latency for 99% of all requests was at or under this value.")); + } + + if (double.TryParse(values[7], out double latencyP99_9)) + { + metrics.Add(new Metric( + "Latency-P99.9", + latencyP99_9, + MetricUnit.Milliseconds, + relativity: MetricRelativity.LowerIsBetter, + description: "The latency for 99.9% of all requests was at or under this value.")); + } + } + } + } + } +} diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs new file mode 100644 index 0000000000..4241c551fe --- /dev/null +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs @@ -0,0 +1,509 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +namespace VirtualClient.Actions +{ + using System; + using System.Collections.Generic; + using System.IO.Abstractions; + using System.Linq; + using System.Text.RegularExpressions; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Microsoft.Extensions.Logging; + using Polly; + using VirtualClient; + using VirtualClient.Common; + using VirtualClient.Common.Extensions; + using VirtualClient.Common.Telemetry; + using VirtualClient.Contracts; + using VirtualClient.Logging; + + /// + /// Kafka server executor. + /// + [SupportedPlatforms("linux-arm64,linux-x64,win-arm64,win-x64")] + public class KafkaServerExecutor : KafkaExecutor + { + private readonly int port = 9092; + private IFileSystem fileSystem; + private ISystemManagement systemManagement; + private List serverProcesses; + private bool disposed; + + /// + /// Initializes a new instance of the class. + /// + /// An enumeration of dependencies that can be used for dependency injection. + /// An enumeration of key-value pairs that can control the execution of the component. + public KafkaServerExecutor(IServiceCollection dependencies, IDictionary parameters = null) + : base(dependencies, parameters) + { + this.systemManagement = this.Dependencies.GetService(); + this.fileSystem = this.systemManagement.FileSystem; + this.serverProcesses = new List(); + this.ServerRetryPolicy = Policy.Handle(exc => !(exc is OperationCanceledException)) + .WaitAndRetryAsync(10, (retries) => TimeSpan.FromSeconds(retries)); + } + + /// + /// Client used to communicate with the locally hosted instance of the + /// Virtual Client API. + /// + protected IApiClient ApiClient + { + get + { + return this.ServerApiClient; + } + } + + /// + /// A retry policy to apply to the server when starting to handle transient issues that + /// would otherwise prevent it from starting successfully. + /// + protected IAsyncPolicy ServerRetryPolicy { get; set; } + + /// + /// Path to Kafka server executable - start. + /// + protected string KafkaStartScriptPath { get; set; } + + /// + /// Path to Kafka server executable - stop. + /// + protected string KafkaStopScriptPath { get; set; } + + /// + /// Path to Kafka storage package. + /// + protected string KafkaStorageScriptPath { get; set; } + + /// + /// Path to Kafka kraft folder. + /// + protected string KafkaKraftDirectoryPath { get; set; } + + /// + /// True/false whether the Producer test has ran once + /// + protected bool OneTimeConfigSetup { get; set; } + + /// + /// Disposes of resources used by the executor including shutting down any + /// instances of Kafka server running. + /// + protected override void Dispose(bool disposing) + { + if (disposing) + { + if (!this.disposed && this.serverProcesses.Any()) + { + try + { + Console.WriteLine("Disposed"); + // We MUST stop the server instances from running before VC exits or they will + // continue running until explicitly stopped. This is a problem for running Kafka + // workloads back to back because the requisite ports will be in use already on next + // VC startup. + this.KillServerInstancesAsync(CancellationToken.None) + .GetAwaiter().GetResult(); + } + catch + { + // Best effort + } + + this.disposed = true; + } + } + } + + /// + /// Initializes the environment and dependencies for server of Kafka workload. + /// + /// Provides context information that will be captured with telemetry events. + /// A token that can be used to cancel the operation. + /// + protected override async Task InitializeAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + await base.InitializeAsync(telemetryContext, cancellationToken); + this.KafkaKraftDirectoryPath = this.Combine(this.KafkaPackagePath, "config", "kraft"); + switch (this.Platform) + { + case PlatformID.Win32NT: + this.KafkaStartScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-server-start.bat"); + this.KafkaStopScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-server-stop.bat"); + this.KafkaStorageScriptPath = this.Combine(this.KafkaPackagePath, "bin", "windows", "kafka-storage.bat"); + break; + + case PlatformID.Unix: + this.KafkaStartScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-server-start.sh"); + this.KafkaStopScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-server-stop.sh"); + this.KafkaStorageScriptPath = this.Combine(this.KafkaPackagePath, "bin", "kafka-storage.sh"); + break; + } + + if (!this.fileSystem.File.Exists(this.KafkaStartScriptPath)) + { + throw new DependencyException( + $"Kafka executable not found at path '{this.KafkaStartScriptPath}'", + ErrorReason.WorkloadDependencyMissing); + } + + switch (this.Platform) + { + case PlatformID.Win32NT: + await this.SystemManagement.MakeFileExecutableAsync(this.KafkaStorageScriptPath, this.Platform, cancellationToken); + await this.SystemManagement.MakeFileExecutableAsync(this.KafkaStartScriptPath, this.Platform, cancellationToken); + await this.SystemManagement.MakeFileExecutableAsync(this.KafkaStopScriptPath, this.Platform, cancellationToken); + break; + + case PlatformID.Unix: + await this.SystemManagement.MakeFilesExecutableAsync(this.KafkaPackagePath, this.Platform, cancellationToken); + break; + } + + this.OpenKafkaPorts(cancellationToken); + this.OneTimeConfigSetup = false; + + await this.ConfigurePropertiesFileAsync(telemetryContext, cancellationToken); + } + + /// + /// Executes server side of workload. + /// + /// Provides context information that will be captured with telemetry events. + /// A token that can be used to cancel the operation. + protected override Task ExecuteAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + return this.Logger.LogMessageAsync($"{this.TypeName}.ExecuteServer", telemetryContext, async () => + { + try + { + this.SetServerOnline(false); + + await this.ServerApiClient.PollForHeartbeatAsync(TimeSpan.FromMinutes(5), cancellationToken); + + if (this.ResetServer(telemetryContext)) + { + await this.KillServerInstancesAsync(cancellationToken); + + await this.StartServerInstancesAsync(telemetryContext, cancellationToken); + } + + this.SetServerOnline(true); + + if (this.IsMultiRoleLayout()) + { + await Task.WhenAny(this.serverProcesses); + await this.StopServersAsync(telemetryContext, cancellationToken); + // A cancellation is request, then we allow each of the server instances + // to gracefully exit. If a cancellation was not requested, it means that one + // or more of the server instances exited and we will want to allow the component + // to start over restarting the servers. + if (cancellationToken.IsCancellationRequested) + { + await Task.WhenAll(this.serverProcesses); + } + } + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.StartServerInstancesError", + LogLevel.Error, + telemetryContext.Clone().AddError(exc)); + this.SetServerOnline(false); + throw; + } + }); + } + + private Task StartServerInstancesAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + this.serverProcesses.Clear(); + + EventContext relatedContext = telemetryContext.Clone() + .AddContext("serverInstanceCount", this.ServerInstanceCount) + .AddContext("brokerPortRange", $"{this.port}-{this.port + this.ServerInstanceCount}") + .AddContext("controllerPortRange", $"2{this.port}-2{this.port + this.ServerInstanceCount}"); + + return this.Logger.LogMessageAsync($"{this.TypeName}.StartServerInstances", relatedContext, async () => + { + try + { + relatedContext.AddContext("command", this.PlatformSpecificCommandType); + relatedContext.AddContext("workingDir", this.KafkaPackagePath); + + // Get Cluster Id + string clusterId = await this.GetClusterId(this.PlatformSpecificCommandType, this.KafkaPackagePath, relatedContext, cancellationToken); + clusterId = Regex.Replace(clusterId, $"(\n\r)|(\r\n)", " "); + + // Start kafka servers based on server instances provided + for (int serverInstance = 0; serverInstance < this.ServerInstanceCount; serverInstance++) + { + string propertiesFilePath = this.PlatformSpecifics.Combine(this.KafkaKraftDirectoryPath, $"server-{serverInstance + 1}.properties"); + + // Format log directories + string formatLogDirCmdArgs = $"format -t {clusterId} -c {propertiesFilePath}"; + formatLogDirCmdArgs = this.GetPlatformFormattedCommandArguement(this.KafkaStorageScriptPath, formatLogDirCmdArgs); + + await this.StartServerAndWaitForExitAsync("Format Directory: Complete", this.PlatformSpecificCommandType, formatLogDirCmdArgs, this.KafkaPackagePath, relatedContext, cancellationToken); + + // Give permissions to log directories + if (this.Platform == PlatformID.Unix) + { + string logDirPath = $"/tmp/kraft-combined-logs-{serverInstance + 1}"; + await this.SystemManagement.MakeFilesExecutableAsync(logDirPath, this.Platform, cancellationToken); + } + + // Start kafka server + int port = this.port + serverInstance; + string commandArguments = this.GetPlatformFormattedCommandArguement(this.KafkaStartScriptPath, propertiesFilePath); + this.serverProcesses.Add(this.StartServerAndWaitForExitAsync( + $"Kafka server process exited (port = {port})...", + this.PlatformSpecificCommandType, + commandArguments, + this.KafkaPackagePath, + relatedContext, + cancellationToken, + "Kafka", + addCleanupTasks: false)); + } + } + catch (OperationCanceledException) + { + // Expected whenever certain operations (e.g. Task.Delay) are cancelled. + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.StartServerInstancesError", + LogLevel.Error, + telemetryContext.Clone().AddError(exc)); + + throw; + } + }); + } + + private async Task GetClusterId(string command, string workingDirectory, EventContext telemetryContext, CancellationToken cancellationToken) + { + try + { + string clusterIdCmdArgs = this.GetPlatformFormattedCommandArguement(this.KafkaStorageScriptPath, "random-uuid"); + telemetryContext.AddContext(nameof(clusterIdCmdArgs), clusterIdCmdArgs); + string output; + using (IProcessProxy process = await this.RunProcessAsync(command, clusterIdCmdArgs, workingDirectory, telemetryContext, cancellationToken, runElevated: true)) + { + if (!cancellationToken.IsCancellationRequested) + { + await this.LogProcessDetailsAsync(process, telemetryContext, "ClusterId"); + process.ThrowIfWorkloadFailed(); + } + + output = process.StandardOutput.ToString(); + telemetryContext.AddContext("Cluster Id", output); + } + + return output; + } + catch (OperationCanceledException) + { + // Expected whenever certain operations (e.g. Task.Delay) are cancelled. + return null; + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.GetClusterIdError", + LogLevel.Error, + telemetryContext.Clone().AddError(exc)); + + throw; + } + } + + private Task StartServerAndWaitForExitAsync(string logMessage, string command, string commandArguments, string workingDirectory, EventContext telemetryContext, CancellationToken cancellationToken, string toolName = null, bool addCleanupTasks = true) + { + return (this.ServerRetryPolicy ?? Policy.NoOpAsync()).ExecuteAsync(async () => + { + try + { + telemetryContext.AddContext(nameof(commandArguments), commandArguments); + using (IProcessProxy process = await this.RunProcessAsync(command, commandArguments, workingDirectory, telemetryContext, cancellationToken, runElevated: true)) + { + if (!cancellationToken.IsCancellationRequested) + { + ConsoleLogger.Default.LogMessage(logMessage, telemetryContext); + await this.LogProcessDetailsAsync(process, telemetryContext, toolName); + process.ThrowIfWorkloadFailed(); + } + } + } + catch (OperationCanceledException) + { + // Expected whenever certain operations (e.g. Task.Delay) are cancelled. + } + catch (Exception exc) + { + this.Logger.LogMessage( + $"{this.TypeName}.StartServerAndWaitForExitAsyncError", + LogLevel.Error, + telemetryContext.Clone().AddError(exc)); + + throw; + } + }); + } + + private async Task ConfigurePropertiesFileAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + string logDir = "/tmp/kraft-combined-logs-"; + + for (int serverInstance = 1; serverInstance <= this.ServerInstanceCount; serverInstance++) + { + this.fileSystem.Directory.DeleteAsync(logDir + Convert.ToString(serverInstance)); + telemetryContext.AddContext("DeletedKraftLogDirectory" + Convert.ToString(serverInstance), logDir); + } + + List controllerQuorumVotersList = new List(); + for (int serverInstance = 1; serverInstance <= this.ServerInstanceCount; serverInstance++) + { + string controllerPort = $"2{this.port + serverInstance - 1}"; + string controllerConnectString; + + if (this.IsMultiRoleLayout()) + { + controllerConnectString = $"{serverInstance}@{this.ServerIpAddress}:{controllerPort}"; + } + else + { + controllerConnectString = $"{serverInstance}@localhost:{controllerPort}"; + } + + controllerQuorumVotersList.Add(controllerConnectString); + } + + string controllerQuorumVoters = string.Join(",", controllerQuorumVotersList); + + for (int serverInstance = 1; serverInstance <= this.ServerInstanceCount; serverInstance++) + { + int brokerPort = this.port + serverInstance - 1; + int controllerPort = int.Parse($"2{this.port + serverInstance - 1}"); + string oldServerProperties = this.PlatformSpecifics.Combine(this.KafkaKraftDirectoryPath, "server.properties"); + string newServerProperties = this.PlatformSpecifics.Combine(this.KafkaKraftDirectoryPath, $"server-{serverInstance}.properties"); + this.fileSystem.File.Copy(oldServerProperties, newServerProperties, true); + + await this.fileSystem.File.ReplaceInFileAsync( + newServerProperties, @"node.id *= *[^\n]*", $"node.id={serverInstance}", cancellationToken); + + await this.fileSystem.File.ReplaceInFileAsync( + newServerProperties, @"log.dirs *= *[^\n]*", $"log.dirs=/tmp/kraft-combined-logs-{serverInstance}", cancellationToken); + + if (this.IsMultiRoleLayout()) + { + await this.fileSystem.File.ReplaceInFileAsync( + newServerProperties, @"listeners *= *[^\n]*", $"listeners=PLAINTEXT://{this.ServerIpAddress}:{brokerPort},CONTROLLER://{this.ServerIpAddress}:{controllerPort}", cancellationToken); + } + else + { + await this.fileSystem.File.ReplaceInFileAsync( + newServerProperties, @"listeners *= *[^\n]*", $"listeners=PLAINTEXT://:{brokerPort},CONTROLLER://:{controllerPort}", cancellationToken); + } + + await this.fileSystem.File.ReplaceInFileAsync( + newServerProperties, @"controller.quorum.voters *= *[^\n]*", $"controller.quorum.voters={controllerQuorumVoters}", cancellationToken); + + telemetryContext.AddContext(nameof(this.OneTimeConfigSetup), this.OneTimeConfigSetup); + telemetryContext.AddContext(nameof(newServerProperties), newServerProperties); + } + + this.OneTimeConfigSetup = true; + } + + private Task KillServerInstancesAsync(CancellationToken cancellationToken) + { + this.Logger.LogTraceMessage($"{this.TypeName}.KillServerInstances"); + IEnumerable processes = this.SystemManagement.ProcessManager.GetProcesses("java"); + + if (processes?.Any() == true) + { + foreach (IProcessProxy process in processes) + { + process.SafeKill(); + } + } + + return this.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken); + } + + private async Task StopServersAsync(EventContext telemetryContext, CancellationToken cancellationToken) + { + this.Logger.LogTraceMessage($"{this.TypeName}.StopServerInstances"); + string kafkaStopArgs = this.KafkaStopScriptPath; + if (this.Platform == PlatformID.Win32NT) + { + kafkaStopArgs = $"/c {this.KafkaStopScriptPath}"; + } + + await this.StartServerAndWaitForExitAsync( + $"Stop Kafka servers...", + this.PlatformSpecificCommandType, + kafkaStopArgs, + this.KafkaPackagePath, + telemetryContext, + cancellationToken, + "Kafka"); + + this.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken); + } + + private bool ResetServer(EventContext telemetryContext) + { + bool shouldReset = true; + if (this.serverProcesses?.Any() == true) + { + // Depending upon how the server Task instances are created, the Task may be in a status + // of Running or WaitingForActivation. The server is running in either of these 2 states. + shouldReset = !this.serverProcesses.All(p => p.Status == TaskStatus.Running || p.Status == TaskStatus.WaitingForActivation); + } + + if (shouldReset) + { + this.Logger.LogTraceMessage($"Restart Kafka Server(s)...", telemetryContext); + } + else + { + this.Logger.LogTraceMessage($"Kafka Server(s) Running...", telemetryContext); + } + + return shouldReset; + } + + private void OpenKafkaPorts(CancellationToken cancellationToken) + { + List ports = new List(); + for (int serverInstance = 0; serverInstance < this.ServerInstanceCount; serverInstance++) + { + int brokerPort = this.port + serverInstance; + int controllerPort = int.Parse($"2{this.port + serverInstance}"); + ports.Add(brokerPort); + ports.Add(controllerPort); + } + + this.systemManagement.FirewallManager.EnableInboundConnectionsAsync( + new List + { + new FirewallEntry( + "Kafka: Allow Multiple Machines communications", + "Allows individual machine instances to communicate with other machine in client-server scenario", + "tcp", + new List(ports)) + }, + cancellationToken); + } + } +} diff --git a/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json b/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json new file mode 100644 index 0000000000..a3d7826046 --- /dev/null +++ b/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json @@ -0,0 +1,169 @@ +{ + "Description": "Kafka Benchmark", + "Metadata": { + "RecommendedMinimumExecutionTime": "02:00:00", + "SupportedPlatforms": "win-arm64,win-x64,linux-x64,linux-arm64", + "SupportedOperatingSystems": "Windows,Linux" + }, + "Parameters": { + "ServerInstanceCount": 3, + "ClientInstanceCount": 3, + "RecordSize": 100, + "Partitions": 6, + "RecordCount": 5000000 + }, + "Actions": [ + { + "Type": "KafkaServerExecutor", + "Parameters": { + "Scenario": "kafka_server", + "PackageName": "kafka", + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "Role": "Server" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "setup-topic-replication-factor-one-sync", + "PackageName": "kafka", + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "Partitions": "$.Parameters.Partitions", + "WarmUp": true, + "CommandLine": "--create --if-not-exists --topic sync-test-rep-one --partitions {Partitions} --replication-factor 1 --bootstrap-server {0}:9092", + "Role": "Client", + "CommandType": "CreateTopic" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "setup-topic-replication-factor-dynamic-sync", + "PackageName": "kafka", + "WarmUp": true, + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "Partitions": "$.Parameters.Partitions", + "CommandLine": "--create --if-not-exists --topic sync-test-rep-dynamic --partitions {Partitions} --replication-factor $ReplicationFactor --bootstrap-server {0}:9092", + "Role": "Client", + "CommandType": "CreateTopic" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "setup-topic-replication-factor-dynamic-async", + "PackageName": "kafka", + "WarmUp": true, + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "Partitions": "$.Parameters.Partitions", + "CommandLine": "--create --if-not-exists --topic async-test-rep-dynamic --partitions {Partitions} --replication-factor $ReplicationFactor --bootstrap-server {0}:9092", + "Role": "Client", + "CommandType": "CreateTopic" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "producer-batchSize-8196-sync", + "PackageName": "kafka", + "WarmUp": true, + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "RecordSize": "$.Parameters.RecordSize", + "RecordCount": "$.Parameters.RecordCount", + "CommandLine": "--topic sync-test-rep-one --num-records {RecordCount} --record-size {RecordSize} --throughput -1 --producer-props acks=1 bootstrap.servers={0}:9092 buffer.memory=67108864 batch.size=8196", + "Role": "Client", + "CommandType": "ProducerTest" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "producer-batchSize-64000-sync", + "PackageName": "kafka", + "WarmUp": true, + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "RecordSize": "$.Parameters.RecordSize", + "RecordCount": "$.Parameters.RecordCount", + "CommandLine": "--topic sync-test-rep-dynamic --num-records {RecordCount} --record-size {RecordSize} --throughput -1 --producer-props acks=1 bootstrap.servers={0}:9092 buffer.memory=67108864 batch.size=64000", + "Role": "Client", + "CommandType": "ProducerTest" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "consumer-sync", + "PackageName": "kafka", + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "RecordCount": "$.Parameters.RecordCount", + "CommandLine": "--topic sync-test-rep-one --messages {RecordCount} --bootstrap-server {0}:9092", + "Role": "Client", + "CommandType": "ConsumerTest" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "producer-batchSize-8196-async-{ClientInstanceCount}client", + "PackageName": "kafka", + "WarmUp": true, + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "ClientInstanceCount": "$.Parameters.ClientInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "RecordSize": "$.Parameters.RecordSize", + "RecordCount": "$.Parameters.RecordCount", + "CommandLine": "--topic async-test-rep-dynamic --num-records {RecordCount} --record-size {RecordSize} --throughput -1 --producer-props acks=1 bootstrap.servers={0}:9092 buffer.memory=67108864 batch.size=8196", + "Role": "Client", + "CommandType": "ProducerTest" + } + }, + { + "Type": "KafkaClientExecutor", + "Parameters": { + "Scenario": "consumer-async-{ClientInstanceCount}client", + "PackageName": "kafka", + "ServerInstanceCount": "$.Parameters.ServerInstanceCount", + "ClientInstanceCount": "$.Parameters.ClientInstanceCount", + "MaxClientInstanceCount": "$.Parameters.ClientInstanceCount", + "RecordCount": "$.Parameters.RecordCount", + "CommandLine": "--topic async-test-rep-dynamic --messages {RecordCount} --bootstrap-server {0}:9092", + "Role": "Client", + "CommandType": "ConsumerTest" + } + } + ], + "Dependencies": [ + { + "Type": "DependencyPackageInstallation", + "Parameters": { + "Scenario": "DownloadKafkaPackage", + "BlobContainer": "packages", + "BlobName": "kafka_2.13-3.8.0-2.zip", + "PackageName": "kafka", + "Extract": true + } + }, + { + "Type": "JDKPackageDependencyInstallation", + "Parameters": { + "Scenario": "InstallJDKPackage", + "BlobContainer": "packages", + "BlobName": "microsoft-jdk-21.0.1.zip", + "PackageName": "javadevelopmentkit", + "Extract": true + } + }, + { + "Type": "ApiServer", + "Parameters": { + "Scenario": "StartAPIServer" + } + } + ] +} \ No newline at end of file diff --git a/website/docs/workloads/kafka/kafka-profiles.md b/website/docs/workloads/kafka/kafka-profiles.md new file mode 100644 index 0000000000..7058319f0f --- /dev/null +++ b/website/docs/workloads/kafka/kafka-profiles.md @@ -0,0 +1,116 @@ +--- +slug: /workloads/kafka-profiles +--- + +# Kafka Workload Profiles + +* [Workload details](./kafka.md) +* [Client/server workloads](/docs/guides/client-server) +* [PERF-KAFKA.json](https://github.com/microsoft/VirtualClient/blob/main/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json) + +## PERF-KAFKA.json + +Runs Kafka's producer and consumer performance tools against local or remote KRaft brokers. +Supports linux-x64, linux-arm64, win-x64, and win-arm64. Internet/package-store access is required +to install dependencies on a fresh system. The profile recommends a minimum execution time of two hours; +the time to complete an iteration depends on the system, record counts, and concurrency. + +### Profile parameters + +Supply overrides with `--parameters="Name=Value,,,Name=Value"`. + +| Parameter | Default | Purpose | +|---|---|---| +| `ServerInstanceCount` | 3 | Number of broker processes on the single Server system | +| `ClientInstanceCount` | 3 | Concurrent clients for the explicitly concurrent producer/consumer actions | +| `RecordSize` | 100 | Bytes per producer record | +| `Partitions` | 6 | Partitions per topic | +| `RecordCount` | 5000000 | Records requested per producer or consumer process | + +The profile passes `ClientInstanceCount` as the executors' `MaxClientInstanceCount`. +The effective upper limit is `min(Partitions, 6)`: topic-creation actions validate against the profile's +`Partitions`, but producer and consumer actions do not receive that parameter and validate against the +executor default of 6. Increasing `Partitions` alone does not allow more than 6 client instances in +the shipped profile. + +### Component parameters and fixed settings + +| Parameter | Purpose | +|---|---| +| `Scenario` | Action/scenario identifier, also used to label emitted metrics | +| `PackageName` | Registered Kafka package, `kafka` | +| `Role` | `Server` or `Client` | +| `JdkPackageName` | Registered JDK package; defaults to `javadevelopmentkit` | +| `CommandType` | Client operation: `CreateTopic`, `ProducerTest`, or `ConsumerTest` | +| `CommandLine` | Arguments for the selected Kafka script; the executor formats the server address | +| `WarmUp` | Run the action once per component lifetime; defaults to false | +| `ClientInstanceCount` | Client process concurrency; executor default is 3 | +| `MaxClientInstanceCount` | Count validated against partitions; executor default is 3 | +| `ServerInstanceCount` | Broker count; executor default is 3 | +| `Partitions` | Partition count; executor default is 6 | + +Executors retain compatibility aliases `ServerInstances`, `ClientInstances`, and `MaxClientInstances` +when used in custom component definitions. The shipped profile uses the canonical parameter names above. + +Broker ports are fixed at 9092 plus the zero-based broker index: defaults are 9092, 9093, and 9094. +Controller ports are formed by prefixing `2` to the broker port: defaults are 29092, 29093, and 29094. +`Port` is not a configurable server property in this executor. Permit these ports and the configured +Virtual Client API port (default 4500) between the participating systems in host firewalls and network rules. + +Producer commands use unlimited target throughput (`--throughput -1`), `acks=1`, +`buffer.memory=67108864`, and batch sizes of 8196 or 64000. These settings belong to the profile's +`CommandLine` definitions, not additional global parameter overrides. + +### Single-system examples + +An environment layout is not required. Both roles execute locally. Run with elevated privileges: + +```bash +sudo ./VirtualClient --profile=PERF-KAFKA.json --system=Demo --timeout=1440 + +# Reduced record count for a smoke run, not a representative performance result. +sudo ./VirtualClient --profile=PERF-KAFKA.json --system=Demo --timeout=30 --parameters="RecordCount=100000" +``` + +```powershell +# Elevated PowerShell +.\VirtualClient.exe --profile=PERF-KAFKA.json --system=Demo --timeout=1440 +``` + +The timeout is in minutes. Ensure it allows all required actions to finish. + +### Client/server topology + +Supply the same environment layout to both systems. Roles are `Client` and `Server`; names must match +the corresponding `--client-id` or system name, and IP addresses must be reachable. +Use exactly one Server entry. `ServerInstanceCount` controls the brokers on that system; distributing +those brokers across multiple Server entries is not supported by this executor. + +```json +{ + "clients": [ + { "name": "Client01", "role": "Client", "ipAddress": "10.1.0.1" }, + { "name": "Server01", "role": "Server", "ipAddress": "10.1.0.2" } + ] +} +``` + +```bash +# Server system +sudo ./VirtualClient --profile=PERF-KAFKA.json --system=Demo --timeout=1440 --client-id=Server01 --layout=/path/to/layout.json + +# Client system +sudo ./VirtualClient --profile=PERF-KAFKA.json --system=Demo --timeout=1440 --client-id=Client01 --layout=/path/to/layout.json +``` + +On Windows, use `.\VirtualClient.exe` from an elevated shell and a Windows path for `--layout`. +Apply identical profile-parameter overrides on Client and Server systems. + +### Interpreting results + +The scenario names containing `sync` and `async` are profile labels; consult the command and client count +when interpreting concurrency. The actions without an explicit `ClientInstanceCount` also use the +executor's default of 3, so a `sync` label does not imply a single process. + +Topic-creation and producer actions have `WarmUp=true`. Consumer actions generate the emitted metrics +described in [workload details](./kafka.md). Check both process logs and metric output when validating a run. diff --git a/website/docs/workloads/kafka/kafka.md b/website/docs/workloads/kafka/kafka.md new file mode 100644 index 0000000000..e6c98d98b7 --- /dev/null +++ b/website/docs/workloads/kafka/kafka.md @@ -0,0 +1,133 @@ +--- +slug: /workloads/kafka +--- + +# Kafka + +Apache Kafka is a distributed event-streaming platform. The Virtual Client workload runs Kafka's bundled +`kafka-producer-perf-test` and `kafka-consumer-perf-test` tools against a Kafka cluster using KRaft +(Kafka's built-in metadata quorum, without ZooKeeper). + +* [Kafka documentation](https://kafka.apache.org/38/documentation/) +* [Kafka source](https://github.com/apache/kafka) +* [Workload profiles](./kafka-profiles.md) + +## Supported platforms + +* linux-x64 +* linux-arm64 +* win-x64 +* win-arm64 + +The workload requires elevated privileges, dependency-package access, and sufficient memory and disk space +for the configured brokers, concurrent Java clients, and records. Use dedicated benchmark systems. +This is a plaintext benchmark configuration, not a production Kafka deployment. + +## Workload flow + +The profile supports a single system or separate Client and Server systems. A multi-system layout +must contain exactly one Server system; multiple brokers run on that system, not across Server VMs: + +1. Download Kafka and the Microsoft Build of OpenJDK, then start the Virtual Client API. +2. Generate per-broker KRaft configuration files and a cluster UUID, format the log directories, and start + the configured number of brokers. Brokers on each server use the same host and different ports. +3. Create `sync-test-rep-one`, `sync-test-rep-dynamic`, and `async-test-rep-dynamic` topics. +4. Produce records with batch sizes of 8196 and 64000, then consume the synchronous topic. +5. Produce and consume the asynchronous topic using the configured client-instance count. + +In a multi-system layout, clients wait for the server API heartbeat and online signal before starting. +`ServerInstanceCount` controls broker processes on the Server system; it does not specify the number of VMs. + +The dynamic replication factor is 1 on Windows or when the server-instance count is at most 3. +On Linux it is `ServerInstanceCount - 2` for counts of 4 or 5, and 4 for larger counts. + +Topic creation and producer actions are marked `WarmUp` in the shipped profile. They run once per +component execution lifetime, rather than every profile iteration. Only consumer executions currently +emit benchmark metrics. Producer process output is logged, but the presence of a producer parser does +not mean the profile emits producer metrics. + +## Packages and dependencies + +| Package name | Blob in the `packages` container | Purpose | +|---|---|---| +| `kafka` | `kafka_2.13-3.8.0-2.zip` | Kafka 3.8.0, Scala 2.13, KRaft configuration, and bundled performance tools | +| `javadevelopmentkit` | `microsoft-jdk-21.0.1.zip` | Platform-specific Java runtime | + +The JDK installer registers the platform-specific runtime. The executors set `JAVA_HOME` and explicitly +pass it through elevated Linux commands because `sudo` may remove environment variables. + +Kafka is distributed under Apache License 2.0 and includes third-party notices. Preserve the `LICENSE` +and `NOTICE` files when using or redistributing the package. See the +[Microsoft Build of OpenJDK documentation](https://learn.microsoft.com/java/openjdk/) for JDK licensing +and distribution information. + +### Windows package adjustments + +The `-2` Kafka package contains Windows script changes: + +* `kafka-run-class.bat` uses a wildcard library classpath to avoid "The input line is too long". +* `kafka-server-start.bat` does not depend on WMIC for detecting 32-bit Windows; supported Windows + platforms are 64-bit. +* Kafka and ZooKeeper stop scripts use PowerShell/CIM instead of the deprecated `wmic.exe`. + +These changes are already packaged; no manual script editing is required. Do not substitute the +unmodified upstream archive when verifying this workload. + +## Metrics + +Metrics are logged with tool name `Kafka-Benchmark` and the action's scenario name. Names, units, and +relativity below describe the implementation, including its existing naming conventions. + +### Emitted consumer metrics + +| Metric | Unit | Relativity | Meaning | +|---|---|---|---| +| `data_consumed_in_megabyte` | megabytes | Higher is better | Total data consumed | +| `megabyte_per_sec_throughput` | megabytes/sec | Higher is better | Overall data throughput | +| `data_consumed_in_nMsg` | operations | Higher is better | Total messages consumed | +| `nMsg_per_sec_throughput` | operations/sec | Higher is better | Overall message throughput | +| `fetch_time_in_millisec` | milliseconds | Lower is better | Fetch time reported by the consumer tool | +| `fetch_megabyte_per_sec` | megabytes/sec | Higher is better | Fetch data throughput | +| `fetch_nMsg_per_sec` | operations/sec | Higher is better | Fetch message throughput | + +Concurrent consumer processes each emit their own metric set. Do not interpret a single process's +throughput as the aggregate of all clients. + +### Producer parser metrics + +`KafkaProducerMetricsParser` can parse the following metrics from producer output. The current executor +does **not** emit these metrics during producer actions. + +| Metric | Unit | Relativity | +|---|---|---| +| `Total_Records_Sent` | operations | Higher is better | +| `Records_Per_Sec` | operations/sec | Higher is better | +| `Latency-Avg` | milliseconds | Lower is better | +| `Latency-Max` | milliseconds | Lower is better | +| `Latency-P50` | milliseconds | Lower is better | +| `Latency-P95` | milliseconds | Lower is better | +| `Latency-P99` | milliseconds | Lower is better | +| `Latency-P99.9` | milliseconds | Lower is better | + +## Troubleshooting + +| Symptom | Checks | +|---|---| +| Package missing or download fails | Verify the package-store URL, container, exact blob names, access, and package registration. A cached package is not proof that a fresh system can download it. | +| Java cannot start on Linux | Verify the selected JDK architecture, executable permissions, and `JAVA_HOME` in the elevated process. | +| Windows classpath or WMIC errors | Confirm the profile downloaded `kafka_2.13-3.8.0-2.zip`, not an older cached or upstream archive. | +| Client waits for the server | Verify layout names/roles/IP addresses, server logs, Virtual Client API reachability, and Kafka broker/controller ports. | +| Broker fails to start | Check Java memory, disk capacity, port conflicts, KRaft configuration, and log-directory permissions. | +| No producer metrics | Expected: the executor captures benchmark metrics only for consumer commands. Inspect producer process logs for generation results. | +| Results parsing fails | Inspect the final CSV output from the performance tool. The parsers expect the packaged Kafka version's output format. | + +### State and cleanup + +The workload rewrites per-broker configuration and resets `/tmp/kraft-combined-logs-` directories. +On Windows, Kafka/Java path handling determines the corresponding log location. Server reset and disposal +can terminate Java processes on the benchmark host, including Java processes unrelated to Kafka. +Do not run on a system hosting other Java applications or valuable Kafka data. + +For repeatable comparisons, use dedicated disposable systems and reset Kafka processes, logs/topics, +Virtual Client state, and dependency caches between runs. Compare identical OS images, machine sizes, +profiles, and parameters. Report performance variation separately from command and metric-schema parity. From 5c9f057b3eaf34d9be49d93508fcda204fa8b112 Mon Sep 17 00:00:00 2001 From: Alex Williams-Ferreira Date: Fri, 16 Oct 2026 16:20:37 -0700 Subject: [PATCH 2/4] Bump Kafka migration build version to 3.5.10 Build and verify all six local prerelease packages. Publication remains gated on the reviewed public merge. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 62841113a9..efe3085afe 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.5.9 +3.5.10 From d7bd5bba4197f94c6e91114f383bc87b2d626a3f Mon Sep 17 00:00:00 2001 From: Alex Williams-Ferreira Date: Mon, 19 Oct 2026 09:09:06 -0700 Subject: [PATCH 3/4] Use the copied Kafka package with explicit modification notices Preserve all original package entries except notice-only changes in four Windows scripts. Copy revision 3 from internal storage to the public store with verified matching length, MD5 and content type; retain revision 2 for the original baseline. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Kafka/KafkaProfileTests.cs | 2 +- .../VirtualClient.Main/profiles/PERF-KAFKA.json | 2 +- website/docs/workloads/kafka/kafka.md | 11 ++++++----- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs index 2020d151ac..2a6eb6141b 100644 --- a/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs +++ b/src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs @@ -95,7 +95,7 @@ public void KafkaWorkloadProfileUsesTheWmicFreePackageOnWindows(string profile) dependency => dependency.Parameters.TryGetValue("PackageName", out IConvertible packageName) && packageName.ToString() == "kafka"); - Assert.AreEqual("kafka_2.13-3.8.0-2.zip", kafkaPackage.Parameters["BlobName"].ToString()); + Assert.AreEqual("kafka_2.13-3.8.0-3.zip", kafkaPackage.Parameters["BlobName"].ToString()); } } diff --git a/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json b/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json index a3d7826046..c5a6f49f89 100644 --- a/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json +++ b/src/VirtualClient/VirtualClient.Main/profiles/PERF-KAFKA.json @@ -144,7 +144,7 @@ "Parameters": { "Scenario": "DownloadKafkaPackage", "BlobContainer": "packages", - "BlobName": "kafka_2.13-3.8.0-2.zip", + "BlobName": "kafka_2.13-3.8.0-3.zip", "PackageName": "kafka", "Extract": true } diff --git a/website/docs/workloads/kafka/kafka.md b/website/docs/workloads/kafka/kafka.md index e6c98d98b7..77c36bd243 100644 --- a/website/docs/workloads/kafka/kafka.md +++ b/website/docs/workloads/kafka/kafka.md @@ -50,7 +50,7 @@ not mean the profile emits producer metrics. | Package name | Blob in the `packages` container | Purpose | |---|---|---| -| `kafka` | `kafka_2.13-3.8.0-2.zip` | Kafka 3.8.0, Scala 2.13, KRaft configuration, and bundled performance tools | +| `kafka` | `kafka_2.13-3.8.0-3.zip` | Kafka 3.8.0, Scala 2.13, KRaft configuration, and bundled performance tools | | `javadevelopmentkit` | `microsoft-jdk-21.0.1.zip` | Platform-specific Java runtime | The JDK installer registers the platform-specific runtime. The executors set `JAVA_HOME` and explicitly @@ -63,15 +63,16 @@ and distribution information. ### Windows package adjustments -The `-2` Kafka package contains Windows script changes: +The `-3` Kafka package contains Windows script changes: * `kafka-run-class.bat` uses a wildcard library classpath to avoid "The input line is too long". * `kafka-server-start.bat` does not depend on WMIC for detecting 32-bit Windows; supported Windows platforms are 64-bit. * Kafka and ZooKeeper stop scripts use PowerShell/CIM instead of the deprecated `wmic.exe`. -These changes are already packaged; no manual script editing is required. Do not substitute the -unmodified upstream archive when verifying this workload. +These changes are already packaged; no manual script editing is required. Revision `-3` adds explicit +modification notices to the scripts changed in revision `-2`, without changing script logic, binaries, +or existing license/notice files. Do not substitute the unmodified upstream archive when verifying this workload. ## Metrics @@ -115,7 +116,7 @@ does **not** emit these metrics during producer actions. |---|---| | Package missing or download fails | Verify the package-store URL, container, exact blob names, access, and package registration. A cached package is not proof that a fresh system can download it. | | Java cannot start on Linux | Verify the selected JDK architecture, executable permissions, and `JAVA_HOME` in the elevated process. | -| Windows classpath or WMIC errors | Confirm the profile downloaded `kafka_2.13-3.8.0-2.zip`, not an older cached or upstream archive. | +| Windows classpath or WMIC errors | Confirm the profile downloaded `kafka_2.13-3.8.0-3.zip`, not an older cached or upstream archive. | | Client waits for the server | Verify layout names/roles/IP addresses, server logs, Virtual Client API reachability, and Kafka broker/controller ports. | | Broker fails to start | Check Java memory, disk capacity, port conflicts, KRaft configuration, and log-directory permissions. | | No producer metrics | Expected: the executor captures benchmark metrics only for consumer commands. Inspect producer process logs for generation results. | From 4ecafbe50561aa7c27eae262386690a1039d8b02 Mon Sep 17 00:00:00 2001 From: Alex Williams-Ferreira Date: Mon, 19 Oct 2026 13:33:21 -0700 Subject: [PATCH 4/4] Fix inherited Kafka broker state and shutdown lifecycle Apply the internal lifecycle correction separately from migration adaptations. Preserve live KRaft logs and formatted state, terminate owned elevated broker trees rather than unrelated Java processes, and add repeated-iteration/disposal regression coverage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Kafka/KafkaServerExecutorTests.cs | 127 +++++++++++++++++ .../Kafka/KafkaExecutor.cs | 20 ++- .../Kafka/KafkaServerExecutor.cs | 131 +++++++++++------- website/docs/workloads/kafka/kafka.md | 13 +- 4 files changed, 234 insertions(+), 57 deletions(-) diff --git a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs index 667bd0dce1..12819109cc 100644 --- a/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs +++ b/src/VirtualClient/VirtualClient.Actions.UnitTests/Kafka/KafkaServerExecutorTests.cs @@ -5,12 +5,14 @@ namespace VirtualClient.Actions { using System; using System.Collections.Generic; + using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using Moq; using NUnit.Framework; using VirtualClient; + using VirtualClient.Common; using VirtualClient.Common.Telemetry; using VirtualClient.Contracts; using VirtualClient.TestExtensions; @@ -138,6 +140,131 @@ public async Task KafkaServerExecutorExecutesWorkloadAsExpectedForMultipleInstan } } + [Test] + [TestCase(PlatformID.Unix)] + [TestCase(PlatformID.Win32NT)] + public async Task KafkaServerExecutorPreservesKraftStateAcrossInitializations(PlatformID platform) + { + this.SetupTests(platform); + this.fixture.FileSystem.Setup(fs => fs.File.ReadAllTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("node.id=1"); + + using (var executor = new TestKafkaServerExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + await executor.InitializeAsync(EventContext.None, CancellationToken.None); + await executor.InitializeAsync(EventContext.None, CancellationToken.None); + + this.fixture.FileSystem.Verify(fs => fs.Directory.Delete("/tmp/kraft-combined-logs-1", true), Times.Once); + this.fixture.FileSystem.Verify(fs => fs.File.Copy(It.IsAny(), It.IsAny(), true), Times.Once); + } + } + + [Test] + [TestCase(PlatformID.Unix)] + [TestCase(PlatformID.Win32NT)] + public async Task KafkaServerExecutorKeepsLiveBrokersAcrossIterationsAndStopsOwnedProcessesOnDisposal(PlatformID platform) + { + this.SetupTests(platform); + this.fixture.Layout = null; + this.fixture.FileSystem.Setup(fs => fs.File.ReadAllTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("node.id=1"); + RunningBrokerProcess broker = null; + int starts = 0; + int formats = 0; + this.fixture.ProcessManager.OnCreateProcess = (command, arguments, workingDirectory) => + { + if (arguments.Contains("kafka-server-start")) + { + starts++; + broker = new RunningBrokerProcess + { + StartInfo = new ProcessStartInfo(command, arguments) + }; + return broker; + } + + if (arguments.Contains(" format ")) + { + formats++; + } + + return this.fixture.Process; + }; + + using (var executor = new TestKafkaServerExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + await executor.ExecuteAsync(CancellationToken.None); + await executor.ExecuteAsync(CancellationToken.None); + + Assert.AreEqual(1, starts); + Assert.AreEqual(1, formats); + Assert.IsFalse(broker.HasExited); + this.fixture.FileSystem.Verify(fs => fs.Directory.Delete("/tmp/kraft-combined-logs-1", true), Times.Once); + } + + Assert.IsTrue(broker.HasExited); + Assert.IsTrue(broker.WasDisposed); + } + + [Test] + public async Task KafkaServerExecutorDoesNotReformatStorageWhenABrokerRestarts() + { + this.SetupTests(PlatformID.Unix); + this.fixture.FileSystem.Setup(fs => fs.File.ReadAllTextAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync("node.id=1"); + int starts = 0; + int formats = 0; + this.fixture.ProcessManager.OnCreateProcess = (command, arguments, workingDirectory) => + { + if (arguments.Contains("kafka-server-start")) + { + starts++; + } + + if (arguments.Contains(" format ")) + { + formats++; + } + + return this.fixture.Process; + }; + + using (var executor = new TestKafkaServerExecutor(this.fixture.Dependencies, this.fixture.Parameters)) + { + await executor.ExecuteAsync(CancellationToken.None); + await executor.ExecuteAsync(CancellationToken.None); + + Assert.AreEqual(2, starts); + Assert.AreEqual(1, formats); + } + } + + private class RunningBrokerProcess : InMemoryProcess, IProcessProxy + { + private readonly TaskCompletionSource exited = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + public RunningBrokerProcess() + { + this.OnHasExited = () => this.exited.Task.IsCompleted; + this.OnKill = () => this.exited.TrySetResult(true); + this.OnDispose = () => this.WasDisposed = true; + } + + public bool WasDisposed { get; private set; } + + async Task IProcessProxy.WaitForExitAsync(CancellationToken cancellationToken, TimeSpan? timeout) + { + if (timeout != null) + { + await this.exited.Task.WaitAsync(timeout.Value).ConfigureAwait(false); + } + else + { + await this.exited.Task.WaitAsync(cancellationToken).ConfigureAwait(false); + } + } + } + private class TestKafkaServerExecutor : KafkaServerExecutor { public string TestKafkaStartScriptPath diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs index bdcfb80199..8986e058c9 100644 --- a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaExecutor.cs @@ -288,8 +288,24 @@ protected async Task RunProcessAsync( this.Logger.LogTraceMessage($"Executing: {command} {SensitiveData.ObscureSecrets(commandArguments)}".Trim(), relatedContext); beforeExecution?.Invoke(process); - await process.StartAndWaitAsync(cancellationToken) - .ConfigureAwait(false); + bool completed = false; + try + { + await process.StartAndWaitAsync(cancellationToken).ConfigureAwait(false); + completed = true; + } + finally + { + if (cancellationToken.IsCancellationRequested) + { + process.SafeKill(this.Logger, TimeSpan.FromSeconds(30)); + } + + if (!completed) + { + process.Dispose(); + } + } } return process; diff --git a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs index 4241c551fe..8844ede913 100644 --- a/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs +++ b/src/VirtualClient/VirtualClient.Actions/Kafka/KafkaServerExecutor.cs @@ -4,6 +4,7 @@ namespace VirtualClient.Actions { using System; + using System.Collections.Concurrent; using System.Collections.Generic; using System.IO.Abstractions; using System.Linq; @@ -27,10 +28,12 @@ namespace VirtualClient.Actions public class KafkaServerExecutor : KafkaExecutor { private readonly int port = 9092; + private readonly ConcurrentDictionary brokerProcesses = new ConcurrentDictionary(); private IFileSystem fileSystem; private ISystemManagement systemManagement; private List serverProcesses; private bool disposed; + private bool storageFormatted; /// /// Initializes a new instance of the class. @@ -86,7 +89,7 @@ protected IApiClient ApiClient protected string KafkaKraftDirectoryPath { get; set; } /// - /// True/false whether the Producer test has ran once + /// True once the broker configuration and initial log directories have been prepared. /// protected bool OneTimeConfigSetup { get; set; } @@ -96,28 +99,19 @@ protected IApiClient ApiClient /// protected override void Dispose(bool disposing) { - if (disposing) + try { - if (!this.disposed && this.serverProcesses.Any()) + if (disposing && !this.disposed) { - try - { - Console.WriteLine("Disposed"); - // We MUST stop the server instances from running before VC exits or they will - // continue running until explicitly stopped. This is a problem for running Kafka - // workloads back to back because the requisite ports will be in use already on next - // VC startup. - this.KillServerInstancesAsync(CancellationToken.None) - .GetAwaiter().GetResult(); - } - catch - { - // Best effort - } - + this.StopServersAsync(EventContext.None, CancellationToken.None).GetAwaiter().GetResult(); + this.ServerCancellationSource?.Dispose(); this.disposed = true; } } + finally + { + base.Dispose(disposing); + } } /// @@ -166,9 +160,10 @@ protected override async Task InitializeAsync(EventContext telemetryContext, Can } this.OpenKafkaPorts(cancellationToken); - this.OneTimeConfigSetup = false; - - await this.ConfigurePropertiesFileAsync(telemetryContext, cancellationToken); + if (!this.OneTimeConfigSetup) + { + await this.ConfigurePropertiesFileAsync(telemetryContext, cancellationToken); + } } /// @@ -188,7 +183,7 @@ protected override Task ExecuteAsync(EventContext telemetryContext, Cancellation if (this.ResetServer(telemetryContext)) { - await this.KillServerInstancesAsync(cancellationToken); + await this.StopServersAsync(telemetryContext, cancellationToken); await this.StartServerInstancesAsync(telemetryContext, cancellationToken); } @@ -224,6 +219,8 @@ protected override Task ExecuteAsync(EventContext telemetryContext, Cancellation private Task StartServerInstancesAsync(EventContext telemetryContext, CancellationToken cancellationToken) { this.serverProcesses.Clear(); + this.ServerCancellationSource?.Dispose(); + this.ServerCancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); EventContext relatedContext = telemetryContext.Clone() .AddContext("serverInstanceCount", this.ServerInstanceCount) @@ -238,8 +235,12 @@ private Task StartServerInstancesAsync(EventContext telemetryContext, Cancellati relatedContext.AddContext("workingDir", this.KafkaPackagePath); // Get Cluster Id - string clusterId = await this.GetClusterId(this.PlatformSpecificCommandType, this.KafkaPackagePath, relatedContext, cancellationToken); - clusterId = Regex.Replace(clusterId, $"(\n\r)|(\r\n)", " "); + string clusterId = null; + if (!this.storageFormatted) + { + clusterId = await this.GetClusterId(this.PlatformSpecificCommandType, this.KafkaPackagePath, relatedContext, cancellationToken); + clusterId = Regex.Replace(clusterId, $"(\n\r)|(\r\n)", " "); + } // Start kafka servers based on server instances provided for (int serverInstance = 0; serverInstance < this.ServerInstanceCount; serverInstance++) @@ -247,10 +248,13 @@ private Task StartServerInstancesAsync(EventContext telemetryContext, Cancellati string propertiesFilePath = this.PlatformSpecifics.Combine(this.KafkaKraftDirectoryPath, $"server-{serverInstance + 1}.properties"); // Format log directories - string formatLogDirCmdArgs = $"format -t {clusterId} -c {propertiesFilePath}"; - formatLogDirCmdArgs = this.GetPlatformFormattedCommandArguement(this.KafkaStorageScriptPath, formatLogDirCmdArgs); + if (!this.storageFormatted) + { + string formatLogDirCmdArgs = $"format -t {clusterId} -c {propertiesFilePath}"; + formatLogDirCmdArgs = this.GetPlatformFormattedCommandArguement(this.KafkaStorageScriptPath, formatLogDirCmdArgs); - await this.StartServerAndWaitForExitAsync("Format Directory: Complete", this.PlatformSpecificCommandType, formatLogDirCmdArgs, this.KafkaPackagePath, relatedContext, cancellationToken); + await this.StartServerAndWaitForExitAsync("Format Directory: Complete", this.PlatformSpecificCommandType, formatLogDirCmdArgs, this.KafkaPackagePath, relatedContext, cancellationToken); + } // Give permissions to log directories if (this.Platform == PlatformID.Unix) @@ -268,10 +272,12 @@ private Task StartServerInstancesAsync(EventContext telemetryContext, Cancellati commandArguments, this.KafkaPackagePath, relatedContext, - cancellationToken, + this.ServerCancellationSource.Token, "Kafka", - addCleanupTasks: false)); + isBroker: true)); } + + this.storageFormatted = true; } catch (OperationCanceledException) { @@ -326,14 +332,29 @@ private async Task GetClusterId(string command, string workingDirectory, } } - private Task StartServerAndWaitForExitAsync(string logMessage, string command, string commandArguments, string workingDirectory, EventContext telemetryContext, CancellationToken cancellationToken, string toolName = null, bool addCleanupTasks = true) + private Task StartServerAndWaitForExitAsync(string logMessage, string command, string commandArguments, string workingDirectory, EventContext telemetryContext, CancellationToken cancellationToken, string toolName = null, bool isBroker = false) { return (this.ServerRetryPolicy ?? Policy.NoOpAsync()).ExecuteAsync(async () => { + IProcessProxy brokerProcess = null; try { telemetryContext.AddContext(nameof(commandArguments), commandArguments); - using (IProcessProxy process = await this.RunProcessAsync(command, commandArguments, workingDirectory, telemetryContext, cancellationToken, runElevated: true)) + using (IProcessProxy process = await this.RunProcessAsync( + command, + commandArguments, + workingDirectory, + telemetryContext, + cancellationToken, + runElevated: true, + beforeExecution: created => + { + if (isBroker) + { + brokerProcess = created; + this.brokerProcesses.TryAdd(created, 0); + } + }).ConfigureAwait(false)) { if (!cancellationToken.IsCancellationRequested) { @@ -356,6 +377,13 @@ private Task StartServerAndWaitForExitAsync(string logMessage, string command, s throw; } + finally + { + if (brokerProcess != null) + { + this.brokerProcesses.TryRemove(brokerProcess, out _); + } + } }); } @@ -365,7 +393,7 @@ private async Task ConfigurePropertiesFileAsync(EventContext telemetryContext, C for (int serverInstance = 1; serverInstance <= this.ServerInstanceCount; serverInstance++) { - this.fileSystem.Directory.DeleteAsync(logDir + Convert.ToString(serverInstance)); + await this.fileSystem.Directory.DeleteAsync(logDir + Convert.ToString(serverInstance)); telemetryContext.AddContext("DeletedKraftLogDirectory" + Convert.ToString(serverInstance), logDir); } @@ -427,38 +455,39 @@ await this.fileSystem.File.ReplaceInFileAsync( private Task KillServerInstancesAsync(CancellationToken cancellationToken) { this.Logger.LogTraceMessage($"{this.TypeName}.KillServerInstances"); - IEnumerable processes = this.SystemManagement.ProcessManager.GetProcesses("java"); - - if (processes?.Any() == true) + foreach (IProcessProxy process in this.brokerProcesses.Keys) { - foreach (IProcessProxy process in processes) + process.SafeKill(this.Logger, TimeSpan.FromSeconds(30)); + if (!process.HasExited) { - process.SafeKill(); + throw new WorkloadException("A Kafka broker process did not exit during cleanup.", ErrorReason.WorkloadFailed); } } - return this.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken); + return Task.CompletedTask; } private async Task StopServersAsync(EventContext telemetryContext, CancellationToken cancellationToken) { this.Logger.LogTraceMessage($"{this.TypeName}.StopServerInstances"); - string kafkaStopArgs = this.KafkaStopScriptPath; - if (this.Platform == PlatformID.Win32NT) + this.SetServerOnline(false); + if (this.ServerCancellationSource != null) { - kafkaStopArgs = $"/c {this.KafkaStopScriptPath}"; + await this.ServerCancellationSource.CancelAsync().ConfigureAwait(false); } - await this.StartServerAndWaitForExitAsync( - $"Stop Kafka servers...", - this.PlatformSpecificCommandType, - kafkaStopArgs, - this.KafkaPackagePath, - telemetryContext, - cancellationToken, - "Kafka"); - - this.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken); + await this.KillServerInstancesAsync(cancellationToken).ConfigureAwait(false); + if (this.serverProcesses.Any()) + { + try + { + await Task.WhenAll(this.serverProcesses).WaitAsync(TimeSpan.FromSeconds(30), cancellationToken).ConfigureAwait(false); + } + catch (TimeoutException exc) + { + throw new WorkloadException("Kafka broker shutdown did not complete within 30 seconds.", exc, ErrorReason.WorkloadFailed); + } + } } private bool ResetServer(EventContext telemetryContext) diff --git a/website/docs/workloads/kafka/kafka.md b/website/docs/workloads/kafka/kafka.md index 77c36bd243..17cec58c3b 100644 --- a/website/docs/workloads/kafka/kafka.md +++ b/website/docs/workloads/kafka/kafka.md @@ -124,10 +124,15 @@ does **not** emit these metrics during producer actions. ### State and cleanup -The workload rewrites per-broker configuration and resets `/tmp/kraft-combined-logs-` directories. -On Windows, Kafka/Java path handling determines the corresponding log location. Server reset and disposal -can terminate Java processes on the benchmark host, including Java processes unrelated to Kafka. -Do not run on a system hosting other Java applications or valuable Kafka data. +The workload initially rewrites per-broker configuration and resets +`/tmp/kraft-combined-logs-` directories. It preserves the configuration and formatted KRaft state +across profile iterations and broker restarts; it does not delete live brokers' log directories. +On Windows, Kafka/Java path handling determines the corresponding log location. + +Shutdown targets the broker process trees started by this executor, including their elevated wrappers, +with bounded exit waits. It does not enumerate and terminate unrelated Java processes. Stale brokers +from an interrupted earlier application must be stopped separately before another run can bind their ports. +Use a dedicated system without valuable Kafka data. For repeatable comparisons, use dedicated disposable systems and reset Kafka processes, logs/topics, Virtual Client state, and dependency caches between runs. Compare identical OS images, machine sizes,