Trying to spin up a prometheus

CodexDistTestCore/CodexDistTestCore.csproj

@@ -12,6 +12,7 @@
     <PackageReference Include="nunit" Version="3.13.3" />
     <PackageReference Include="NUnit3TestAdapter" Version="4.4.2" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.5.0" />
+    <PackageReference Include="SharpZipLib" Version="1.4.2" />
   </ItemGroup>
 
 </Project>

CodexDistTestCore/DistTest.cs

@@ -9,6 +9,7 @@ namespace CodexDistTestCore
         private TestLog log = null!;
         private FileManager fileManager = null!;
         private K8sManager k8sManager = null!;
+        private MetricsAggregator metricsAggregator = null!;
 
         [OneTimeSetUp]
         public void GlobalSetup()
@@ -48,6 +49,7 @@ namespace CodexDistTestCore
 
                 fileManager = new FileManager(log);
                 k8sManager = new K8sManager(log, fileManager);
+                metricsAggregator = new MetricsAggregator(log, k8sManager);
             }
         }
 
@@ -57,7 +59,7 @@ namespace CodexDistTestCore
             try
             {
                 log.EndTest();
-                IncludeLogsOnTestFailure();
+                IncludeLogsAndMetricsOnTestFailure();
                 k8sManager.DeleteAllResources();
                 fileManager.DeleteAllTestFiles();
             }
@@ -78,19 +80,57 @@ namespace CodexDistTestCore
             return new OfflineCodexNodes(k8sManager, numberOfNodes);
         }
 
-        private void IncludeLogsOnTestFailure()
+        public MetricsAccess GatherMetrics(ICodexNodeGroup group)
+        {
+            return GatherMetrics(group.ToArray());
+        }
+
+        public MetricsAccess GatherMetrics(params IOnlineCodexNode[] nodes)
+        {
+            Assert.That(nodes.All(n => HasMetricsEnable(n)),
+                "Incorrect test setup: Metrics were not enabled on (all) provided OnlineCodexNodes. " +
+                "To use metrics, please use 'EnableMetrics()' when setting up Codex nodes.");
+
+            return metricsAggregator.BeginCollectingMetricsFor(nodes);
+        }
+
+        public void AssertWithTimeout<T>(Func<T> operation, T isEqualTo, string message)
+        {
+            AssertWithTimeout(operation, isEqualTo, TimeSpan.FromMinutes(10), message);
+        }
+
+        public void AssertWithTimeout<T>(Func<T> operation, T isEqualTo, TimeSpan timeout, string message)
+        {
+            var start = DateTime.UtcNow;
+
+            while (true)
+            {
+                var result = operation();
+                if (result!.Equals(isEqualTo)) return;
+                if (DateTime.UtcNow - start > timeout)
+                {
+                    Assert.That(result, Is.EqualTo(isEqualTo), message);
+                    return;
+                }
+
+                Utils.Sleep(TimeSpan.FromSeconds(10));
+            }
+        }
+
+        private void IncludeLogsAndMetricsOnTestFailure()
         {
             var result = TestContext.CurrentContext.Result;
             if (result.Outcome.Status == NUnit.Framework.Interfaces.TestStatus.Failed)
             {
-                if (IsDownloadingLogsEnabled())
+                if (IsDownloadingLogsAndMetricsEnabled())
                 {
-                    log.Log("Downloading all CodexNode logs because of test failure...");
+                    log.Log("Downloading all CodexNode logs and metrics because of test failure...");
                     k8sManager.ForEachOnlineGroup(DownloadLogs);
+                    metricsAggregator.DownloadAllMetrics();
                 }
                 else
                 {
-                    log.Log("Skipping download of all CodexNode logs due to [DontDownloadLogsOnFailure] attribute.");
+                    log.Log("Skipping download of all CodexNode logs and metrics due to [DontDownloadLogsAndMetricsOnFailure] attribute.");
                 }
             }
         }
@@ -105,11 +145,16 @@ namespace CodexDistTestCore
             }
         }
 
