Merge pull request #2043 from rabbitmq/issue-2042

Use Pod name instead of IP for quorum check

Author: Zerpet

bin/kubectl-rabbitmq-plugin/commands_test.go

@@ -66,11 +66,36 @@ func TestMain(m *testing.M) {
 	}
 
 	// Check if cluster operator is ready before running tests
-	if err := checkClusterOperatorReady(); err != nil {
-		os.Remove(pluginBinaryPath)
-		log.Fatalf("Failed to run integration tests: %v", err)
+	// Retry for up to 2 minutes to handle CI startup delays
+	log.Println("Waiting for cluster operator to be ready...")
+
+	checkReady := func() bool {
+		if err := checkClusterOperatorReady(); err != nil {
+			log.Printf("Cluster operator not ready yet: %v", err)
+			return false
+		}
+		return true
+	}
+
+	// TestMain doesn't have access to *testing.T; therefore, we can't use require.Eventually from testify 😞
+	timeout := time.After(2 * time.Minute)
+	ticker := time.NewTicker(2 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-timeout:
+			os.Remove(pluginBinaryPath)
+			log.Fatal("Timed out waiting for cluster operator to be ready after 2 minutes")
+		case <-ticker.C:
+			if checkReady() {
+				log.Println("Cluster operator is ready!")
+				goto ready
+			}
+		}
 	}
 
+ready:
 	code := m.Run()
 
 	os.Remove(pluginBinaryPath)

controllers/reconcile_status.go

@@ -103,11 +103,11 @@ func (r *RabbitmqClusterReconciler) getPodEndpoints(ctx context.Context, rmq *ra
 }
 
 // checkNodeQuorumStatus checks quorum status for a specific node/pod
