Issue 61: Support external connectivity

pkg/pravega/bookie.go

@@ -17,7 +17,7 @@ import (
 	"github.com/pravega/pravega-operator/pkg/apis/pravega/v1alpha1"
 	"github.com/pravega/pravega-operator/pkg/utils/k8sutil"
 	"github.com/sirupsen/logrus"
-	"k8s.io/api/apps/v1beta1"
+	"k8s.io/api/apps/v1beta2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -84,11 +84,11 @@ func makeBookieHeadlessService(pravegaCluster *v1alpha1.PravegaCluster) *corev1.
 	}
 }
 
-func makeBookieStatefulSet(pravegaCluster *v1alpha1.PravegaCluster) *v1beta1.StatefulSet {
-	return &v1beta1.StatefulSet{
+func makeBookieStatefulSet(pravegaCluster *v1alpha1.PravegaCluster) *v1beta2.StatefulSet {
+	return &v1beta2.StatefulSet{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "StatefulSet",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1beta2",
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      k8sutil.StatefulSetNameForBookie(pravegaCluster.Name),
@@ -98,11 +98,14 @@ func makeBookieStatefulSet(pravegaCluster *v1alpha1.PravegaCluster) *v1beta1.Sta
 				*k8sutil.AsOwnerRef(pravegaCluster),
 			},
 		},
-		Spec: v1beta1.StatefulSetSpec{
-			ServiceName:          k8sutil.HeadlessServiceNameForBookie(pravegaCluster.Name),
-			Replicas:             &pravegaCluster.Spec.Bookkeeper.Replicas,
-			PodManagementPolicy:  v1beta1.ParallelPodManagement,
-			Template:             makeBookieStatefulTemplate(pravegaCluster),
+		Spec: v1beta2.StatefulSetSpec{
+			ServiceName:         k8sutil.HeadlessServiceNameForBookie(pravegaCluster.Name),
+			Replicas:            &pravegaCluster.Spec.Bookkeeper.Replicas,
+			PodManagementPolicy: v1beta2.ParallelPodManagement,
+			Template:            makeBookieStatefulTemplate(pravegaCluster),
+			Selector: &metav1.LabelSelector{
+				MatchLabels: k8sutil.LabelsForBookie(pravegaCluster),
+			},
 			VolumeClaimTemplates: makeBookieVolumeClaimTemplates(&pravegaCluster.Spec.Bookkeeper),
 		},
 	}
@@ -157,6 +160,29 @@ func makeBookiePodSpec(clusterName string, bookkeeperSpec *v1alpha1.BookkeeperSp
 				},
 			},
 		},
+		Affinity: &corev1.Affinity{
+			PodAntiAffinity: &corev1.PodAntiAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{
+							MatchExpressions: []metav1.LabelSelectorRequirement{
+								{
+									Key:      "component",
+									Operator: metav1.LabelSelectorOpIn,
+									Values:   []string{"bookie"},
+								},
+								{
+									Key:      "pravega_cluster",
+									Operator: metav1.LabelSelectorOpIn,
+									Values:   []string{clusterName},
+								},
+							},
+						},
+						TopologyKey: "kubernetes.io/hostname",
+					},
+				},
+			},
+		},
 	}
 }
 

pkg/pravega/pravega_controller.go

@@ -18,7 +18,7 @@ import (
 	"github.com/operator-framework/operator-sdk/pkg/sdk"
 	api "github.com/pravega/pravega-operator/pkg/apis/pravega/v1alpha1"
 	"github.com/pravega/pravega-operator/pkg/utils/k8sutil"
-	"k8s.io/api/apps/v1beta1"
+	"k8s.io/api/apps/v1beta2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -43,11 +43,11 @@ func deployController(pravegaCluster *api.PravegaCluster) (err error) {
 	return nil
 }
 
-func makeControllerDeployment(pravegaCluster *api.PravegaCluster) *v1beta1.Deployment {
-	return &v1beta1.Deployment{
+func makeControllerDeployment(pravegaCluster *api.PravegaCluster) *v1beta2.Deployment {
+	return &v1beta2.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Deployment",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1beta2",
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      k8sutil.DeploymentNameForController(pravegaCluster.Name),
@@ -56,14 +56,17 @@ func makeControllerDeployment(pravegaCluster *api.PravegaCluster) *v1beta1.Deplo
 				*k8sutil.AsOwnerRef(pravegaCluster),
 			},
 		},
-		Spec: v1beta1.DeploymentSpec{
+		Spec: v1beta2.DeploymentSpec{
 			Replicas: &pravegaCluster.Spec.Pravega.ControllerReplicas,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: k8sutil.LabelsForController(pravegaCluster),
 				},
 				Spec: *makeControllerPodSpec(pravegaCluster.Name, &pravegaCluster.Spec.Pravega),
 			},
+			Selector: &metav1.LabelSelector{
+				MatchLabels: k8sutil.LabelsForController(pravegaCluster),
+			},
 		},
 	}
 }

pkg/pravega/pravega_segmentstore.go

@@ -19,7 +19,7 @@ import (
 	api "github.com/pravega/pravega-operator/pkg/apis/pravega/v1alpha1"
 	"github.com/pravega/pravega-operator/pkg/utils/k8sutil"
 	"github.com/sirupsen/logrus"
-	"k8s.io/api/apps/v1beta1"
+	"k8s.io/api/apps/v1beta2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -44,6 +44,14 @@ func deploySegmentStore(pravegaCluster *api.PravegaCluster) (err error) {
 		return err
 	}
 