-        private bool IsDownloadingLogsEnabled()
+        private bool IsDownloadingLogsAndMetricsEnabled()
         {
             var testProperties = TestContext.CurrentContext.Test.Properties;
             return !testProperties.ContainsKey(PodLogDownloader.DontDownloadLogsOnFailureKey);
         }
+
+        private bool HasMetricsEnable(IOnlineCodexNode n)
+        {
+            return ((OnlineCodexNode)n).Group.Origin.MetricsEnabled;
+        }
     }
 
     public static class GlobalTestFailure

CodexDistTestCore/K8sCp.cs

@@ -0,0 +1,79 @@
+using ICSharpCode.SharpZipLib.Tar;
+using k8s;
+using System.Text;
+
+namespace CodexDistTestCore
+{
+    // From: https://github.com/kubernetes-client/csharp/blob/master/examples/cp/Cp.cs
+    public class K8sCp
+    {
+        private readonly Kubernetes client;
+
+        public K8sCp(Kubernetes client)
+        {
+            this.client = client;
+        }
+
+        public async Task<int> CopyFileToPodAsync(string podName, string @namespace, string containerName, Stream inputFileStream, string destinationFilePath, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            var handler = new ExecAsyncCallback(async (stdIn, stdOut, stdError) =>
+            {
+                var fileInfo = new FileInfo(destinationFilePath);
+                try
+                {
+                    using (var memoryStream = new MemoryStream())
+                    {
+                        using (var tarOutputStream = new TarOutputStream(memoryStream, Encoding.Default))
+                        {
+                            tarOutputStream.IsStreamOwner = false;
+
+                            var fileSize = inputFileStream.Length;
+                            var entry = TarEntry.CreateTarEntry(fileInfo.Name);
+
+                            entry.Size = fileSize;
+
+                            tarOutputStream.PutNextEntry(entry);
+                            await inputFileStream.CopyToAsync(tarOutputStream);
+                            tarOutputStream.CloseEntry();
+                        }
+
+                        memoryStream.Position = 0;
+
+                        await memoryStream.CopyToAsync(stdIn);
+                        await stdIn.FlushAsync();
+                    }
+
+                }
+                catch (Exception ex)
+                {
+                    throw new IOException($"Copy command failed: {ex.Message}");
+                }
+
+                using StreamReader streamReader = new StreamReader(stdError);
+                while (streamReader.EndOfStream == false)
+                {
+                    string error = await streamReader.ReadToEndAsync();
+                    throw new IOException($"Copy command failed: {error}");
+                }
+            });
+
+            string destinationFolder = GetFolderName(destinationFilePath);
+
+            return await client.NamespacedPodExecAsync(
+                podName,
+                @namespace,
+                containerName,
+                new string[] { "sh", "-c", $"tar xmf - -C {destinationFolder}" },
+                false,
+                handler,
+                cancellationToken);
+        }
+
+        private static string GetFolderName(string filePath)
+        {
+            var folderName = Path.GetDirectoryName(filePath);
+
+            return string.IsNullOrEmpty(folderName) ? "." : folderName;
+        }
+    }
+}

CodexDistTestCore/K8sManager.cs

@@ -58,6 +58,18 @@
             K8s(k => k.FetchPodLog(node, logHandler));
         }
 
+        public PrometheusInfo BringOnlinePrometheus()
+        {
+            PrometheusInfo? info = null;
+            K8s(k => info = k.BringOnlinePrometheus(codexGroupNumberSource.GetNextServicePort()));
+            return info!;
+        }
+
+        public void UploadFileToPod(string podName, string containerName, Stream fileStream, string destinationPath)
+        {
+            K8s(k => k.UploadFileToPod(podName, containerName, fileStream, destinationPath));
+        }
+
         private CodexNodeGroup CreateOnlineCodexNodes(OfflineCodexNodes offline)
         {
             var containers = CreateContainers(offline);

CodexDistTestCore/K8sOperations.cs

@@ -1,5 +1,6 @@
 using CodexDistTestCore.Config;
 using k8s;
+using k8s.KubeConfigModels;
 using k8s.Models;
 using NUnit.Framework;
 
@@ -57,7 +58,30 @@ namespace CodexDistTestCore
             logHandler.Log(stream);
         }
 
