Issue 61: Support external connectivity

deploy/rbac.yaml

@@ -33,6 +33,17 @@ rules:
 
 ---
 
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: pravega-operator
+rules:
+- apiGroups: [""]
+  resources: ["nodes"]
+  verbs: ["get", "watch", "list"]
+
+---
+
 kind: RoleBinding
 apiVersion: rbac.authorization.k8s.io/v1beta1
 metadata:
@@ -44,3 +55,17 @@ roleRef:
   kind: Role
   name: pravega-operator
   apiGroup: rbac.authorization.k8s.io
+
+---
+
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: default-account-pravega-operator
+subjects:
+- kind: ServiceAccount
+  name: default
+roleRef:
+  kind: ClusterRole
+  name: pravega-operator
+  apiGroup: rbac.authorization.k8s.io

pkg/apis/pravega/v1alpha1/types.go

@@ -33,9 +33,15 @@ type PravegaCluster struct {
 }
 
 type PravegaClusterSpec struct {
-	ZookeeperUri string         `json:"zookeeperUri"`
-	Bookkeeper   BookkeeperSpec `json:"bookkeeper"`
-	Pravega      PravegaSpec    `json:"pravega"`
+	ZookeeperUri   string         `json:"zookeeperUri"`
+	ExternalAccess ExternalAccess `json:"externalAccess"`
+	Bookkeeper     BookkeeperSpec `json:"bookkeeper"`
+	Pravega        PravegaSpec    `json:"pravega"`
+}
+
+type ExternalAccess struct {
+	Enabled bool           `json:"enabled"`
+	Type    v1.ServiceType `json:"type,omitempty"`
 }
 
 type ImageSpec struct {

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"
+	appsv1 "k8s.io/api/apps/v1"
 	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) *appsv1.StatefulSet {
+	return &appsv1.StatefulSet{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "StatefulSet",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1",
 		},
 		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: appsv1.StatefulSetSpec{
+			ServiceName:         k8sutil.HeadlessServiceNameForBookie(pravegaCluster.Name),
+			Replicas:            &pravegaCluster.Spec.Bookkeeper.Replicas,
+			PodManagementPolicy: appsv1.ParallelPodManagement,
+			Template:            makeBookieStatefulTemplate(pravegaCluster),
+			Selector: &metav1.LabelSelector{
+				MatchLabels: k8sutil.LabelsForBookie(pravegaCluster),
+			},
 			VolumeClaimTemplates: makeBookieVolumeClaimTemplates(&pravegaCluster.Spec.Bookkeeper),
 		},
 	}

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"
+	appsv1 "k8s.io/api/apps/v1"
 	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) *appsv1.Deployment {
+	return &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Deployment",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1",
 		},
 		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: appsv1.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),
+			},
 		},
 	}
 }
