Better support for dual-stack Kubernetes clusters (#671)

Author: thegridman

Makefile

@@ -1650,6 +1650,11 @@ kind:   ## Run a default KinD cluster
 	kind create cluster --name $(KIND_CLUSTER) --wait 10m --config $(SCRIPTS_DIR)/kind-config.yaml --image $(KIND_IMAGE)
 	$(SCRIPTS_DIR)/kind-label-node.sh
 
+.PHONY: kind-dual
+kind-dual:   ## Run a KinD cluster configured for a dual stack IPv4 and IPv6 network
+	kind create cluster --name $(KIND_CLUSTER) --wait 10m --config $(SCRIPTS_DIR)/kind-config-dual.yaml --image $(KIND_IMAGE)
+	$(SCRIPTS_DIR)/kind-label-node.sh
+
 # ----------------------------------------------------------------------------------------------------------------------
 # Start a Kind cluster
 # ----------------------------------------------------------------------------------------------------------------------

api/v1/coherence_types.go

@@ -562,14 +562,22 @@ func (in *CoherenceSpec) GetManagementPort() int32 {
 	}
 }
 
-// GetPersistenceSpec returns the Coherence persistence spcification.
+// GetPersistenceSpec returns the Coherence persistence specification.
 func (in *CoherenceSpec) GetPersistenceSpec() *PersistenceSpec {
 	if in == nil {
 		return nil
 	}
 	return in.Persistence
 }
 
+// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+func (in *CoherenceSpec) GetWkaIPFamily() corev1.IPFamily {
+	if in == nil || in.WKA == nil || in.WKA.IPFamily == nil {
+		return corev1.IPFamilyUnknown
+	}
+	return *in.WKA.IPFamily
+}
+
 // ----- CoherenceWKASpec struct --------------------------------------------
 
 // CoherenceWKASpec configures Coherence well-known-addressing to use an
@@ -599,6 +607,11 @@ type CoherenceWKASpec struct {
 	// More info: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
 	// +optional
 	Annotations map[string]string `json:"annotations,omitempty" protobuf:"bytes,12,rep,name=annotations"`
+
+	// IPFamily is the IP family to use for the WKA service (and also the StatefulSet headless service).
+	// Valid values are "IPv4" or "IPv6".
+	// +optional
+	IPFamily *corev1.IPFamily `json:"ipFamily,omitempty"`
 }
 
 // ----- CoherenceTracingSpec struct ----------------------------------------

api/v1/coherencejobresource_types.go

@@ -334,6 +334,19 @@ func (in *CoherenceJob) IsBeforeVersion(version string) bool {
 	return true
 }
 
+// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+func (in *CoherenceJob) GetWkaIPFamily() corev1.IPFamily {
+	if in == nil {
+		return corev1.IPFamilyUnknown
+	}
+	return in.Spec.GetWkaIPFamily()
+}
+
+// GetHeadlessServiceIPFamily always returns an empty array as this is not applicable to Jobs.
+func (in *CoherenceJob) GetHeadlessServiceIPFamily() []corev1.IPFamily {
+	return nil
+}
+
 // ----- CoherenceJobList type ----------------------------------------------
 
 // CoherenceJobResourceSpec defines the specification of a CoherenceJob resource.
@@ -487,6 +500,14 @@ func (in *CoherenceJobResourceSpec) GetReplicas() int32 {
 	return *in.CoherenceResourceSpec.Replicas
 }
 
+// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+func (in *CoherenceJobResourceSpec) GetWkaIPFamily() corev1.IPFamily {
+	if in == nil || in.Coherence == nil {
+		return corev1.IPFamilyUnknown
+	}
+	return in.Coherence.GetWkaIPFamily()
+}
+
 // UpdateJob updates a JobSpec from the fields in this spec
 func (in *CoherenceJobResourceSpec) UpdateJob(spec *batchv1.JobSpec) {
 	if in == nil {

api/v1/coherenceresource.go

@@ -21,8 +21,12 @@ type CoherenceResource interface {
 	GetCoherenceClusterName() string
 	// GetWkaServiceName returns the name of the headless Service used for Coherence WKA.
 	GetWkaServiceName() string
+	// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+	GetWkaIPFamily() corev1.IPFamily
 	// GetHeadlessServiceName returns the name of the headless Service used for the StatefulSet.
 	GetHeadlessServiceName() string
+	// GetHeadlessServiceIPFamily always returns an empty array as this is not applicable to Jobs.
+	GetHeadlessServiceIPFamily() []corev1.IPFamily
 	// GetReplicas returns the number of replicas required for a deployment.
 	// The Replicas field is a pointer and may be nil so this method will
 	// return either the actual Replicas value or the default (DefaultReplicas const)

api/v1/coherenceresource_types.go

@@ -150,6 +150,11 @@ func (in *Coherence) GetWkaServiceName() string {
 	return in.Name + WKAServiceNameSuffix
 }
 
+// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+func (in *Coherence) GetWkaIPFamily() corev1.IPFamily {
+	return in.Spec.GetWkaIPFamily()
+}
+
 // GetHeadlessServiceName returns the name of the headless Service used for the StatefulSet.
 func (in *Coherence) GetHeadlessServiceName() string {
 	if in == nil {
@@ -158,6 +163,14 @@ func (in *Coherence) GetHeadlessServiceName() string {
 	return in.Name + HeadlessServiceNameSuffix
 }
 
+// GetHeadlessServiceIPFamily returns the IP Family of the headless Service used for the StatefulSet.
+func (in *Coherence) GetHeadlessServiceIPFamily() []corev1.IPFamily {
+	if in == nil {
+		return nil
+	}
+	return in.Spec.HeadlessServiceIpFamilies
+}
+
 // GetReplicas returns the number of replicas required for a deployment.
 // The Replicas field is a pointer and may be nil so this method will
 // return either the actual Replicas value or the default (DefaultReplicas const)
@@ -527,6 +540,10 @@ type CoherenceStatefulSetResourceSpec struct {
 	// one of the node labels used to set the Coherence site or rack value.
 	// +optional
 	RollingUpdateLabel *string `json:"rollingUpdateLabel,omitempty"`
+	// HeadlessServiceIpFamilies is the optional array of IP families that can be configured for
+	// the headless service used for the StatefulSet.
+	// +optional
+	HeadlessServiceIpFamilies []corev1.IPFamily `json:"headlessServiceIpFamilies,omitempty"`
 }
 
 // RollingUpdateStrategyType is a string enumeration type that enumerates

api/v1/coherenceresourcespec_types.go

@@ -331,6 +331,14 @@ func (in *CoherenceResourceSpec) SetReplicas(replicas int32) {
 	}
 }
 
+// GetWkaIPFamily returns the IP Family of the headless Service used for Coherence WKA.
+func (in *CoherenceResourceSpec) GetWkaIPFamily() corev1.IPFamily {
+	if in == nil || in.Coherence == nil {
+		return corev1.IPFamilyUnknown
+	}
+	return in.Coherence.GetWkaIPFamily()
+}
+
 // GetRestartPolicy returns the name of the application image to use
 func (in *CoherenceResourceSpec) GetRestartPolicy() *corev1.RestartPolicy {
 	if in == nil {
@@ -553,6 +561,12 @@ func (in *CoherenceResourceSpec) CreateWKAService(deployment CoherenceResource)
 		},
 	}
 
+	ip := deployment.GetWkaIPFamily()
+	if ip != corev1.IPFamilyUnknown {
+		svc.Spec.IPFamilyPolicy = ptr.To(corev1.IPFamilyPolicySingleStack)
+		svc.Spec.IPFamilies = []corev1.IPFamily{ip}
+	}
+
 	return Resource{
 		Kind: ResourceTypeService,
 		Name: svc.GetName(),
@@ -592,6 +606,15 @@ func (in *CoherenceResourceSpec) CreateHeadlessService(deployment CoherenceResou
 		},
 	}
 
+	ipFamilies := deployment.GetHeadlessServiceIPFamily()
+	if len(ipFamilies) == 1 {
+		svc.Spec.IPFamilyPolicy = ptr.To(corev1.IPFamilyPolicySingleStack)
+		svc.Spec.IPFamilies = ipFamilies
+	} else if len(ipFamilies) > 1 {
+		svc.Spec.IPFamilyPolicy = ptr.To(corev1.IPFamilyPolicyPreferDualStack)
+		svc.Spec.IPFamilies = ipFamilies
+	}
+
 	return Resource{
 		Kind: ResourceTypeService,
 		Name: svc.GetName(),

api/v1/create_job_wka_services_test.go

@@ -12,10 +12,11 @@ import (
 	coh "github.com/oracle/coherence-operator/api/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 	"testing"
 )
 
-func TestCreateWKAServiceForMinimalJonDeployment(t *testing.T) {
+func TestCreateWKAServiceForMinimalJsonDeployment(t *testing.T) {
 	// Create the test deployment
 	deployment := &coh.CoherenceJob{
 		ObjectMeta: metav1.ObjectMeta{
@@ -315,6 +316,60 @@ func TestCreateWKAServiceForJobWithAdditionalAnnotations(t *testing.T) {
 	assertWKAServiceForJob(t, deployment, expected)
 }
 
+func TestCreateJobWKAServiceWithIPFamily(t *testing.T) {
+	// Create the test deployment
+	deployment := &coh.CoherenceJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "test-ns",
+			Name:      "test",
+		},
+		Spec: coh.CoherenceJobResourceSpec{
+			CoherenceResourceSpec: coh.CoherenceResourceSpec{
+				Coherence: &coh.CoherenceSpec{
+					WKA: &coh.CoherenceWKASpec{
+						IPFamily: ptr.To(corev1.IPv4Protocol),
+					},
+				},
+			},
+			Cluster: "test-cluster",
+		},
+	}
+
+	// create the expected WKA service
+	labels := deployment.CreateCommonLabels()
+	labels[coh.LabelCoherenceCluster] = "test-cluster"
+	labels[coh.LabelComponent] = coh.LabelComponentWKA
+
+	// The selector for the service (match all Pods with the same cluster label)
+	selector := make(map[string]string)
+	selector[coh.LabelCoherenceCluster] = "test-cluster"
+	selector[coh.LabelComponent] = coh.LabelComponentCoherencePod
+	selector[coh.LabelCoherenceWKAMember] = "true"
+
+	expected := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "test-ns",
+			Name:      "test-wka",
+			Labels:    labels,
+			Annotations: map[string]string{
+				"service.alpha.kubernetes.io/tolerate-unready-endpoints": "true",
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			ClusterIP: corev1.ClusterIPNone,
+			// Pods must be part of the WKA service even if not ready
+			PublishNotReadyAddresses: true,
+			Ports:                    getDefaultServicePorts(),
+			Selector:                 selector,
+			IPFamilyPolicy:           ptr.To(corev1.IPFamilyPolicySingleStack),
+			IPFamilies:               []corev1.IPFamily{corev1.IPv4Protocol},
+		},
+	}
+
+	// assert that the Services are as expected
+	assertWKAServiceForJob(t, deployment, expected)
+}
+
 func assertWKAServiceForJob(t *testing.T, deployment *coh.CoherenceJob, expected *corev1.Service) {
 	g := NewGomegaWithT(t)