Issue 534: Provide support to configure segment store port

charts/pravega/values.yaml

@@ -135,6 +135,7 @@ options:
   bookkeeper.write.outstanding.bytes.max: "33554432"
   pravegaservice.cache.size.max: "1073741824"
   pravegaservice.cache.time.seconds.max: "600"
+  pravegaservice.service.listener.port: "12345"
   hdfs.block.size: "67108864"
   writer.flush.threshold.bytes: "67108864"
   writer.flush.size.bytes.max: "67108864"

go.mod

@@ -3,7 +3,9 @@ module github.com/pravega/pravega-operator
 go 1.13
 
 require (
+	github.com/alexkohler/nakedret v1.0.0 // indirect
 	github.com/hashicorp/go-version v1.1.0
+	github.com/jgautheron/goconst v1.4.0 // indirect
 	github.com/onsi/ginkgo v1.12.0
 	github.com/onsi/gomega v1.9.0
 	github.com/operator-framework/operator-sdk v0.17.0

go.sum

@@ -85,12 +85,15 @@ github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWX
 github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/StackExchange/wmi v0.0.0-20180116203802-5d049714c4a6/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/agnivade/levenshtein v1.0.1/go.mod h1:CURSv5d9Uaml+FovSIICkLbAUZ9S4RqaHDIsdSBg7lM=
+github.com/alecthomas/gocyclo v0.0.0-20150208221726-aa8f8b160214 h1:YI/8G3uLbYyowJeOPVL6BMKe2wbL54h0FdEKmncU6lU=
 github.com/alecthomas/gocyclo v0.0.0-20150208221726-aa8f8b160214/go.mod h1:Ef5UOtJdJ5rVFObdOVsrNgKV/Wf4I+daTCSk8GTrHIk=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/alexkohler/nakedret v1.0.0 h1:S/bzOFhZHYUJp6qPmdXdFHS5nlWGFmLmoc8QOydvotE=
+github.com/alexkohler/nakedret v1.0.0/go.mod h1:tfDQbtPt67HhBK/6P0yNktIX7peCxfOp0jO9007DrLE=
 github.com/aliyun/aliyun-oss-go-sdk v2.0.4+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
 github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
@@ -541,6 +544,8 @@ github.com/jackc/pgx v3.2.0+incompatible/go.mod h1:0ZGrqGqkRlliWnWB4zKnWtjbSWbGk
 github.com/jellevandenhooff/dkim v0.0.0-20150330215556-f50fe3d243e1/go.mod h1:E0B/fFc00Y+Rasa88328GlI/XbtyysCtTHZS8h7IrBU=
 github.com/jessevdk/go-flags v0.0.0-20180331124232-1c38ed7ad0cc/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
+github.com/jgautheron/goconst v1.4.0 h1:hp9XKUpe/MPyDamUbfsrGpe+3dnY2whNK4EtB86dvLM=
+github.com/jgautheron/goconst v1.4.0/go.mod h1:aAosetZ5zaeC/2EfMeRswtxUFBpe2Hr7HzkgX4fanO4=
 github.com/jimstudt/http-authentication v0.0.0-20140401203705-3eca13d6893a/go.mod h1:wK6yTYYcgjHE1Z1QtXACPDjcFJyBskHEdagmnq3vsP8=
 github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jmespath/go-jmespath v0.0.0-20160803190731-bd40a432e4c7/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
@@ -1420,6 +1425,7 @@ mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed h1:WX1yoOaKQfddO/mLzdV4wp
 mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed/go.mod h1:Xkxe497xwlCKkIaQYRfC7CSLworTXY9RMqwhhCm+8Nc=
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b h1:DxJ5nJdkhDlLok9K6qO+5290kphDJbHOQO1DFFFTeBo=
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b/go.mod h1:2odslEg/xrtNQqCYg2/jCoyKnw3vv5biOc3JnIcYfL4=
+mvdan.cc/unparam v0.0.0-20190209190245-fbb59629db34 h1:duVSyluuJA+u0BnkcLR01smoLrGgDTfWt5c8ODYG8fU=
 mvdan.cc/unparam v0.0.0-20190209190245-fbb59629db34/go.mod h1:H6SUd1XjIs+qQCyskXg5OFSrilMRUkD8ePJpHKDPaeY=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
 rsc.io/letsencrypt v0.0.3/go.mod h1:buyQKZ6IXrRnB7TdkHP0RyEybLx18HHyOSoTyoOLqNY=

pkg/controller/pravega/pravega_segmentstore.go

@@ -13,6 +13,7 @@ package pravega
 import (
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 
 	api "github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
@@ -151,6 +152,7 @@ func makeSegmentstorePodSpec(p *api.PravegaCluster) corev1.PodSpec {
 		volumes = append(volumes, v)
 	}
 
+	containerport, _ := strconv.Atoi(p.Spec.Pravega.Options["pravegaservice.service.listener.port"])
 	podSpec := corev1.PodSpec{
 		Containers: []corev1.Container{
 			{
@@ -163,7 +165,7 @@ func makeSegmentstorePodSpec(p *api.PravegaCluster) corev1.PodSpec {
 				Ports: []corev1.ContainerPort{
 					{
 						Name:          "server",
-						ContainerPort: 12345,
+						ContainerPort: int32(containerport),
 					},
 				},
 				EnvFrom:      environment,
@@ -173,7 +175,7 @@ func makeSegmentstorePodSpec(p *api.PravegaCluster) corev1.PodSpec {
 				ReadinessProbe: &corev1.Probe{
 					Handler: corev1.Handler{
 						Exec: &corev1.ExecAction{
-							Command: util.HealthcheckCommand(12345),
+							Command: util.HealthcheckCommand(int32(containerport)),
 						},
 					},
 					// Segment Stores can take a few minutes to become ready when the cluster
@@ -186,7 +188,7 @@ func makeSegmentstorePodSpec(p *api.PravegaCluster) corev1.PodSpec {
 				LivenessProbe: &corev1.Probe{
 					Handler: corev1.Handler{
 						Exec: &corev1.ExecAction{
-							Command: util.HealthcheckCommand(12345),
+							Command: util.HealthcheckCommand(int32(containerport)),
 						},
 					},
 					// In the readiness probe we allow the pod to take up to 5 minutes
@@ -281,6 +283,9 @@ func MakeSegmentstoreConfigMap(p *api.PravegaCluster) *corev1.ConfigMap {
 		"-Dpravegaservice.clusterName=" + p.Name,
 	}
 
+	if _, ok := p.Spec.Pravega.Options["pravegaservice.service.listener.port"]; !ok {
+		p.Spec.Pravega.Options["pravegaservice.service.listener.port"] = "12345"
+	}
 	if match, _ := util.CompareVersions(p.Spec.Version, "0.4.0", ">="); match {
 		// Pravega < 0.4 uses a Java version that does not support the options below
 		jvmOpts = append(jvmOpts,
@@ -487,6 +492,7 @@ func configureCaBundleSecret(podSpec *corev1.PodSpec, p *api.PravegaCluster) {
 }
 
 func MakeSegmentStoreHeadlessService(p *api.PravegaCluster) *corev1.Service {
+	serviceport, _ := strconv.Atoi(p.Spec.Pravega.Options["pravegaservice.service.listener.port"])
 	return &corev1.Service{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Service",
@@ -501,7 +507,7 @@ func MakeSegmentStoreHeadlessService(p *api.PravegaCluster) *corev1.Service {
 			Ports: []corev1.ServicePort{
 				{
 					Name:     "server",
-					Port:     12345,
+					Port:     int32(serviceport),
 					Protocol: "TCP",
 				},
 			},
@@ -549,6 +555,7 @@ func MakeSegmentStoreExternalServices(p *api.PravegaCluster) []*corev1.Service {
 	var service *corev1.Service
 	serviceType := getSSServiceType(p)
 	services := make([]*corev1.Service, p.Spec.Pravega.SegmentStoreReplicas)
+	serviceport, _ := strconv.Atoi(p.Spec.Pravega.Options["pravegaservice.service.listener.port"])
 	for i := int32(0); i < p.Spec.Pravega.SegmentStoreReplicas; i++ {
 		ssPodName := p.ServiceNameForSegmentStore(i)
 		annotationMap := p.Spec.Pravega.SegmentStoreServiceAnnotations
@@ -573,9 +580,9 @@ func MakeSegmentStoreExternalServices(p *api.PravegaCluster) []*corev1.Service {
 				Ports: []corev1.ServicePort{
 					{
 						Name:       "server",
-						Port:       12345,
+						Port:       int32(serviceport),
 						Protocol:   "TCP",
-						TargetPort: intstr.FromInt(12345),
+						TargetPort: intstr.FromInt(serviceport),
 					},
 				},
 				ExternalTrafficPolicy: corev1.ServiceExternalTrafficPolicyTypeLocal,
@@ -590,7 +597,7 @@ func MakeSegmentStoreExternalServices(p *api.PravegaCluster) []*corev1.Service {
 			service.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyTypeLocal
 		}
 		if p.Spec.Pravega.SegmentStoreLoadBalancerIP != "" {
-			service.Spec.Ports[0].Port = 12345 + i
+			service.Spec.Ports[0].Port = int32(serviceport) + i
 			service.Spec.LoadBalancerIP = p.Spec.Pravega.SegmentStoreLoadBalancerIP
 		}
 		services[i] = service

pkg/controller/pravega/pravega_segmentstore_test.go

@@ -12,6 +12,7 @@ package pravega_test
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
@@ -146,7 +147,8 @@ var _ = Describe("PravegaSegmentstore", func() {
 				})
 
 				It("should create a config-map", func() {
-					_ = pravega.MakeSegmentstoreConfigMap(p)
+					cm := pravega.MakeSegmentstoreConfigMap(p)
+					Ω(strings.Contains(cm.Data["JAVA_OPTS"], "-Dpravegaservice.service.listener.port=12345")).Should(BeTrue())
 					Ω(err).Should(BeNil())
 				})
 				It("should create a config-map with empty tier2", func() {
@@ -219,7 +221,8 @@ var _ = Describe("PravegaSegmentstore", func() {
 						ControllerJvmOptions:   []string{"-XX:MaxDirectMemorySize=1g", "-XX:MaxRAMPercentage=50.0"},
 						SegmentStoreJVMOptions: []string{"-XX:MaxDirectMemorySize=1g", "-XX:MaxRAMPercentage=50.0"},
 						Options: map[string]string{
-							"dummy-key": "dummy-value",
+							"dummy-key":                            "dummy-value",
+							"pravegaservice.service.listener.port": "443",
 						},
 						LongTermStorage: &v1beta1.LongTermStorageSpec{
 							FileSystem: &v1beta1.FileSystemSpec{
@@ -265,7 +268,8 @@ var _ = Describe("PravegaSegmentstore", func() {
 				})
 
 				It("should create a config-map", func() {
-					_ = pravega.MakeSegmentstoreConfigMap(p)
+					cm := pravega.MakeSegmentstoreConfigMap(p)
+					Ω(strings.Contains(cm.Data["JAVA_OPTS"], "-Dpravegaservice.service.listener.port=443")).Should(BeTrue())
 					Ω(err).Should(BeNil())
 				})
 

pkg/controller/pravegacluster/pravegacluster_controller.go

@@ -15,6 +15,7 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
+	"strings"
 	"time"
 
 	pravegav1beta1 "github.com/pravega/pravega-operator/pkg/apis/pravega/v1beta1"
@@ -254,8 +255,8 @@ func (r *ReconcilePravegaCluster) reconcileControllerConfigMap(p *pravegav1beta1
 }
 
 func (r *ReconcilePravegaCluster) reconcileSegmentStoreConfigMap(p *pravegav1beta1.PravegaCluster) (err error) {
-
 	currentConfigMap := &corev1.ConfigMap{}
+	segmentStorePortUpdated := false
 	configMap := pravega.MakeSegmentstoreConfigMap(p)
 	controllerutil.SetControllerReference(p, configMap, r.scheme)
 	err = r.client.Get(context.TODO(), types.NamespacedName{Name: p.ConfigMapNameForSegmentstore(), Namespace: p.Namespace}, currentConfigMap)
@@ -271,12 +272,13 @@ func (r *ReconcilePravegaCluster) reconcileSegmentStoreConfigMap(p *pravegav1bet
 		err = r.client.Get(context.TODO(), types.NamespacedName{Name: p.ConfigMapNameForSegmentstore(), Namespace: p.Namespace}, currentConfigMap)
 		eq := util.CompareConfigMap(currentConfigMap, configMap)
 		if !eq {
+			segmentStorePortUpdated = r.checkSegmentStorePortUpdated(p, currentConfigMap)
 			err := r.client.Update(context.TODO(), configMap)
 			if err != nil {
 				return err
 			}
 			//restarting sts pods
-			if !r.checkVersionUpgradeTriggered(p) {
+			if !r.checkVersionUpgradeTriggered(p) && !segmentStorePortUpdated {
 				err = r.restartStsPod(p)
 				if err != nil {
 					return err
@@ -286,6 +288,22 @@ func (r *ReconcilePravegaCluster) reconcileSegmentStoreConfigMap(p *pravegav1bet
 	}
 	return nil
 }
+func (r *ReconcilePravegaCluster) checkSegmentStorePortUpdated(p *pravegav1beta1.PravegaCluster, cm *corev1.ConfigMap) bool {
+	if val, ok := p.Spec.Pravega.Options["pravegaservice.service.listener.port"]; ok {
+		eq := false
+		data := strings.Split(cm.Data["JAVA_OPTS"], " ")
+		key := fmt.Sprintf("-Dpravegaservice.service.listener.port=%v", val)
+		for _, checkstring := range data {
+			if checkstring == key {
+				eq = true
+			}
+		}
+		if !eq {
+			return true
+		}
+	}
+	return false
+}
 
 func (r *ReconcilePravegaCluster) checkVersionUpgradeTriggered(p *pravegav1beta1.PravegaCluster) bool {
 	currentPravegaCluster := &pravegav1beta1.PravegaCluster{}
@@ -390,9 +408,26 @@ func (r *ReconcilePravegaCluster) reconcileControllerService(p *pravegav1beta1.P
 func (r *ReconcilePravegaCluster) reconcileSegmentStoreService(p *pravegav1beta1.PravegaCluster) (err error) {
 	headlessService := pravega.MakeSegmentStoreHeadlessService(p)
 	controllerutil.SetControllerReference(p, headlessService, r.scheme)
-	err = r.client.Create(context.TODO(), headlessService)
-	if err != nil && !errors.IsAlreadyExists(err) {
-		return err
+	currentService := &corev1.Service{}
+	err = r.client.Get(context.TODO(), types.NamespacedName{Name: headlessService.Name, Namespace: p.Namespace}, currentService)
+
+	if err != nil {
+		if errors.IsNotFound(err) {
+			err = r.client.Create(context.TODO(), headlessService)
+			if err != nil && !errors.IsAlreadyExists(err) {
+				return err
+			}
+		}
+	} else {
+
+		if currentService.Spec.Ports[0].Port != headlessService.Spec.Ports[0].Port {
+			currentService.Spec.Ports[0].Port = headlessService.Spec.Ports[0].Port
+			currentService.Spec.Ports[0].TargetPort = headlessService.Spec.Ports[0].TargetPort
+			err = r.client.Update(context.TODO(), currentService)
+			if err != nil {
+				return fmt.Errorf("failed to update headless service port (%s): %v", currentService.Name, err)
+			}
+		}
 	}
 
 	if p.Spec.ExternalAccess.Enabled {
@@ -409,6 +444,15 @@ func (r *ReconcilePravegaCluster) reconcileSegmentStoreService(p *pravegav1beta1
 					}
 				}
 			} else {
+				if service.Spec.Ports[0].Port != currentservice.Spec.Ports[0].Port {
+					currentservice.Spec.Ports[0].Port = service.Spec.Ports[0].Port
+					currentservice.Spec.Ports[0].TargetPort = service.Spec.Ports[0].TargetPort
+					err = r.client.Update(context.TODO(), currentservice)
+					if err != nil {
+						return fmt.Errorf("failed to update external service port (%s): %v", currentservice.Name, err)
+					}
+				}
+
 				eq := reflect.DeepEqual(currentservice.Annotations["external-dns.alpha.kubernetes.io/hostname"], service.Annotations["external-dns.alpha.kubernetes.io/hostname"])
 				if !eq {
 					err := r.client.Delete(context.TODO(), currentservice)
@@ -606,6 +650,19 @@ func (r *ReconcilePravegaCluster) deploySegmentStore(p *pravegav1beta1.PravegaCl
 			if err != nil {
 				return err
 			}
+			if statefulSet.Spec.Template.Spec.Containers[0].Ports[0].ContainerPort != sts.Spec.Template.Spec.Containers[0].Ports[0].ContainerPort {
+				sts.Spec.Template.Spec.Containers[0].Ports[0].ContainerPort = statefulSet.Spec.Template.Spec.Containers[0].Ports[0].ContainerPort
+				sts.Spec.Template.Spec.Containers[0].ReadinessProbe = statefulSet.Spec.Template.Spec.Containers[0].ReadinessProbe
+				sts.Spec.Template.Spec.Containers[0].LivenessProbe = statefulSet.Spec.Template.Spec.Containers[0].LivenessProbe
+				err = r.client.Update(context.TODO(), sts)
+				if err != nil {
+					return fmt.Errorf("failed to update stateful set: %v", err)
+				}
+				err = r.restartStsPod(p)
+				if err != nil {
+					return err
+				}
+			}
 			owRefs := sts.GetOwnerReferences()
 			if hasOldVersionOwnerReference(owRefs) {
 				log.Printf("Deleting SSS STS as it has old version owner ref.")

pkg/test/e2e/e2eutil/pravegacluster_util.go

@@ -13,6 +13,7 @@ package e2eutil
 import (
 	goctx "context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -885,9 +886,27 @@ func CheckStsExists(t *testing.T, f *framework.Framework, ctx *framework.TestCtx
 	if err != nil {
 		return fmt.Errorf("sts doesnt exist: %v", err)
 	}
+
 	return nil
 }
 
+func CheckConfigMapUpdated(t *testing.T, f *framework.Framework, ctx *framework.TestCtx, pravega *api.PravegaCluster, cmName string, key string, value string) error {
+	cm := &corev1.ConfigMap{}
+	err := f.Client.Get(goctx.TODO(), types.NamespacedName{Namespace: pravega.Namespace, Name: cmName}, cm)
+	if err != nil {
+		return fmt.Errorf("failed to obtain configmap: %v", err)
+	}
+	if cm != nil {
+		optvalue := cm.Data[key]
+		if strings.Contains(optvalue, value) {
+			t.Logf("configmap is updated %s", cm.Name)
+			return nil
+		}
+	}
+
+	return fmt.Errorf("config map is not updated")
+}
+
 func RestartTier2(t *testing.T, f *framework.Framework, ctx *framework.TestCtx, namespace string) error {
 	t.Log("restarting tier2 storage")
 	tier2 := NewTier2(namespace)

test/e2e/cmchanges_test.go

@@ -11,8 +11,10 @@
 package e2e
 
 import (
+	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	. "github.com/onsi/gomega"
 	framework "github.com/operator-framework/operator-sdk/pkg/test"
@@ -56,21 +58,34 @@ func testCMUpgradeCluster(t *testing.T) {
 	pravega, err = pravega_e2eutil.GetPravegaCluster(t, f, ctx, pravega)
 	g.Expect(err).NotTo(HaveOccurred())
 
-	//updating pravega option
+	//updating pravega options
 	pravega.Spec.Pravega.Options["bookkeeper.bkAckQuorumSize"] = "2"
+	pravega.Spec.Pravega.Options["pravegaservice.service.listener.port"] = "443"
 
 	//updating pravegacluster
 	err = pravega_e2eutil.UpdatePravegaCluster(t, f, ctx, pravega)
 	g.Expect(err).NotTo(HaveOccurred())
 
-	//checking if the upgrade of options was successful
+	//checking if the upgrade of option was successful
 	err = pravega_e2eutil.WaitForCMPravegaClusterToUpgrade(t, f, ctx, pravega)
 	g.Expect(err).NotTo(HaveOccurred())
 
 	// This is to get the latest Pravega cluster object
 	pravega, err = pravega_e2eutil.GetPravegaCluster(t, f, ctx, pravega)
 	g.Expect(err).NotTo(HaveOccurred())
 
+	// Check configmap is  Updated
+	cmName := fmt.Sprintf("%s-pravega-segmentstore", pravega.Name)
+	value := "pravegaservice.service.listener.port=443"
+	err = pravega_e2eutil.CheckConfigMapUpdated(t, f, ctx, pravega, cmName, "JAVA_OPTS", value)
+	g.Expect(err).NotTo(HaveOccurred())
+
+	// Sleeping for 1 min before read/write data
+	time.Sleep(60 * time.Second)
+
+	err = pravega_e2eutil.WriteAndReadData(t, f, ctx, pravega)
+	g.Expect(err).NotTo(HaveOccurred())
+
 	//updating pravega option
 	pravega.Spec.Pravega.Options["pravegaservice.containerCount"] = "10"