+        public PrometheusInfo BringOnlinePrometheus(int servicePort)
+        {
+            EnsureTestNamespace();
+
+            var spec = new K8sPrometheusSpecs();
+            CreatePrometheusDeployment(spec);
+            CreatePrometheusService(spec, servicePort);
+            WaitUntilPrometheusOnline(spec);
+
+            return new PrometheusInfo(servicePort, FetchNewPod());
+        }
+
+        public void UploadFileToPod(string podName, string containerName, Stream fileStream, string destinationPath)
+        {
+            var cp = new K8sCp(client);
+            Utils.Wait(cp.CopyFileToPodAsync(podName, K8sCluster.K8sNamespace, containerName, fileStream, destinationPath));
+        }
+
         private void FetchPodInfo(CodexNodeGroup online)
+        {
+            online.PodInfo = FetchNewPod();
+        }
+
+        private PodInfo FetchNewPod()
         {
             var pods = client.ListNamespacedPod(K8sNamespace).Items;
 
@@ -65,12 +89,13 @@ namespace CodexDistTestCore
             Assert.That(newPods.Length, Is.EqualTo(1), "Expected only 1 pod to be created. Test infra failure.");
 
             var newPod = newPods.Single();
-            online.PodInfo = new PodInfo(newPod.Name(), newPod.Status.PodIP);
+            var info = new PodInfo(newPod.Name(), newPod.Status.PodIP);
 
-            Assert.That(!string.IsNullOrEmpty(online.PodInfo.Name), "Invalid pod name received. Test infra failure.");
+            Assert.That(!string.IsNullOrEmpty(info.Name), "Invalid pod name received. Test infra failure.");
-            Assert.That(!string.IsNullOrEmpty(online.PodInfo.Ip), "Invalid pod IP received. Test infra failure.");
+            Assert.That(!string.IsNullOrEmpty(info.Ip), "Invalid pod IP received. Test infra failure.");
 
             knownPods.Add(newPod.Name());
+            return info;
         }
 
         #region Waiting
@@ -103,6 +128,16 @@ namespace CodexDistTestCore
             WaitUntil(() => !IsTestNamespaceOnline());
         }
 
+        private void WaitUntilPrometheusOnline(K8sPrometheusSpecs spec)
+        {
+            var deploymentName = spec.GetDeploymentName();
+            WaitUntil(() =>
+            {
+                var deployment = client.ReadNamespacedDeployment(deploymentName, K8sNamespace);
+                return deployment?.Status.AvailableReplicas != null && deployment.Status.AvailableReplicas > 0;
+            });
+        }
+
         private void WaitUntil(Func<bool> predicate)
         {
             var start = DateTime.UtcNow;
@@ -166,6 +201,11 @@ namespace CodexDistTestCore
             online.Service = null;
         }
 
+        private void CreatePrometheusService(K8sPrometheusSpecs spec, int servicePort)
+        {
+            client.CreateNamespacedService(spec.CreatePrometheusService(servicePort), K8sNamespace);
+        }
+
         #endregion
 
         #region Deployment management
@@ -232,6 +272,7 @@ namespace CodexDistTestCore
                     Env = dockerImage.CreateEnvironmentVariables(offline, container)
                 });
             }
+
             return result;
         }
 
@@ -242,6 +283,11 @@ namespace CodexDistTestCore
             online.Deployment = null;
         }
 
+        private void CreatePrometheusDeployment(K8sPrometheusSpecs spec)
+        {
+            client.CreateNamespacedDeployment(spec.CreatePrometheusDeployment(), K8sNamespace);
+        }
+
         #endregion
 
         #region Namespace management

CodexDistTestCore/K8sPrometheusSpecs.cs

