workloads: Migrate Kafka source, tests and profile to open source

VERSION

@@ -1 +1 @@
-3.5.9
+3.5.10

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

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.

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<DependencyException>(() => 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<string> expectedCommands = new List<string>();
+
+            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<string> expectedCommands = new List<string>();
+
+            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<string, IConvertible> specifics = new Dictionary<string, IConvertible>()
+                {
+                    { 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");
+            }
+        }
+    }
+}

src/VirtualClient/VirtualClient.Actions.FunctionalTests/Kafka/KafkaProfileTests.cs

@@ -0,0 +1,118 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace VirtualClient.Actions
+{
+    using System;
+    using System.Linq;
+    using System.Runtime.InteropServices;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using NUnit.Framework;
+    using VirtualClient;
+    using VirtualClient.Contracts;
+    using VirtualClient.Contracts.Validation;
+    using VirtualClient.TestExtensions;
+
+    [TestFixture]
+    [Category("Functional")]
+    public class KafkaProfileTests
+    {
+        private DependencyFixture mockFixture;
+
+        [OneTimeSetUp]
+        public void SetupFixture()
+        {
+            this.mockFixture = new DependencyFixture();
+
+            this.mockFixture.Setup(PlatformID.Unix, Architecture.X64, "ClientAgent").SetupLayout(
+                new ClientInstance("ClientAgent", "1.2.3.4", "Client"),
+                new ClientInstance("ServerAgent", "1.2.3.5", "Server"));
+
+            ComponentTypeCache.Instance.LoadComponentTypes(MockFixture.TestAssemblyDirectory);
+        }
+
+        [Test]
+        [TestCase(PlatformID.Unix, Architecture.X64)]
+        [TestCase(PlatformID.Unix, Architecture.Arm64)]
+        [TestCase(PlatformID.Win32NT, Architecture.X64)]
+        [TestCase(PlatformID.Win32NT, Architecture.Arm64)]
+        public async Task KafkaWorkloadProfileInstallsTheExpectedDependencies(PlatformID platform, Architecture architecture)
+        {
+            this.mockFixture.Setup(platform, architecture, "ClientAgent").SetupLayout(
+                new ClientInstance("ClientAgent", "1.2.3.4", "Client"),
+                new ClientInstance("ServerAgent", "1.2.3.5", "Server"));
+            this.mockFixture.SetupFile(this.mockFixture.Combine(
+                this.mockFixture.PackagesDirectory,
+                "microsoft-jdk-21.0.1",
+                PlatformSpecifics.GetPlatformArchitectureName(platform, architecture),
+                "bin",
+                platform == PlatformID.Win32NT ? "java.exe" : "java"));
+
+            using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(
+                "PERF-KAFKA.json", this.mockFixture.Dependencies, dependenciesOnly: true))
+            {
+                await executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None);
+
+                WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "kafka");
+                WorkloadAssert.WorkloadPackageInstalled(this.mockFixture, "javadevelopmentkit");
+            }
+        }
+
+        [Test]
+        public void KafkaWorkloadProfileMeetsTheSchemaRequirements()
+        {
+            using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(
+                "PERF-KAFKA.json", this.mockFixture.Dependencies))
+            {
+                ValidationResult result = SchemaRules.Instance.Validate(executor.Profile);
+
+                Assert.IsTrue(result.IsValid, string.Join(Environment.NewLine, result.ValidationErrors));
+            }
+        }
+
+        [Test]
+        [TestCase("PERF-KAFKA.json")]
+        public void KafkaWorkloadProfileParametersAreInlinedCorrectly(string profile)
+        {
+            using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies))
+            {
+                WorkloadAssert.ParameterReferencesInlined(executor.Profile);
+            }
+        }
+
+        [Test]
+        [TestCase("PERF-KAFKA.json")]
+        public void KafkaWorkloadProfileUsesTheWmicFreePackageOnWindows(string profile)
+        {
+            this.mockFixture.Setup(PlatformID.Win32NT, Architecture.X64, "ClientAgent").SetupLayout(
+                new ClientInstance("ClientAgent", "1.2.3.4", "Client"),
+                new ClientInstance("ServerAgent", "1.2.3.5", "Server"));
+
+            using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies))
+            {
+                ExecutionProfileElement kafkaPackage = executor.Profile.Dependencies.Single(
+                    dependency => dependency.Parameters.TryGetValue("PackageName", out IConvertible packageName)
+                        && packageName.ToString() == "kafka");
+
+                Assert.AreEqual("kafka_2.13-3.8.0-2.zip", kafkaPackage.Parameters["BlobName"].ToString());
+            }
+        }
+
+        [Test]
+        [TestCase("PERF-KAFKA.json")]
+        public void KafkaWorkloadProfileActionsWillNotBeExecutedIfTheDependencyPackagesDoesNotExist(string profile)
+        {
+            // We ensure the workload package does not exist.
+            this.mockFixture.PackageManager.Clear();
+
+            using (ProfileExecutor executor = TestProfileResources.CreateProfileExecutor(profile, this.mockFixture.Dependencies))
+            {
+                executor.ExecuteDependencies = false;
+                DependencyException error = Assert.ThrowsAsync<DependencyException>(() => executor.ExecuteAsync(ProfileTiming.OneIteration(), CancellationToken.None));
+                Assert.AreEqual(ErrorReason.WorkloadDependencyMissing, error.Reason);
+                Assert.IsFalse(this.mockFixture.ProcessManager.Commands.Any());
+            }
+        }
+    }
+}