diff --git a/VERSION b/VERSION index 62841113a9..efe3085afe 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.5.9 +3.5.10 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..2a6eb6141b --- /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-3.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..c5a6f49f89 --- /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-3.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..77c36bd243 --- /dev/null +++ b/website/docs/workloads/kafka/kafka.md @@ -0,0 +1,134 @@ +--- +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-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 +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 `-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. 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 + +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-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. | +| 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.