@@ -0,0 +1,107 @@
+using CodexDistTestCore.Config;
+using k8s.Models;
+
+namespace CodexDistTestCore
+{
+    public class K8sPrometheusSpecs
+    {
+        public const string ContainerName = "dtest-prom";
+        public const string ConfigFilepath = "/etc/prometheus/prometheus.yml";
+        private const string dockerImage = "prom/prometheus:v2.30.3";
+        private const string portName = "prom-1";
+
+        public string GetDeploymentName()
+        {
+            return "test-prom";
+        }
+
+        public V1Deployment CreatePrometheusDeployment()
+        {
+            var deploymentSpec = new V1Deployment
+            {
+                ApiVersion = "apps/v1",
+                Metadata =  new V1ObjectMeta
+                {
+                    Name = GetDeploymentName(),
+                    NamespaceProperty = K8sCluster.K8sNamespace
+                },
+                Spec = new V1DeploymentSpec
+                {
+                    Replicas = 1,
+                    Selector = new V1LabelSelector
+                    {
+                        MatchLabels = CreateSelector()
+                    },
+                    Template = new V1PodTemplateSpec
+                    {
+                        Metadata = new V1ObjectMeta
+                        {
+                            Labels = CreateSelector()
+                        },
+                        Spec = new V1PodSpec
+                        {
+                            Containers = new List<V1Container>
+                            {
+                                new V1Container
+                                {
+                                    Name = ContainerName,
+                                    Image = dockerImage,
+                                    Ports = new List<V1ContainerPort>
+                                    {
+                                        new V1ContainerPort
+                                        {
+                                            ContainerPort = 9090,
+                                            Name = portName
+                                        }
+                                    },
+                                    Command = new List<string>
+                                    {
+                                        $"--web.enable-lifecycle --config.file={ConfigFilepath}"
+                                    },
+                                }
+                            }
+                        }
+                    }
+                }
+            };
+
+            return deploymentSpec;
+        }
+
+        public V1Service CreatePrometheusService(int servicePort)
+        {
+            var serviceSpec = new V1Service
+            {
+                ApiVersion = "v1",
+                Metadata = new V1ObjectMeta
+                {
+                    Name = "codex-prom-service",
+                    NamespaceProperty = K8sCluster.K8sNamespace
+                },
+                Spec = new V1ServiceSpec
+                {
+                    Type = "NodePort",
+                    Selector = CreateSelector(),
+                    Ports = new List<V1ServicePort>
+                    {
+                        new V1ServicePort
+                        {
+                            Name = "prom-service",
+                            Protocol = "TCP",
+                            Port = 9090,
+                            TargetPort = portName,
+                            NodePort = servicePort
+                        }
+                    }
+                }
+            };
+
+            return serviceSpec;
+        }
+
+        private Dictionary<string, string> CreateSelector()
+        {
+            return new Dictionary<string, string> { { "test-prom", "dtest-prom" } };
+        }
+    }
+}

CodexDistTestCore/MetricsAccess.cs

@@ -0,0 +1,15 @@
+namespace CodexDistTestCore
+{
+    public interface IMetricsAccess
+    {
+        int GetMostRecentInt(string metricName, IOnlineCodexNode node);
+    }
+
+    public class MetricsAccess : IMetricsAccess
+    {
+        public int GetMostRecentInt(string metricName, IOnlineCodexNode node)
+        {
+            return 0;
+        }
+    }
+}

CodexDistTestCore/MetricsAggregator.cs

