[Feature] Add Internal Port (#713)

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Change Log
 
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
+- Add InternalPort to ServerGroupSpec to allow user to expose tcp connection over localhost for sidecars
 
 ## [1.1.7](https://github.com/arangodb/kube-arangodb/tree/1.1.7) (2021-04-14)
 - Bump Kubernetes Dependencies to 1.19.x

pkg/apis/deployment/v1/server_group_spec.go

@@ -139,6 +139,8 @@ type ServerGroupSpec struct {
 	ShutdownMethod *ServerGroupShutdownMethod `json:"shutdownMethod,omitempty"`
 	// ShutdownDelay define how long operator should delay finalizer removal after shutdown
 	ShutdownDelay *int `json:"shutdownDelay,omitempty"`
+	// InternalPort define port used in internal communication, can be accessed over localhost via sidecar
+	InternalPort *int `json:"internalPort,omitempty"`
 }
 
 // ServerGroupSpecSecurityContext contains specification for pod security context
@@ -499,6 +501,12 @@ func (s ServerGroupSpec) Validate(group ServerGroup, used bool, mode DeploymentM
 	} else if s.GetCount() != 0 {
 		return errors.WithStack(errors.Wrapf(ValidationError, "Invalid count value %d for un-used group. Expected 0", s.GetCount()))
 	}
+	if port := s.InternalPort; port != nil {
+		switch p := *port; p {
+		case 8529:
+			return errors.WithStack(errors.Wrapf(ValidationError, "Port %d already in use", p))
+		}
+	}
 	return nil
 }
 

pkg/apis/deployment/v1/zz_generated.deepcopy.go

@@ -1512,6 +1512,11 @@ func (in *ServerGroupSpec) DeepCopyInto(out *ServerGroupSpec) {
 		*out = new(int)
 		**out = **in
 	}
+	if in.InternalPort != nil {
+		in, out := &in.InternalPort, &out.InternalPort
+		*out = new(int)
+		**out = **in
+	}
 	return
 }
 

pkg/apis/deployment/v2alpha1/arango_member.go

@@ -23,6 +23,7 @@
 package v2alpha1
 
 import (
+	"github.com/arangodb/kube-arangodb/pkg/apis/deployment"
 	meta "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -47,3 +48,17 @@ type ArangoMember struct {
 	Spec            ArangoMemberSpec   `json:"spec,omitempty"`
 	Status          ArangoMemberStatus `json:"status,omitempty"`
 }
+
+// AsOwner creates an OwnerReference for the given member
+func (a *ArangoMember) AsOwner() meta.OwnerReference {
+	trueVar := true
+	return meta.OwnerReference{
+		APIVersion: SchemeGroupVersion.String(),
+		Kind:       deployment.ArangoMemberResourceKind,
+		Name:       a.Name,
+		UID:        a.UID,
+		Controller: &trueVar,
+		// For now BlockOwnerDeletion does not work on OpenShift, so we leave it out.
+		//BlockOwnerDeletion: &trueVar,
+	}
+}

pkg/apis/deployment/v2alpha1/deployment_status_members.go

@@ -285,3 +285,19 @@ func (ds DeploymentStatusMembers) MembersOfGroup(group ServerGroup) MemberStatus
 		return MemberStatusList{}
 	}
 }
+
+// PodNames returns all members pod names
+func (ds DeploymentStatusMembers) PodNames() []string {
+	var n []string
+
+	ds.ForeachServerGroup(func(group ServerGroup, list MemberStatusList) error {
+		for _, m := range list {
+			if m.PodName != "" {
+				n = append(n, m.PodName)
+			}
+		}
+		return nil
+	})
+
+	return n
+}

pkg/apis/deployment/v2alpha1/member_phase.go