-func (r *RabbitmqClusterReconciler) checkNodeQuorumStatus(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster, podIP string, podName string) nodeQuorumCheck {
+func (r *RabbitmqClusterReconciler) checkNodeQuorumStatus(ctx context.Context, rmq *rabbitmqv1beta1.RabbitmqCluster, podName string) nodeQuorumCheck {
 	logger := ctrl.LoggerFrom(ctx)
 
 	// Get client for this specific pod
-	rabbitClient, err := rabbitmqclient.GetRabbitmqClientForPod(ctx, r.APIReader, rmq, podIP)
+	rabbitClient, err := rabbitmqclient.GetRabbitmqClientForPod(ctx, r.APIReader, rmq, podName)
 	if err != nil {
 		logger.V(1).Info("Failed to get client for pod", "pod", podName, "error", err)
 		return nodeQuorumCheck{
@@ -173,25 +173,12 @@ func (r *RabbitmqClusterReconciler) checkQuorumStatus(ctx context.Context, rmq *
 			podName = endpoint.TargetRef.Name
 		}
 
-		// Get pod IP
-		if len(endpoint.Addresses) == 0 {
-			logger.V(1).Info("Endpoint has no addresses", "podName", podName)
-			resultsChan <- nodeQuorumCheck{
-				podName: podName,
-				status:  "unavailable",
-				err:     fmt.Errorf("no addresses for endpoint"),
-			}
-			continue
-		}
-
-		podIP := endpoint.Addresses[0]
-
 		wg.Add(1)
-		go func(ip, name string) {
+		go func(name string) {
 			defer wg.Done()
-			result := r.checkNodeQuorumStatus(ctx, rmq, ip, name)
+			result := r.checkNodeQuorumStatus(ctx, rmq, name)
 			resultsChan <- result
-		}(podIP, podName)
+		}(podName)
 	}
 
 	// Wait for all checks to complete

docs/examples/tls/certificate.yaml

@@ -13,11 +13,14 @@ metadata:
   namespace: examples
 spec:
   dnsNames:
-    - "*.tls-nodes.examples.svc.cluster.local"
-    - "*.tls.examples.svc.cluster.local"
-    - "*.tls.examples.svc"
-    - "*.examples.pod"
-    - "*.examples.pod.cluster.local"
+    - "tls-server-0.tls-nodes.examples.svc.cluster.local"
+    - "tls-server-1.tls-nodes.examples.svc.cluster.local"
+    - "tls-server-2.tls-nodes.examples.svc.cluster.local"
+    - "tls-server-0.tls-nodes.examples.svc"
+    - "tls-server-1.tls-nodes.examples.svc"
+    - "tls-server-2.tls-nodes.examples.svc"
+    - "tls.examples.svc.cluster.local"
+    - "tls.examples.svc"
   secretName: tls-secret
   issuerRef:
     kind: Issuer

internal/rabbitmqclient/client.go

@@ -16,7 +16,6 @@ import (
 	"crypto/x509"
 	"fmt"
 	"net/http"
-	"strings"
 
 	rabbithole "github.com/michaelklishin/rabbit-hole/v2"
 	rabbitmqv1beta1 "github.com/rabbitmq/cluster-operator/v2/api/v1beta1"
@@ -33,10 +32,10 @@ type ClientInfo struct {
 	Transport *http.Transport
 }
 
-// GetRabbitmqClientForPod creates a rabbithole client for a specific pod using its IP address.
-// It fetches credentials from the default user secret and connects directly to the pod IP.
-func GetRabbitmqClientForPod(ctx context.Context, k8sClient client.Reader, rmq *rabbitmqv1beta1.RabbitmqCluster, podIP string) (*rabbithole.Client, error) {
-	info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+// GetRabbitmqClientForPod creates a rabbithole client for a specific pod using its stable DNS name.
+// It fetches credentials from the default user secret and connects directly to the pod via its stable DNS.
+func GetRabbitmqClientForPod(ctx context.Context, k8sClient client.Reader, rmq *rabbitmqv1beta1.RabbitmqCluster, podName string) (*rabbithole.Client, error) {
+	info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 	if err != nil {
 		return nil, err
 	}
@@ -57,9 +56,9 @@ func GetRabbitmqClientForPod(ctx context.Context, k8sClient client.Reader, rmq *
 	return rabbitmqClient, nil
 }
 
-// GetClientInfoForPod creates ClientInfo for a specific pod IP.
+// GetClientInfoForPod creates ClientInfo for a specific pod using its stable DNS name.
 // This is useful for checking individual pods instead of going through the service.
-func GetClientInfoForPod(ctx context.Context, k8sClient client.Reader, rmq *rabbitmqv1beta1.RabbitmqCluster, podIP string) (*ClientInfo, error) {
+func GetClientInfoForPod(ctx context.Context, k8sClient client.Reader, rmq *rabbitmqv1beta1.RabbitmqCluster, podName string) (*ClientInfo, error) {
 	// Fetch the default user secret
 	secretName := rmq.ChildResourceName("default-user")
 	secret := &corev1.Secret{}
@@ -82,13 +81,17 @@ func GetClientInfoForPod(ctx context.Context, k8sClient client.Reader, rmq *rabb
 		return nil, fmt.Errorf("password not found in secret %s", secretName)
 	}
 
-	// Build management API URL using pod IP
+	// Build management API URL using pod stable DNS name
 	var port int
 	var scheme string
 
 	// Create HTTP transport
 	var transport *http.Transport
 
+	// Construct the headless service name and pod FQDN
+	headlessServiceName := rmq.ChildResourceName("nodes")
+	podFQDN := fmt.Sprintf("%s.%s.%s.svc", podName, headlessServiceName, rmq.Namespace)
+
 	// Use TLS only if there's no other alternative
 	if rmq.Spec.TLS.DisableNonTLSListeners {
 		port = 15671
@@ -123,9 +126,9 @@ func GetClientInfoForPod(ctx context.Context, k8sClient client.Reader, rmq *rabb
 			tlsConfig.RootCAs = certPool
 		}
 
-		// https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/#pods
-		podIpDnsName := strings.ReplaceAll(podIP, ".", "-")
-		tlsConfig.ServerName = fmt.Sprintf("%s.%s.pod", podIpDnsName, rmq.Namespace)
+		// Use the pod's stable DNS name for TLS ServerName
+		// This matches the SAN entries users typically configure in their certificates
+		tlsConfig.ServerName = podFQDN
 
 		transport = &http.Transport{
 			TLSClientConfig: tlsConfig,
@@ -135,8 +138,8 @@ func GetClientInfoForPod(ctx context.Context, k8sClient client.Reader, rmq *rabb
 		scheme = "http"
 	}
 
-	// Use pod IP directly instead of service name
-	baseURL := fmt.Sprintf("%s://%s:%d", scheme, podIP, port)
+	// Use pod stable DNS name
+	baseURL := fmt.Sprintf("%s://%s:%d", scheme, podFQDN, port)
 
 	return &ClientInfo{
 		BaseURL:   baseURL,

internal/rabbitmqclient/client_test.go

@@ -71,43 +71,44 @@ var _ = Describe("RabbitMQ Client", func() {
 			})
 
 			It("returns client info with correct credentials", func() {
-				podIP := "10.0.0.1"
-				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(info).NotTo(BeNil())
 				Expect(info.Username).To(Equal("test-user"))
 				Expect(info.Password).To(Equal("test-password"))
 			})
 
-			It("returns the correct base URL for non-TLS with pod IP", func() {
-				podIP := "10.0.0.1"
-				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+			It("returns the correct base URL for non-TLS with pod DNS name", func() {
+				podName := "test-cluster-server-0"
+				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
-				Expect(info.BaseURL).To(Equal("http://10.0.0.1:15672"))
+				Expect(info.BaseURL).To(Equal("http://test-cluster-server-0.test-cluster-nodes.test-namespace.svc:15672"))
 			})
 
-			It("returns the correct base URL for TLS with pod IP", func() {
+			It("returns the correct base URL for TLS with pod DNS name", func() {
 				rmq.Spec.TLS.SecretName = "tls-secret"
 				rmq.Spec.TLS.DisableNonTLSListeners = true
-				podIP := "10.0.0.2"
-				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-1"
+				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
-				Expect(info.BaseURL).To(Equal("https://10.0.0.2:15671"))
+				Expect(info.BaseURL).To(Equal("https://test-cluster-server-1.test-cluster-nodes.test-namespace.svc:15671"))
 			})
 
-			It("returns an HTTP transport for TLS", func() {
+			It("returns an HTTP transport for TLS with correct ServerName", func() {
 				rmq.Spec.TLS.SecretName = "tls-secret"
 				rmq.Spec.TLS.DisableNonTLSListeners = true
-				podIP := "10.0.0.1"
-				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(info.Transport).NotTo(BeNil())
 				Expect(info.Transport.TLSClientConfig).NotTo(BeNil())
+				Expect(info.Transport.TLSClientConfig.ServerName).To(Equal("test-cluster-server-0.test-cluster-nodes.test-namespace.svc"))
 			})
 
 			It("returns nil transport for non-TLS", func() {
-				podIP := "10.0.0.1"
-				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				info, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(info.Transport).To(BeNil())
 			})
@@ -122,8 +123,8 @@ var _ = Describe("RabbitMQ Client", func() {
 			})
 
 			It("returns an error", func() {
-				podIP := "10.0.0.1"
-				_, err := GetClientInfoForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				_, err := GetClientInfoForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).To(HaveOccurred())
 				Expect(err.Error()).To(ContainSubstring("failed to get default user secret"))
 			})
@@ -141,8 +142,8 @@ var _ = Describe("RabbitMQ Client", func() {
 
 		Context("when TLS is disabled", func() {
 			It("returns a non-TLS client", func() {
-				podIP := "10.0.0.1"
-				client, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				client, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(client).NotTo(BeNil())
 			})
@@ -152,8 +153,8 @@ var _ = Describe("RabbitMQ Client", func() {
 			It("returns a TLS client", func() {
 				rmq.Spec.TLS.SecretName = "tls-secret"
 				rmq.Spec.TLS.DisableNonTLSListeners = true
-				podIP := "10.0.0.1"
-				client, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				client, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(client).NotTo(BeNil())
 			})
@@ -168,8 +169,8 @@ var _ = Describe("RabbitMQ Client", func() {
 			})
 
 			It("returns an error", func() {
-				podIP := "10.0.0.1"
-				_, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podIP)
+				podName := "test-cluster-server-0"
+				_, err := GetRabbitmqClientForPod(ctx, k8sClient, rmq, podName)
 				Expect(err).To(HaveOccurred())
 				Expect(err.Error()).To(ContainSubstring("failed to get default user secret"))
 			})