@@ -0,0 +1,96 @@
+using NUnit.Framework;
+
+namespace CodexDistTestCore
+{
+    public class MetricsAggregator
+    {
+        private readonly TestLog log;
+        private readonly K8sManager k8sManager;
+        private readonly List<OnlineCodexNode> activeMetricsNodes = new List<OnlineCodexNode>();
+        private PrometheusInfo? activePrometheus;
+
+        public MetricsAggregator(TestLog log, K8sManager k8sManager)
+        {
+            this.log = log;
+            this.k8sManager = k8sManager;
+        }
+
+        public MetricsAccess BeginCollectingMetricsFor(IOnlineCodexNode[] nodes)
+        {
+            EnsurePrometheusPod();
+
+            AddNewCodexNodes(nodes);
+
+            // Get IPS and ports from all nodes, format prometheus configuration
+            var config = GeneratePrometheusConfig();
+            // Create config file inside prometheus pod
+            k8sManager.UploadFileToPod(
+                activePrometheus!.PodInfo.Name,
+                K8sPrometheusSpecs.ContainerName,
+                config,
+                K8sPrometheusSpecs.ConfigFilepath);
+
+            // HTTP POST request to the /-/reload endpoint (when the --web.enable-lifecycle flag is enabled). 
+
+            return new MetricsAccess();
+        }
+
+        public void DownloadAllMetrics()
+        {
+        }
+
+        private void EnsurePrometheusPod()
+        {
+            if (activePrometheus != null) return;
+            activePrometheus = k8sManager.BringOnlinePrometheus();
+        }
+
+        private void AddNewCodexNodes(IOnlineCodexNode[] nodes)
+        {
+            activeMetricsNodes.AddRange(nodes.Where(n => !activeMetricsNodes.Contains(n)).Cast<OnlineCodexNode>());
+        }
+
+        private Stream GeneratePrometheusConfig()
+        {
+            var stream = new MemoryStream();
+            using var writer = new StreamWriter(stream);
+
+            writer.WriteLine("global:");
+            writer.WriteLine("  scrape_interval: 30s");
+            writer.WriteLine("  scrape_timeout: 10s");
+            writer.WriteLine("");
+            writer.WriteLine("rule_files:");
+            writer.WriteLine("  - alert.yml");
+            writer.WriteLine("");
+            writer.WriteLine("scrape_configs:");
+            writer.WriteLine("  - job_name: services");
+            writer.WriteLine("    metrics_path: /metrics");
+            writer.WriteLine("    static_configs:");
+            writer.WriteLine("      - targets:");
+            writer.WriteLine("          - 'prometheus:9090'");
+
+            foreach (var node in activeMetricsNodes)
+            {
+                var ip = node.Group.PodInfo!.Ip;
+                var port = node.Container.ServicePort;
+                writer.WriteLine($"          - '{ip}:{port}'");
+            }
+
+            return stream;
+        }
+
+
+    }
+
+    public class PrometheusInfo
+    {
+        public PrometheusInfo(int servicePort, PodInfo podInfo)
+        {
+            ServicePort = servicePort;
+            PodInfo = podInfo;
+        }
+
+        public int ServicePort { get; }
+        public PodInfo PodInfo { get; }
+    }
+}

CodexDistTestCore/PodLogDownloader.cs

@@ -8,9 +8,9 @@ namespace CodexDistTestCore
     }
 
     [AttributeUsage(AttributeTargets.Method, AllowMultiple = false)]