+	services := makeSegmentStoreServices(pravegaCluster)
+	for _, service := range services {
+		err = sdk.Create(service)
+		if err != nil && !errors.IsAlreadyExists(err) {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -56,29 +64,35 @@ func destroySegmentstoreCacheVolumes(metadata metav1.ObjectMeta) {
 	}
 }
 
-func makeSegmentStoreStatefulSet(pravegaCluster *api.PravegaCluster) *v1beta1.StatefulSet {
-	return &v1beta1.StatefulSet{
+func makeSegmentStoreStatefulSet(pravegaCluster *api.PravegaCluster) *v1beta2.StatefulSet {
+	return &v1beta2.StatefulSet{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "StatefulSet",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1beta2",
 		},
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      k8sutil.StatefulSetNameForSegmentstore(pravegaCluster.Name),
 			Namespace: pravegaCluster.Namespace,
 			OwnerReferences: []metav1.OwnerReference{
 				*k8sutil.AsOwnerRef(pravegaCluster),
 			},
+			Annotations: map[string]string{
+				"service-per-pod-label": v1beta2.StatefulSetPodNameLabel,
+			},
 		},
-		Spec: v1beta1.StatefulSetSpec{
+		Spec: v1beta2.StatefulSetSpec{
 			ServiceName:         "segmentstore",
 			Replicas:            &pravegaCluster.Spec.Pravega.SegmentStoreReplicas,
-			PodManagementPolicy: v1beta1.ParallelPodManagement,
+			PodManagementPolicy: v1beta2.ParallelPodManagement,
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
 					Labels: k8sutil.LabelsForSegmentStore(pravegaCluster),
 				},
 				Spec: makeSegmentstorePodSpec(pravegaCluster),
 			},
+			Selector: &metav1.LabelSelector{
+				MatchLabels: k8sutil.LabelsForSegmentStore(pravegaCluster),
+			},
 			VolumeClaimTemplates: makeCacheVolumeClaimTemplate(&pravegaCluster.Spec.Pravega),
 		},
 	}
@@ -123,6 +137,29 @@ func makeSegmentstorePodSpec(pravegaCluster *api.PravegaCluster) corev1.PodSpec
 				},
 			},
 		},
+		Affinity: &corev1.Affinity{
+			PodAntiAffinity: &corev1.PodAntiAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{
+							MatchExpressions: []metav1.LabelSelectorRequirement{
+								{
+									Key:      "component",
+									Operator: metav1.LabelSelectorOpIn,
+									Values:   []string{"pravega-segmentstore"},
+								},
+								{
+									Key:      "pravega_cluster",
+									Operator: metav1.LabelSelectorOpIn,
+									Values:   []string{pravegaCluster.Name},
+								},
+							},
+						},
+						TopologyKey: "kubernetes.io/hostname",
+					},
+				},
+			},
+		},
 	}
 
 	configureTier2Filesystem(&podSpec, &pravegaSpec)
@@ -245,3 +282,40 @@ func configureTier2Filesystem(podSpec *corev1.PodSpec, pravegaSpec *api.PravegaS
 		})
 	}
 }
+
+func makeSegmentStoreServices(pravegaCluster *api.PravegaCluster) []*corev1.Service {
+	var service *corev1.Service
+	services := make([]*corev1.Service, pravegaCluster.Spec.Pravega.SegmentStoreReplicas)
+
+	for i := int32(0); i < pravegaCluster.Spec.Pravega.SegmentStoreReplicas; i++ {
+		service = &corev1.Service{
+			TypeMeta: metav1.TypeMeta{
+				Kind:       "Service",
+				APIVersion: "v1",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      k8sutil.ServiceNameForSegmentStore(pravegaCluster.Name, i),
+				Namespace: pravegaCluster.Namespace,
+				Labels:    k8sutil.LabelsForSegmentStore(pravegaCluster),
+				OwnerReferences: []metav1.OwnerReference{
+					*k8sutil.AsOwnerRef(pravegaCluster),
+				},
+			},
+			Spec: corev1.ServiceSpec{
+				Type:                  corev1.ServiceTypeNodePort,
+				ExternalTrafficPolicy: corev1.ServiceExternalTrafficPolicyTypeLocal,
+				Ports: []corev1.ServicePort{
+					{
+						Name: "wire",
+						Port: 12345,
+					},
+				},
+				Selector: map[string]string{
+					v1beta2.StatefulSetPodNameLabel: fmt.Sprintf("%s-%d", k8sutil.StatefulSetNameForSegmentstore(pravegaCluster.Name), i),
+				},
+			},
+		}
+		services[i] = service
+	}
+	return services
+}

pkg/utils/k8sutil/pravega.go

@@ -32,6 +32,10 @@ func ServiceNameForController(clusterName string) string {
 	return fmt.Sprintf("%s-pravega-controller", clusterName)
 }
 
+func ServiceNameForSegmentStore(clusterName string, index int32) string {
+	return fmt.Sprintf("%s-pravega-segmentstore-%d", clusterName, index)
+}
+
 func HeadlessServiceNameForBookie(clusterName string) string {
 	return fmt.Sprintf("%s-bookie-headless", clusterName)
 }