using System.Text; using EntKube.Clusters.Domain; using k8s; using k8s.Models; using Microsoft.Extensions.Logging; namespace EntKube.Clusters.Features.AdoptCluster.Components.RabbitMQ; /// /// Checks whether RabbitMQ Cluster Operator is installed and healthy. /// The RabbitMQ Cluster Operator manages the lifecycle of RabbitMQ clusters /// on Kubernetes — provisioning, scaling, upgrades, and TLS configuration. /// /// This check verifies: /// 1. RabbitMQ Cluster Operator pods running /// 2. RabbitmqCluster CRD registered /// 3. Any existing RabbitMQ clusters managed by the operator /// public class RabbitMQCheck : IAdoptionCheck { private readonly ILogger logger; public string ComponentName => "rabbitmq"; public string DisplayName => "RabbitMQ (Message Broker)"; public RabbitMQCheck(ILogger logger) { this.logger = logger; } public async Task CheckAsync(KubernetesCluster cluster, CancellationToken ct = default) { Kubernetes client = BuildClient(cluster); List details = new(); List missing = new(); // Check 1: RabbitMQ operator pods in rabbitmq-system namespace. List operatorPods = await FindOperatorPods(client, ct); if (operatorPods.Count > 0) { details.AddRange(operatorPods.Select(p => $"Operator pod: {p}")); } else { missing.Add("No RabbitMQ Cluster Operator pods found in rabbitmq-system namespace"); return new ComponentCheckResult(ComponentName, ComponentStatus.NotInstalled, details, missing); } // Check 2: RabbitmqCluster CRD exists. bool crdExists = await RabbitmqClusterCrdExists(client, ct); if (crdExists) { details.Add("RabbitmqCluster CRD (rabbitmqclusters.rabbitmq.com) present"); } else { missing.Add("RabbitmqCluster CRD not found"); } // Check 3: Messaging Topology Operator (optional — manages queues, exchanges, etc.) bool topologyOperator = await TopologyOperatorExists(client, ct); if (topologyOperator) { details.Add("Messaging Topology Operator present"); } else { details.Add("Messaging Topology Operator not found (optional)"); } // Check 4: Count existing RabbitmqCluster instances. int clusterCount = await CountRabbitmqClusters(client, ct); details.Add($"Managed RabbitMQ clusters: {clusterCount}"); // Discover version from operator pod image. string? version = await GetVersionFromPods(client, ct); DiscoveredConfiguration config = new( Version: version, Namespace: "rabbitmq-system", HelmReleaseName: null, Values: new Dictionary { ["operatorReplicas"] = operatorPods.Count.ToString(), ["managedClusters"] = clusterCount.ToString(), ["topologyOperator"] = topologyOperator.ToString(), ["crdInstalled"] = crdExists.ToString() }); if (missing.Count > 0) { return new ComponentCheckResult(ComponentName, ComponentStatus.Degraded, details, missing, config); } return new ComponentCheckResult(ComponentName, ComponentStatus.Installed, details, missing, config); } private async Task GetVersionFromPods(Kubernetes client, CancellationToken ct) { try { V1PodList podList = await client.CoreV1.ListNamespacedPodAsync( "rabbitmq-system", labelSelector: "app.kubernetes.io/name=rabbitmq-cluster-operator", cancellationToken: ct); foreach (V1Pod pod in podList.Items) { if (pod.Spec?.Containers?.Count > 0) { string? image = pod.Spec.Containers[0].Image; if (image?.Contains(':') == true) { return image.Split(':').Last().TrimStart('v'); } } } } catch { } return null; } private async Task> FindOperatorPods(Kubernetes client, CancellationToken ct) { List pods = new(); try { V1PodList podList = await client.CoreV1.ListNamespacedPodAsync( "rabbitmq-system", labelSelector: "app.kubernetes.io/name=rabbitmq-cluster-operator", cancellationToken: ct); foreach (V1Pod pod in podList.Items) { if (pod.Status?.Phase == "Running") { pods.Add(pod.Metadata.Name); } } } catch (k8s.Autorest.HttpOperationException ex) when (ex.Response.StatusCode == System.Net.HttpStatusCode.NotFound) { logger.LogDebug("rabbitmq-system namespace not found on cluster"); } catch (Exception ex) { logger.LogWarning(ex, "Error checking for RabbitMQ operator pods"); } return pods; } private async Task RabbitmqClusterCrdExists(Kubernetes client, CancellationToken ct) { try { await client.ApiextensionsV1.ReadCustomResourceDefinitionAsync( "rabbitmqclusters.rabbitmq.com", cancellationToken: ct); return true; } catch { return false; } } private async Task TopologyOperatorExists(Kubernetes client, CancellationToken ct) { try { V1PodList podList = await client.CoreV1.ListNamespacedPodAsync( "rabbitmq-system", labelSelector: "app.kubernetes.io/name=messaging-topology-operator", cancellationToken: ct); return podList.Items.Any(p => p.Status?.Phase == "Running"); } catch { return false; } } private async Task CountRabbitmqClusters(Kubernetes client, CancellationToken ct) { try { object result = await client.CustomObjects.ListClusterCustomObjectAsync( group: "rabbitmq.com", version: "v1beta1", plural: "rabbitmqclusters", cancellationToken: ct); string json = System.Text.Json.JsonSerializer.Serialize(result); using System.Text.Json.JsonDocument doc = System.Text.Json.JsonDocument.Parse(json); if (doc.RootElement.TryGetProperty("items", out System.Text.Json.JsonElement items)) { return items.GetArrayLength(); } return 0; } catch { return 0; } } private static Kubernetes BuildClient(KubernetesCluster cluster) { byte[] kubeConfigBytes = Encoding.UTF8.GetBytes(cluster.KubeConfig); using MemoryStream stream = new(kubeConfigBytes); KubernetesClientConfiguration config = KubernetesClientConfiguration.BuildConfigFromConfigFile(stream, cluster.ContextName); return new Kubernetes(config); } }