Merge pull request #389 from arangodb/lb-source-ranges

Added loadBalancerSourceRanges field to external-access-spec

Makefile

@@ -182,7 +182,7 @@ dashboard/assets.go: $(DASHBOARDSOURCES) $(DASHBOARDDIR)/Dockerfile.build
 
 $(BIN): $(SOURCES) dashboard/assets.go
 	@mkdir -p $(BINDIR)
-	CGO_ENABLED=0 go build -installsuffix netgo -ldflags "-X main.projectVersion=$(VERSION) -X main.projectBuild=$(COMMIT)" -o $(BIN) $(REPOPATH)
+	CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -installsuffix netgo -ldflags "-X main.projectVersion=$(VERSION) -X main.projectBuild=$(COMMIT)" -o $(BIN) $(REPOPATH)
 
 .PHONY: docker
 docker: check-vars $(BIN)

dashboard/.gitignore

@@ -19,3 +19,5 @@
 npm-debug.log*
 yarn-debug.log*
 yarn-error.log*
+
+assets.go

docs/Manual/Deployment/Kubernetes/DeploymentResource.md

@@ -158,6 +158,14 @@ This setting is used when `spec.externalAccess.type` is set to `LoadBalancer` or
 
 If you do not specify this setting, an IP will be chosen automatically by the load-balancer provisioner.
 
+### `spec.externalAccess.loadBalancerSourceRanges: []string`
+
+If specified and supported by the platform (cloud provider), this will restrict traffic through the cloud-provider
+load-balancer will be restricted to the specified client IPs. This field will be ignored if the
+cloud-provider does not support the feature.
+
+More info: https://kubernetes.io/docs/tasks/access-application-cluster/configure-cloud-provider-firewall/
+
 ### `spec.externalAccess.nodePort: int`
 
 This setting specifies the port used to expose the ArangoDB deployment on.
@@ -254,6 +262,15 @@ This setting is used when `spec.sync.externalAccess.type` is set to `NodePort` o
 
 If you do not specify this setting, a random port will be chosen automatically.
 
+### `spec.sync.externalAccess.loadBalancerSourceRanges: []string`
+
+If specified and supported by the platform (cloud provider), this will restrict traffic through the cloud-provider
+load-balancer will be restricted to the specified client IPs. This field will be ignored if the
+cloud-provider does not support the feature.
+
+More info: https://kubernetes.io/docs/tasks/access-application-cluster/configure-cloud-provider-firewall/
+
+
 ### `spec.sync.externalAccess.masterEndpoint: []string`
 
 This setting specifies the master endpoint(s) advertised by the ArangoSync SyncMasters.

pkg/apis/deployment/v1alpha/external_access_spec.go

@@ -24,6 +24,7 @@ package v1alpha
 
 import (
 	"fmt"
+	"net"
 	"net/url"
 
 	"github.com/arangodb/kube-arangodb/pkg/util"
@@ -37,6 +38,11 @@ type ExternalAccessSpec struct {
 	NodePort *int `json:"nodePort,omitempty"`
 	// Optional IP used to configure a load-balancer on, in case of Auto or LoadBalancer type.
 	LoadBalancerIP *string `json:"loadBalancerIP,omitempty"`
+	// If specified and supported by the platform, this will restrict traffic through the cloud-provider
+	// load-balancer will be restricted to the specified client IPs. This field will be ignored if the
+	// cloud-provider does not support the feature.
+	// More info: https://kubernetes.io/docs/tasks/access-application-cluster/configure-cloud-provider-firewall/
+	LoadBalancerSourceRanges []string `json:"loadBalancerSourceRanges,omitempty"`
 	// Advertised Endpoint is passed to the coordinators/single servers for advertising a specific endpoint
 	AdvertisedEndpoint *string `json:"advertisedEndpoint,omitempty"`
 }
@@ -77,6 +83,11 @@ func (s ExternalAccessSpec) Validate() error {
 			return maskAny(fmt.Errorf("Failed to parse advertised endpoint '%s': %s", ep, err))
 		}
 	}