-    public class DontDownloadLogsOnFailureAttribute : PropertyAttribute
+    public class DontDownloadLogsAndMetricsOnFailureAttribute : PropertyAttribute
     {
-        public DontDownloadLogsOnFailureAttribute()
+        public DontDownloadLogsAndMetricsOnFailureAttribute()
             : base(Timing.UseLongTimeoutsKey)
         {
         }

Tests/BasicTests/SimpleTests.cs

@@ -7,89 +7,99 @@ namespace Tests.BasicTests
     [TestFixture]
     public class SimpleTests : DistTest
     {
-        [Test]
+        //[Test]
-        public void GetDebugInfo()
+        //public void GetDebugInfo()
-        {
+        //{
-            var dockerImage = new CodexDockerImage();
+        //    var dockerImage = new CodexDockerImage();
 
-            var node = SetupCodexNodes(1).BringOnline()[0];
+        //    var node = SetupCodexNodes(1).BringOnline()[0];
 
-            var debugInfo = node.GetDebugInfo();
+        //    var debugInfo = node.GetDebugInfo();
 
-            Assert.That(debugInfo.spr, Is.Not.Empty);
+        //    Assert.That(debugInfo.spr, Is.Not.Empty);
-            Assert.That(debugInfo.codex.revision, Is.EqualTo(dockerImage.GetExpectedImageRevision()));
+        //    Assert.That(debugInfo.codex.revision, Is.EqualTo(dockerImage.GetExpectedImageRevision()));
-        }
+        //}
 
-        [Test, DontDownloadLogsOnFailure]
+        //[Test, DontDownloadLogsAndMetricsOnFailure]
-        public void CanAccessLogs()
+        //public void CanAccessLogs()
-        {
+        //{
-            var node = SetupCodexNodes(1).BringOnline()[0];
+        //    var node = SetupCodexNodes(1).BringOnline()[0];
 
-            var log = node.DownloadLog();
+        //    var log = node.DownloadLog();
 
-            log.AssertLogContains("Started codex node");
+        //    log.AssertLogContains("Started codex node");
-        }
+        //}
-
-        [Test]
-        public void OneClientTest()
-        {
-            var primary = SetupCodexNodes(1).BringOnline()[0];
-
-            var testFile = GenerateTestFile(1.MB());
-
-            var contentId = primary.UploadFile(testFile);
-
-            var downloadedFile = primary.DownloadContent(contentId);
-
-            testFile.AssertIsEqual(downloadedFile);
-        }
-
-        [Test]
-        public void TwoClientsOnePodTest()
-        {
-            var group = SetupCodexNodes(2).BringOnline();
-
-            var primary = group[0];
-            var secondary = group[1];
-
-            PerformTwoClientTest(primary, secondary);
-        }
-
-        [Test]
-        public void TwoClientsTwoPodsTest()
-        {
-            var primary = SetupCodexNodes(1).BringOnline()[0];
-
-            var secondary = SetupCodexNodes(1).BringOnline()[0];
-
-            PerformTwoClientTest(primary, secondary);
-        }
-
-        [Test]
-        public void TwoClientsTwoLocationsTest()
-        {
-            var primary = SetupCodexNodes(1)
-                            .At(Location.BensLaptop)
-                            .BringOnline()[0];
-
-            var secondary = SetupCodexNodes(1)
-                            .At(Location.BensOldGamingMachine)
-                            .BringOnline()[0];
-
-            PerformTwoClientTest(primary, secondary);
-        }
 
         [Test]
         public void MetricsExample()
         {
-            var group = SetupCodexNodes(1)
+            var group = SetupCodexNodes(2)
-                            .EnableMetrics()
+                        .EnableMetrics()
-                            .BringOnline();
+                        .BringOnline();
 
-            var metrics = BeginGatheringMetrics(group);
+            var metrics = GatherMetrics(group);
-            
+
+            var primary = group[0];
+            var secondary = group[1];
+            primary.ConnectToPeer(secondary);
+
+            Thread.Sleep(10000);
+
+            AssertWithTimeout(
+                () => metrics.GetMostRecentInt("libp2p_peers", primary),
+                isEqualTo: 1,
+                "Number of peers metric was incorrect.");
         }
 
+        //[Test]
+        //public void OneClientTest()
+        //{
+        //    var primary = SetupCodexNodes(1).BringOnline()[0];
+
+        //    var testFile = GenerateTestFile(1.MB());
+
+        //    var contentId = primary.UploadFile(testFile);
+
+        //    var downloadedFile = primary.DownloadContent(contentId);
+
+        //    testFile.AssertIsEqual(downloadedFile);
+        //}
+
+        //[Test]
+        //public void TwoClientsOnePodTest()
+        //{
+        //    var group = SetupCodexNodes(2).BringOnline();
+
+        //    var primary = group[0];
+        //    var secondary = group[1];
+
+        //    PerformTwoClientTest(primary, secondary);
+        //}
+
+        //[Test]
+        //public void TwoClientsTwoPodsTest()
+        //{
+        //    var primary = SetupCodexNodes(1).BringOnline()[0];
+
+        //    var secondary = SetupCodexNodes(1).BringOnline()[0];
+
+        //    PerformTwoClientTest(primary, secondary);
+        //}
+
+        //[Test]
+        //public void TwoClientsTwoLocationsTest()
+        //{
+        //    var primary = SetupCodexNodes(1)
+        //                    .At(Location.BensLaptop)
+        //                    .BringOnline()[0];
+
+        //    var secondary = SetupCodexNodes(1)
+        //                    .At(Location.BensOldGamingMachine)
+        //                    .BringOnline()[0];
+
+        //    PerformTwoClientTest(primary, secondary);
+        //}
+
         private void PerformTwoClientTest(IOnlineCodexNode primary, IOnlineCodexNode secondary)
         {
             primary.ConnectToPeer(secondary);