@@ -57,3 +57,8 @@ func (p MemberPhase) IsFailed() bool {
 func (p MemberPhase) IsCreatedOrDrain() bool {
 	return p == MemberPhaseCreated || p == MemberPhaseDrain
 }
+
+// String returns string from MemberPhase
+func (p MemberPhase) String() string {
+	return string(p)
+}

pkg/apis/deployment/v2alpha1/member_status.go

@@ -26,6 +26,8 @@ import (
 	"reflect"
 	"time"
 
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+
 	"k8s.io/apimachinery/pkg/types"
 
 	driver "github.com/arangodb/go-driver"
@@ -155,3 +157,8 @@ func (s MemberStatus) IsNotReadySince(timestamp time.Time) bool {
 	// A
 	return s.CreatedAt.Time.Before(timestamp)
 }
+
+// ArangoMemberName create member name from given member
+func (s MemberStatus) ArangoMemberName(deploymentName string, group ServerGroup) string {
+	return k8sutil.CreatePodHostName(deploymentName, group.AsRole(), s.ID)
+}

pkg/apis/deployment/v2alpha1/server_group.go

@@ -23,6 +23,7 @@
 package v2alpha1
 
 import (
+	"encoding/json"
 	"time"
 )
 
@@ -34,14 +35,19 @@ func (g *ServerGroup) UnmarshalJSON(bytes []byte) error {
 		return nil
 	}
 
-	*g = ServerGroupFromRole(string(bytes))
+	var s string
+
+	if err := json.Unmarshal(bytes, &s); err != nil {
+		return err
+	}
+
+	*g = ServerGroupFromRole(s)
 
 	return nil
 }
 
 func (g ServerGroup) MarshalJSON() ([]byte, error) {
-	s := g.AsRole()
-	return []byte(s), nil
+	return json.Marshal(g.AsRole())
 }
 
 const (

pkg/apis/deployment/v2alpha1/server_group_spec.go

@@ -137,6 +137,10 @@ type ServerGroupSpec struct {
 	InitContainers *ServerGroupInitContainers `json:"initContainers,omitempty"`
 	// ShutdownMethod describe procedure of member shutdown taken by Operator
 	ShutdownMethod *ServerGroupShutdownMethod `json:"shutdownMethod,omitempty"`
+	// ShutdownDelay define how long operator should delay finalizer removal after shutdown
+	ShutdownDelay *int `json:"shutdownDelay,omitempty"`
+	// InternalPort define port used in internal communication, can be accessed over localhost via sidecar
+	InternalPort *int `json:"internalPort,omitempty"`
 }
 
 // ServerGroupSpecSecurityContext contains specification for pod security context
@@ -497,6 +501,12 @@ func (s ServerGroupSpec) Validate(group ServerGroup, used bool, mode DeploymentM
 	} else if s.GetCount() != 0 {
 		return errors.WithStack(errors.Wrapf(ValidationError, "Invalid count value %d for un-used group. Expected 0", s.GetCount()))
 	}
+	if port := s.InternalPort; port != nil {
+		switch p := *port; p {
+		case 8529:
+			return errors.WithStack(errors.Wrapf(ValidationError, "Port %d already in use", p))
+		}
+	}
 	return nil
 }
 

pkg/apis/deployment/v2alpha1/zz_generated.deepcopy.go

@@ -1507,6 +1507,16 @@ func (in *ServerGroupSpec) DeepCopyInto(out *ServerGroupSpec) {
 		*out = new(ServerGroupShutdownMethod)
 		**out = **in
 	}
+	if in.ShutdownDelay != nil {
+		in, out := &in.ShutdownDelay, &out.ShutdownDelay
+		*out = new(int)
+		**out = **in
+	}
+	if in.InternalPort != nil {
+		in, out := &in.InternalPort, &out.InternalPort
+		*out = new(int)
+		**out = **in
+	}
 	return
 }
 

pkg/deployment/resources/exporter.go