+	for _, x := range s.LoadBalancerSourceRanges {
+		if _, _, err := net.ParseCIDR(x); err != nil {
+			return maskAny(fmt.Errorf("Failed to parse loadbalancer source range '%s': %s", x, err))
+		}
+	}
 	return nil
 }
 
@@ -95,6 +106,9 @@ func (s *ExternalAccessSpec) SetDefaultsFrom(source ExternalAccessSpec) {
 	if s.LoadBalancerIP == nil {
 		s.LoadBalancerIP = util.NewStringOrNil(source.LoadBalancerIP)
 	}
+	if s.LoadBalancerSourceRanges == nil && len(source.LoadBalancerSourceRanges) > 0 {
+		s.LoadBalancerSourceRanges = append([]string{}, source.LoadBalancerSourceRanges...)
+	}
 	if s.AdvertisedEndpoint == nil {
 		s.AdvertisedEndpoint = source.AdvertisedEndpoint
 	}

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

@@ -427,6 +427,11 @@ func (in *ExternalAccessSpec) DeepCopyInto(out *ExternalAccessSpec) {
 		*out = new(string)
 		**out = **in
 	}
+	if in.LoadBalancerSourceRanges != nil {
+		in, out := &in.LoadBalancerSourceRanges, &out.LoadBalancerSourceRanges
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 	if in.AdvertisedEndpoint != nil {
 		in, out := &in.AdvertisedEndpoint, &out.AdvertisedEndpoint
 		*out = new(string)

pkg/deployment/resources/services.go

@@ -23,6 +23,7 @@
 package resources
 
 import (
+	"strings"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -143,7 +144,9 @@ func (r *Resources) ensureExternalAccessServices(svcs k8sutil.ServiceInterface,
 	eaServiceType := spec.GetType().AsServiceType() // Note: Type auto defaults to ServiceTypeLoadBalancer
 	if existing, err := svcs.Get(eaServiceName, metav1.GetOptions{}); err == nil {
 		// External access service exists
+		updateExternalAccessService := false
 		loadBalancerIP := spec.GetLoadBalancerIP()
+		loadBalancerSourceRanges := spec.LoadBalancerSourceRanges
 		nodePort := spec.GetNodePort()
 		if spec.GetType().IsNone() {
 			if noneIsClusterIP {
@@ -179,12 +182,22 @@ func (r *Resources) ensureExternalAccessServices(svcs k8sutil.ServiceInterface,
 				deleteExternalAccessService = true // Remove the current and replace with proper one
 				createExternalAccessService = true
 			}
+			if strings.Join(existing.Spec.LoadBalancerSourceRanges, ",") != strings.Join(loadBalancerSourceRanges, ",") {
+				updateExternalAccessService = true
+				existing.Spec.LoadBalancerSourceRanges = loadBalancerSourceRanges
+			}
 		} else if spec.GetType().IsNodePort() {
 			if existing.Spec.Type != v1.ServiceTypeNodePort || len(existing.Spec.Ports) != 1 || (nodePort != 0 && existing.Spec.Ports[0].NodePort != int32(nodePort)) {
 				deleteExternalAccessService = true // Remove the current and replace with proper one
 				createExternalAccessService = true
 			}
 		}
+		if updateExternalAccessService && !createExternalAccessService && !deleteExternalAccessService {
+			if _, err := svcs.Update(existing); err != nil {
+				log.Debug().Err(err).Msgf("Failed to update %s external access service", title)
+				return maskAny(err)
+			}
+		}
 	} else if k8sutil.IsNotFound(err) {
 		// External access service does not exist
 		if !spec.GetType().IsNone() || noneIsClusterIP {
@@ -202,7 +215,8 @@ func (r *Resources) ensureExternalAccessServices(svcs k8sutil.ServiceInterface,
 		// Let's create or update the database external access service
 		nodePort := spec.GetNodePort()
 		loadBalancerIP := spec.GetLoadBalancerIP()
-		_, newlyCreated, err := k8sutil.CreateExternalAccessService(svcs, eaServiceName, svcRole, apiObject, eaServiceType, port, nodePort, loadBalancerIP, apiObject.AsOwner())
+		loadBalancerSourceRanges := spec.LoadBalancerSourceRanges
+		_, newlyCreated, err := k8sutil.CreateExternalAccessService(svcs, eaServiceName, svcRole, apiObject, eaServiceType, port, nodePort, loadBalancerIP, loadBalancerSourceRanges, apiObject.AsOwner())
 		if err != nil {
 			log.Debug().Err(err).Msgf("Failed to create %s external access service", title)
 			return maskAny(err)

pkg/util/k8sutil/services.go

@@ -36,6 +36,7 @@ import (
 // ServiceInterface has methods to work with Service resources.
 type ServiceInterface interface {
 	Create(*v1.Service) (*v1.Service, error)
+	Update(*v1.Service) (*v1.Service, error)
 	Delete(name string, options *metav1.DeleteOptions) error
 	Get(name string, options metav1.GetOptions) (*v1.Service, error)
 }
@@ -120,7 +121,7 @@ func CreateHeadlessService(svcs ServiceInterface, deployment metav1.Object, owne
 	}
 	publishNotReadyAddresses := true
 	serviceType := v1.ServiceTypeClusterIP
-	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), ClusterIPNone, "", serviceType, ports, "", publishNotReadyAddresses, owner)
+	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), ClusterIPNone, "", serviceType, ports, "", nil, publishNotReadyAddresses, owner)
 	if err != nil {
 		return "", false, maskAny(err)
 	}
@@ -149,7 +150,7 @@ func CreateDatabaseClientService(svcs ServiceInterface, deployment metav1.Object
 	}
 	serviceType := v1.ServiceTypeClusterIP
 	publishNotReadyAddresses := false
-	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), "", role, serviceType, ports, "", publishNotReadyAddresses, owner)
+	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), "", role, serviceType, ports, "", nil, publishNotReadyAddresses, owner)
 	if err != nil {
 		return "", false, maskAny(err)
 	}
@@ -160,7 +161,7 @@ func CreateDatabaseClientService(svcs ServiceInterface, deployment metav1.Object
 // If the service already exists, nil is returned.
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
-func CreateExternalAccessService(svcs ServiceInterface, svcName, role string, deployment metav1.Object, serviceType v1.ServiceType, port, nodePort int, loadBalancerIP string, owner metav1.OwnerReference) (string, bool, error) {
+func CreateExternalAccessService(svcs ServiceInterface, svcName, role string, deployment metav1.Object, serviceType v1.ServiceType, port, nodePort int, loadBalancerIP string, loadBalancerSourceRanges []string, owner metav1.OwnerReference) (string, bool, error) {
 	deploymentName := deployment.GetName()
 	ports := []v1.ServicePort{
 		v1.ServicePort{
@@ -171,7 +172,7 @@ func CreateExternalAccessService(svcs ServiceInterface, svcName, role string, de
 		},
 	}
 	publishNotReadyAddresses := false
-	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), "", role, serviceType, ports, loadBalancerIP, publishNotReadyAddresses, owner)
+	newlyCreated, err := createService(svcs, svcName, deploymentName, deployment.GetNamespace(), "", role, serviceType, ports, loadBalancerIP, loadBalancerSourceRanges, publishNotReadyAddresses, owner)
 	if err != nil {
 		return "", false, maskAny(err)
 	}
@@ -183,7 +184,7 @@ func CreateExternalAccessService(svcs ServiceInterface, svcName, role string, de
 // If another error occurs, that error is returned.
 // The returned bool is true if the service is created, or false when the service already existed.
 func createService(svcs ServiceInterface, svcName, deploymentName, ns, clusterIP, role string, serviceType v1.ServiceType,
-	ports []v1.ServicePort, loadBalancerIP string, publishNotReadyAddresses bool, owner metav1.OwnerReference) (bool, error) {
+	ports []v1.ServicePort, loadBalancerIP string, loadBalancerSourceRanges []string, publishNotReadyAddresses bool, owner metav1.OwnerReference) (bool, error) {
 	labels := LabelsForDeployment(deploymentName, role)
 	svc := &v1.Service{
 		ObjectMeta: metav1.ObjectMeta{
@@ -203,6 +204,7 @@ func createService(svcs ServiceInterface, svcName, deploymentName, ns, clusterIP
 			ClusterIP:                clusterIP,
 			PublishNotReadyAddresses: publishNotReadyAddresses,
 			LoadBalancerIP:           loadBalancerIP,
+			LoadBalancerSourceRanges: loadBalancerSourceRanges,
 		},
 	}
 	addOwnerRefToObject(svc.GetObjectMeta(), &owner)

pkg/util/k8sutil/services_cache.go

@@ -58,6 +58,15 @@ func (sc *servicesCache) Create(s *v1.Service) (*v1.Service, error) {
 	return result, nil
 }
 
+func (sc *servicesCache) Update(s *v1.Service) (*v1.Service, error) {
+	sc.cache = nil
+	result, err := sc.cli.Update(s)
+	if err != nil {
+		return nil, maskAny(err)
+	}
+	return result, nil
+}
+
 func (sc *servicesCache) Delete(name string, options *metav1.DeleteOptions) error {
 	sc.cache = nil
 	if err := sc.cli.Delete(name, options); err != nil {

tests/load_balancer_source_ranges_test.go

@@ -0,0 +1,158 @@
+//
+// DISCLAIMER
+//
+// Copyright 2019 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+// Author Ewout Prangsma
+// Author Max Neunhoeffer
+//
+
+package tests
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/dchest/uniuri"
+
+	api "github.com/arangodb/kube-arangodb/pkg/apis/deployment/v1alpha"
+	"github.com/arangodb/kube-arangodb/pkg/client"
+	"github.com/arangodb/kube-arangodb/pkg/util"
+	"github.com/arangodb/kube-arangodb/pkg/util/k8sutil"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// tests cursor forwarding with load-balanced conn., specify a source range
+func TestLoadBalancingSourceRanges(t *testing.T) {
+	longOrSkip(t)
+
+	c := client.MustNewInCluster()
+	kubecli := mustNewKubeClient(t)
+	ns := getNamespace(t)
+
+	// Prepare deployment config
+	namePrefix := "test-lb-src-ranges-"
+	depl := newDeployment(namePrefix + uniuri.NewLen(4))
+	depl.Spec.Mode = api.NewMode(api.DeploymentModeCluster)
+	depl.Spec.Image = util.NewString("arangodb/arangodb:latest")
+	depl.Spec.ExternalAccess.Type = api.NewExternalAccessType(api.ExternalAccessTypeLoadBalancer)
+	depl.Spec.ExternalAccess.LoadBalancerSourceRanges = append(depl.Spec.ExternalAccess.LoadBalancerSourceRanges, "1.2.3.0/24", "0.0.0.0/0")
+
+	// Create deployment
+	_, err := c.DatabaseV1alpha().ArangoDeployments(ns).Create(depl)
+	if err != nil {
+		t.Fatalf("Create deployment failed: %v", err)
+	}
+	// Prepare cleanup
+	defer removeDeployment(c, depl.GetName(), ns)
+
+	// Wait for deployment to be ready
+	apiObject, err := waitUntilDeployment(c, depl.GetName(), ns, deploymentIsReady())
+	if err != nil {
+		t.Fatalf("Deployment not running in time: %v", err)
+	}
+
+	// Create a database client
+	ctx := context.Background()
+	clOpts := &DatabaseClientOptions{
+		UseVST:       false,
+		ShortTimeout: true,
+	}
+	client := mustNewArangodDatabaseClient(ctx, kubecli, apiObject, t, clOpts)
+
+	// Wait for cluster to be available
+	if err := waitUntilVersionUp(client, nil); err != nil {
+		t.Fatalf("Cluster not running returning version in time: %v", err)
+	}
+
+	// Now let's use the k8s api to check if the source ranges are present in
+	// the external service spec:
+	svcs := kubecli.CoreV1().Services(ns)
+	eaServiceName := k8sutil.CreateDatabaseExternalAccessServiceName(depl.GetName())
+	// Just in case, give the service some time to appear, it should usually
+	// be there already, when the deployment is ready, however, we have had
+	// unstable tests in the past
+	counter := 0
+	var foundExternalIP string
+	for {
+		if svc, err := svcs.Get(eaServiceName, metav1.GetOptions{}); err == nil {
+			spec := svc.Spec
+			ranges := spec.LoadBalancerSourceRanges
+			if len(ranges) != 2 {
+				t.Errorf("LoadBalancerSourceRanges does not have length 2: %v", ranges)
+			} else {
+				if ranges[0] != "1.2.3.0/24" {
+					t.Errorf("Expecting first LoadBalancerSourceRange to be \"1.2.3.0/24\", but ranges are: %v", ranges)
+				}
+				if ranges[1] != "0.0.0.0/0" {
+					t.Errorf("Expecting second LoadBalancerSourceRange to be \"0.0.0.0/0\", but ranges are: %v", ranges)
+				}
+			}
+			foundExternalIP = spec.LoadBalancerIP
+			break
+		}
+		t.Logf("Service %s cannot be found, waiting for some time...", eaServiceName)
+		time.Sleep(time.Second)
+		counter += 1
+		if counter >= 60 {
+			t.Fatalf("Could not find service %s within 60 seconds, giving up.", eaServiceName)
+		}
+	}
+
+	// Now change the deployment spec to use different ranges:
+	depl, err = updateDeployment(c, depl.GetName(), ns,
+		func(spec *api.DeploymentSpec) {
+			spec.ExternalAccess.LoadBalancerSourceRanges = []string{"4.5.0.0/16"}
+		})
+	if err != nil {
+		t.Fatalf("Failed to update the deployment")
+	}
+
+	// And check again:
+	counter = 0
+	for {
+		time.Sleep(time.Second)
+		if svc, err := svcs.Get(eaServiceName, metav1.GetOptions{}); err == nil {
+			spec := svc.Spec
+			ranges := spec.LoadBalancerSourceRanges
+			good := true
+			if len(ranges) != 1 {
+				t.Logf("LoadBalancerSourceRanges does not have length 1: %v, waiting some more...", ranges)
+				good = false
+			} else {
+				if ranges[0] != "4.5.0.0/16" {
+					t.Logf("Expecting only LoadBalancerSourceRange to be \"4.5.0.0/16\", but ranges are: %v, waiting some more...", ranges)
+					good = false
+				} else {
+					if spec.LoadBalancerIP != foundExternalIP {
+						t.Errorf("Oops, the external IP of the external access service has changed: previously: %s, now: %s", foundExternalIP, spec.LoadBalancerIP)
+					}
+				}
+			}
+			if good {
+				break
+			}
+		}
+		t.Logf("Service %s cannot be found, waiting for some more time...", eaServiceName)
+		counter += 1
+		if counter >= 60 {
+			t.Fatalf("Could not find changed service %s within 60 seconds, giving up.", eaServiceName)
+		}
+	}
+	t.Logf("Success! Service %s was changed correctly.", eaServiceName)
+}