Forward ClusterIP traffic to inter-node communication port before CQL is ready

deploy/operator.yaml

@@ -27,7 +27,6 @@ rules:
   - ""
   resources:
   - nodes
-  - endpoints
   verbs:
   - get
   - list
@@ -283,6 +282,30 @@ rules:
   - patch
   - update
   - delete
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - endpoints
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 
 ---
 apiVersion: v1

deploy/operator/00_clusterrole_def.yaml

@@ -17,7 +17,6 @@ rules:
   - ""
   resources:
   - nodes
-  - endpoints
   verbs:
   - get
   - list
@@ -273,3 +272,27 @@ rules:
   - patch
   - update
   - delete
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - endpoints
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch

helm/scylla-operator/templates/clusterrole_def.yaml

@@ -17,7 +17,6 @@ rules:
   - ""
   resources:
   - nodes
-  - endpoints
   verbs:
   - get
   - list
@@ -273,3 +272,27 @@ rules:
   - patch
   - update
   - delete
+- apiGroups:
+  - discovery.k8s.io
+  resources:
+  - endpointslices
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - endpoints
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch

pkg/cmd/operator/operator.go

@@ -259,6 +259,8 @@ func (o *OperatorOptions) run(ctx context.Context, streams genericclioptions.IOS
 		kubeInformers.Policy().V1().PodDisruptionBudgets(),
 		kubeInformers.Networking().V1().Ingresses(),
 		kubeInformers.Batch().V1().Jobs(),
+		kubeInformers.Discovery().V1().EndpointSlices(),
+		kubeInformers.Core().V1().Endpoints(),
 		scyllaInformers.Scylla().V1().ScyllaClusters(),
 		o.OperatorImage,
 		o.CQLSIngressPort,

pkg/controller/scyllacluster/conditions.go

@@ -22,4 +22,8 @@ const (
 	jobControllerDegradedCondition               = "JobControllerDegraded"
 	configControllerProgressingCondition         = "ConfigControllerProgressing"
 	configControllerDegradedCondition            = "ConfigControllerDegraded"
+	endpointSliceControllerProgressingCondition  = "EndpointSliceControllerProgressing"
+	endpointSliceControllerDegradedCondition     = "EndpointSliceControllerDegraded"
+	endpointsControllerProgressingCondition      = "EndpointsControllerProgressing"
+	endpointsControllerDegradedCondition         = "EndpointsControllerDegraded"
 )

pkg/controller/scyllacluster/controller.go

@@ -17,6 +17,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -29,6 +30,7 @@ import (
 	appsv1informers "k8s.io/client-go/informers/apps/v1"
 	batchv1informers "k8s.io/client-go/informers/batch/v1"
 	corev1informers "k8s.io/client-go/informers/core/v1"
+	discoveryv1informers "k8s.io/client-go/informers/discovery/v1"
 	networkingv1informers "k8s.io/client-go/informers/networking/v1"
 	policyv1informers "k8s.io/client-go/informers/policy/v1"
 	rbacv1informers "k8s.io/client-go/informers/rbac/v1"
@@ -37,6 +39,7 @@ import (
 	appsv1listers "k8s.io/client-go/listers/apps/v1"
 	batchv1listers "k8s.io/client-go/listers/batch/v1"
 	corev1listers "k8s.io/client-go/listers/core/v1"
+	discoveryv1listers "k8s.io/client-go/listers/discovery/v1"
 	networkingv1listers "k8s.io/client-go/listers/networking/v1"
 	policyv1listers "k8s.io/client-go/listers/policy/v1"
 	rbacv1listers "k8s.io/client-go/listers/rbac/v1"
@@ -76,6 +79,8 @@ type Controller struct {
 	ingressLister        networkingv1listers.IngressLister
 	scyllaLister         scyllav1listers.ScyllaClusterLister
 	jobLister            batchv1listers.JobLister
+	endpointSliceLister  discoveryv1listers.EndpointSliceLister
+	endpointsLister      corev1listers.EndpointsLister
 
 	cachesToSync []cache.InformerSynced
 
@@ -100,6 +105,8 @@ func NewController(
 	pdbInformer policyv1informers.PodDisruptionBudgetInformer,
 	ingressInformer networkingv1informers.IngressInformer,
 	jobInformer batchv1informers.JobInformer,
+	endpointSliceInformer discoveryv1informers.EndpointSliceInformer,
+	endpointsInformer corev1informers.EndpointsInformer,
 	scyllaClusterInformer scyllav1informers.ScyllaClusterInformer,
 	operatorImage string,
 	cqlsIngressPort int,
@@ -127,6 +134,8 @@ func NewController(
 		ingressLister:        ingressInformer.Lister(),
 		scyllaLister:         scyllaClusterInformer.Lister(),
 		jobLister:            jobInformer.Lister(),
+		endpointSliceLister:  endpointSliceInformer.Lister(),
+		endpointsLister:      endpointsInformer.Lister(),
 
 		cachesToSync: []cache.InformerSynced{
 			podInformer.Informer().HasSynced,
@@ -140,6 +149,8 @@ func NewController(
 			ingressInformer.Informer().HasSynced,
 			scyllaClusterInformer.Informer().HasSynced,
 			jobInformer.Informer().HasSynced,
+			endpointSliceInformer.Informer().HasSynced,
+			endpointsInformer.Informer().HasSynced,
 		},
 
 		eventRecorder: eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "scyllacluster-controller"}),
@@ -235,6 +246,18 @@ func NewController(
 		DeleteFunc: scc.deleteJob,
 	})
 
+	endpointSliceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    scc.addEndpointSlice,
+		UpdateFunc: scc.updateEndpointSlice,
+		DeleteFunc: scc.deleteEndpointSlice,
+	})
+
+	endpointsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    scc.addEndpoints,
+		UpdateFunc: scc.updateEndpoints,
+		DeleteFunc: scc.deleteEndpoints,
+	})
+
 	return scc, nil
 }
 
@@ -625,3 +648,49 @@ func (scc *Controller) deleteJob(obj interface{}) {
 		scc.handlers.EnqueueOwner,
 	)
 }
+
+func (scc *Controller) addEndpointSlice(obj interface{}) {
+	scc.handlers.HandleAdd(
+		obj.(*discoveryv1.EndpointSlice),
+		scc.handlers.EnqueueOwner,
+	)
+}
+
+func (scc *Controller) updateEndpointSlice(old, cur interface{}) {
+	scc.handlers.HandleUpdate(
+		old.(*discoveryv1.EndpointSlice),
+		cur.(*discoveryv1.EndpointSlice),
+		scc.handlers.EnqueueOwner,
+		scc.deleteEndpointSlice,
+	)
+}
+
+func (scc *Controller) deleteEndpointSlice(obj interface{}) {
+	scc.handlers.HandleDelete(
+		obj,
+		scc.handlers.EnqueueOwner,
+	)
+}
+
+func (scc *Controller) addEndpoints(obj interface{}) {
+	scc.handlers.HandleAdd(
+		obj.(*corev1.Endpoints),
+		scc.handlers.EnqueueOwner,
+	)
+}
+
+func (scc *Controller) updateEndpoints(old, cur interface{}) {
+	scc.handlers.HandleUpdate(
+		old.(*corev1.Endpoints),
+		cur.(*corev1.Endpoints),
+		scc.handlers.EnqueueOwner,
+		scc.deleteEndpoints,
+	)
+}
+
+func (scc *Controller) deleteEndpoints(obj interface{}) {
+	scc.handlers.HandleDelete(
+		obj,
+		scc.handlers.EnqueueOwner,
+	)
+}