@@ -21,14 +21,10 @@
 package resources
 
 import (
-	"fmt"
 	"path/filepath"
-	"sort"
-	"strconv"
-
-	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/probes"
 
 	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil/probes"
 
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
 	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
@@ -63,39 +59,32 @@ func ArangodbExporterContainer(image string, args []string, livenessProbe *probe
 	return c
 }
 
-func createExporterArgs(spec api.DeploymentSpec) []string {
+func createExporterArgs(spec api.DeploymentSpec, groupSpec api.ServerGroupSpec) []string {
 	tokenpath := filepath.Join(k8sutil.ExporterJWTVolumeMountDir, constants.SecretKeyToken)
-	options := make([]k8sutil.OptionPair, 0, 64)
-	scheme := "http"
-	if spec.IsSecure() {
-		scheme = "https"
+	options := k8sutil.CreateOptionPairs(64)
+
+	options.Add("--arangodb.jwt-file", tokenpath)
+
+	if port := groupSpec.InternalPort; port == nil {
+		scheme := "http"
+		if spec.IsSecure() {
+			scheme = "https"
+		}
+		options.Addf("--arangodb.endpoint", "%s://localhost:%d", scheme, k8sutil.ArangoPort)
+	} else {
+		options.Addf("--arangodb.endpoint", "http://localhost:%d", *port)
 	}
-	options = append(options,
-		k8sutil.OptionPair{Key: "--arangodb.jwt-file", Value: tokenpath},
-		k8sutil.OptionPair{Key: "--arangodb.endpoint", Value: scheme + "://localhost:" + strconv.Itoa(k8sutil.ArangoPort)},
-	)
+
 	keyPath := filepath.Join(k8sutil.TLSKeyfileVolumeMountDir, constants.SecretTLSKeyfile)
 	if spec.IsSecure() && spec.Metrics.IsTLS() {
-		options = append(options,
-			k8sutil.OptionPair{Key: "--ssl.keyfile", Value: keyPath},
-		)
+		options.Add("--ssl.keyfile", keyPath)
 	}
 
 	if port := spec.Metrics.GetPort(); port != k8sutil.ArangoExporterPort {
-		options = append(options,
-			k8sutil.OptionPair{Key: "--server.address", Value: fmt.Sprintf(":%d", port)},
-		)
+		options.Addf("--server.address", ":%d", port)
 	}
 
-	args := make([]string, 0, 2+len(options))
-	sort.Slice(options, func(i, j int) bool {
-		return options[i].CompareTo(options[j]) < 0
-	})
-	for _, o := range options {
-		args = append(args, o.Key+"="+o.Value)
-	}
-
-	return args
+	return options.Sort().AsArgs()
 }
 
 func createExporterLivenessProbe(isSecure bool) *probes.HTTPProbeConfig {

pkg/deployment/resources/pod_creator.go

@@ -77,6 +77,9 @@ func createArangodArgs(input pod.Input, additionalOptions ...k8sutil.OptionPair)
 	}
 
 	options.Addf("--server.endpoint", "%s://%s:%d", scheme, input.Deployment.GetListenAddr(), k8sutil.ArangoPort)
+	if port := input.GroupSpec.InternalPort; port != nil {
+		options.Addf("--server.endpoint", "tcp://127.0.0.1:%d", *port)
+	}
 
 	// Authentication
 	options.Merge(pod.JWT().Args(input))
@@ -230,7 +233,7 @@ func createArangoSyncArgs(apiObject metav1.Object, spec api.DeploymentSpec, grou
 		runCmd,
 	}
 
-	args = append(args, options.Sort().AsArgs()...)
+	args = append(args, options.Copy().Sort().AsArgs()...)
 
 	if len(groupSpec.Args) > 0 {
 		args = append(args, groupSpec.Args...)

pkg/deployment/resources/pod_creator_arangod.go

@@ -500,7 +500,7 @@ func (m *MemberArangoDPod) createMetricsExporterSidecar() *core.Container {
 		image = m.spec.Metrics.GetImage()
 	}
 
-	args := createExporterArgs(m.spec)
+	args := createExporterArgs(m.spec, m.groupSpec)
 	if m.spec.Metrics.Mode.Get() == api.MetricsModeSidecar {
 		args = append(args, "--mode=passthru")
 	}