@@ -146,6 +149,12 @@ func makeControllerConfigMap(pravegaCluster *api.PravegaCluster) *corev1.ConfigM
 }
 
 func makeControllerService(pravegaCluster *api.PravegaCluster) *corev1.Service {
+
+	serviceType := corev1.ServiceTypeClusterIP
+	if pravegaCluster.Spec.ExternalAccess.Enabled {
+		serviceType = pravegaCluster.Spec.ExternalAccess.Type
+	}
+
 	return &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Service",
@@ -160,6 +169,7 @@ func makeControllerService(pravegaCluster *api.PravegaCluster) *corev1.Service {
 			},
 		},
 		Spec: corev1.ServiceSpec{
+			Type: serviceType,
 			Ports: []corev1.ServicePort{
 				{
 					Name: "rest",

pkg/pravega/pravega_segmentstore.go

@@ -19,10 +19,11 @@ 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"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 const (
@@ -34,6 +35,21 @@ const (
 )
 
 func deploySegmentStore(pravegaCluster *api.PravegaCluster) (err error) {
+	err = sdk.Create(makeSegmentStoreHeadlessService(pravegaCluster))
+	if err != nil && !errors.IsAlreadyExists(err) {
+		return err
+	}
+
+	if pravegaCluster.Spec.ExternalAccess.Enabled {
+		services := makeSegmentStoreExternalServices(pravegaCluster)
+		for _, service := range services {
+			err = sdk.Create(service)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+	}
+
 	err = sdk.Create(makeSegmentstoreConfigMap(pravegaCluster))
 	if err != nil && !errors.IsAlreadyExists(err) {
 		return err
@@ -56,29 +72,35 @@ func destroySegmentstoreCacheVolumes(metadata metav1.ObjectMeta) {
 	}
 }
 
-func makeSegmentStoreStatefulSet(pravegaCluster *api.PravegaCluster) *v1beta1.StatefulSet {
-	return &v1beta1.StatefulSet{
+func makeSegmentStoreStatefulSet(pravegaCluster *api.PravegaCluster) *appsv1.StatefulSet {
+	return &appsv1.StatefulSet{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "StatefulSet",
-			APIVersion: "apps/v1beta1",
+			APIVersion: "apps/v1",
 		},
 		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": appsv1.StatefulSetPodNameLabel,
+			},
 		},
-		Spec: v1beta1.StatefulSetSpec{
-			ServiceName:         "segmentstore",
+		Spec: appsv1.StatefulSetSpec{
+			ServiceName:         "pravega-segmentstore",
 			Replicas:            &pravegaCluster.Spec.Pravega.SegmentStoreReplicas,
-			PodManagementPolicy: v1beta1.ParallelPodManagement,
+			PodManagementPolicy: appsv1.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),
 		},
 	}
@@ -102,7 +124,7 @@ func makeSegmentstorePodSpec(pravegaCluster *api.PravegaCluster) corev1.PodSpec
 	podSpec := corev1.PodSpec{
 		Containers: []corev1.Container{
 			{
-				Name:            "segmentstore",
+				Name:            "pravega-segmentstore",
 				Image:           pravegaSpec.Image.String(),
 				ImagePullPolicy: pravegaSpec.Image.PullPolicy,
 				Args: []string{
@@ -115,6 +137,7 @@ func makeSegmentstorePodSpec(pravegaCluster *api.PravegaCluster) corev1.PodSpec
 					},
 				},
 				EnvFrom: environment,
+				Env:     k8sutil.DownwardAPIEnv(),
 				VolumeMounts: []corev1.VolumeMount{
 					{
 						Name:      cacheVolumeName,
@@ -148,6 +171,10 @@ func makeSegmentstoreConfigMap(pravegaCluster *api.PravegaCluster) *corev1.Confi
 		"WAIT_FOR":              pravegaCluster.Spec.ZookeeperUri,
 	}
 
+	if pravegaCluster.Spec.ExternalAccess.Enabled {
+		configData["K8_EXTERNAL_ACCESS"] = "true"
+	}
+
 	if pravegaCluster.Spec.Pravega.DebugLogging {
 		configData["log.level"] = "DEBUG"
 	}
@@ -245,3 +272,70 @@ func configureTier2Filesystem(podSpec *corev1.PodSpec, pravegaSpec *api.PravegaS
 		})
 	}
 }
+
+func makeSegmentStoreHeadlessService(pravegaCluster *api.PravegaCluster) *corev1.Service {
+	return &corev1.Service{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Service",
+			APIVersion: "v1",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      k8sutil.HeadlessServiceNameForSegmentStore(pravegaCluster.Name),
+			Namespace: pravegaCluster.Namespace,
+			Labels:    k8sutil.LabelsForSegmentStore(pravegaCluster),
+			OwnerReferences: []metav1.OwnerReference{
+				*k8sutil.AsOwnerRef(pravegaCluster),
+			},
+		},
+		Spec: corev1.ServiceSpec{
+			Ports: []corev1.ServicePort{
+				{
+					Name:     "server",
+					Port:     12345,
+					Protocol: "TCP",
+				},
+			},
+			Selector:  k8sutil.LabelsForSegmentStore(pravegaCluster),
+			ClusterIP: corev1.ClusterIPNone,
+		},
+	}
+}
+
+func makeSegmentStoreExternalServices(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: pravegaCluster.Spec.ExternalAccess.Type,
+				ExternalTrafficPolicy: corev1.ServiceExternalTrafficPolicyTypeLocal,
+				Ports: []corev1.ServicePort{
+					{
+						Name:       "server",
+						Port:       12345,
+						Protocol:   "TCP",
+						TargetPort: intstr.FromInt(12345),
+					},
+				},
+				Selector: map[string]string{
+					appsv1.StatefulSetPodNameLabel: fmt.Sprintf("%s-%d", k8sutil.StatefulSetNameForSegmentstore(pravegaCluster.Name), i),
+				},
+			},
+		}
+		services[i] = service
+	}
+	return services
+}

pkg/utils/k8sutil/k8sutil.go

@@ -18,6 +18,7 @@ import (
 	"github.com/operator-framework/operator-sdk/pkg/k8sclient"
 	"github.com/operator-framework/operator-sdk/pkg/util/k8sutil"
 	api "github.com/pravega/pravega-operator/pkg/apis/pravega/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -51,3 +52,26 @@ func GetWatchNamespaceAllowBlank() (string, error) {
 	}
 	return ns, nil
 }
+
+func DownwardAPIEnv() []corev1.EnvVar {
+	return []corev1.EnvVar{
+		{
+			Name: "POD_NAME",
+			ValueFrom: &corev1.EnvVarSource{
+				FieldRef: &corev1.ObjectFieldSelector{
+					APIVersion: "v1",
+					FieldPath:  "metadata.name",
+				},
+			},
+		},
+		{
+			Name: "POD_NAMESPACE",
+			ValueFrom: &corev1.EnvVarSource{
+				FieldRef: &corev1.ObjectFieldSelector{
+					APIVersion: "v1",
+					FieldPath:  "metadata.namespace",
+				},
+			},
+		},
+	}
+}