Prevent side effects on dryrun in service registry

pkg/registry/core/service/portallocator/BUILD

@@ -23,7 +23,10 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["allocator_test.go"],
+    srcs = [
+        "allocator_test.go",
+        "operation_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//pkg/apis/core:go_default_library",

pkg/registry/core/service/portallocator/operation.go

@@ -33,15 +33,18 @@ type PortAllocationOperation struct {
 	allocated       []int
 	releaseDeferred []int
 	shouldRollback  bool
+	dryRun          bool
 }
 
 // Creates a portAllocationOperation, tracking a set of allocations & releases
-func StartOperation(pa Interface) *PortAllocationOperation {
+// If dryRun is specified, never actually allocate or release anything
+func StartOperation(pa Interface, dryRun bool) *PortAllocationOperation {
 	op := &PortAllocationOperation{}
 	op.pa = pa
 	op.allocated = []int{}
 	op.releaseDeferred = []int{}
 	op.shouldRollback = true
+	op.dryRun = dryRun
 	return op
 }
 
@@ -54,6 +57,10 @@ func (op *PortAllocationOperation) Finish() {
 
 // (Try to) undo any operations we did
 func (op *PortAllocationOperation) Rollback() []error {
+	if op.dryRun {
+		return nil
+	}
+
 	errors := []error{}
 
 	for _, allocated := range op.allocated {
@@ -73,6 +80,10 @@ func (op *PortAllocationOperation) Rollback() []error {
 // Note that even if this fails, we don't rollback; we always want to err on the side of over-allocation,
 // and Commit should be called _after_ the owner is written
 func (op *PortAllocationOperation) Commit() []error {
+	if op.dryRun {
+		return nil
+	}
+
 	errors := []error{}
 
 	for _, release := range op.releaseDeferred {
@@ -95,6 +106,19 @@ func (op *PortAllocationOperation) Commit() []error {
 
 // Allocates a port, and record it for future rollback
 func (op *PortAllocationOperation) Allocate(port int) error {
+	if op.dryRun {
+		if op.pa.Has(port) {
+			return ErrAllocated
+		}
+		for _, a := range op.allocated {
+			if port == a {
+				return ErrAllocated
+			}
+		}
+		op.allocated = append(op.allocated, port)
+		return nil
+	}
+
 	err := op.pa.Allocate(port)
 	if err == nil {
 		op.allocated = append(op.allocated, port)
@@ -104,6 +128,33 @@ func (op *PortAllocationOperation) Allocate(port int) error {
 
 // Allocates a port, and record it for future rollback
 func (op *PortAllocationOperation) AllocateNext() (int, error) {
+	if op.dryRun {
+		// Find the max element of the allocated ports array.
+		// If no ports are already being allocated by this operation,
+		// then choose a sensible guess for a dummy port number
+		var lastPort int
+		for _, allocatedPort := range op.allocated {
+			if allocatedPort > lastPort {
+				lastPort = allocatedPort
+			}
+		}
+		if len(op.allocated) == 0 {
+			lastPort = 32768
+		}
+
+		// Try to find the next non allocated port.
+		// If too many ports are full, just reuse one,
+		// since this is just a dummy value.
+		for port := lastPort + 1; port < 100; port++ {
+			err := op.Allocate(port)
+			if err == nil {
+				return port, nil
+			}
+		}
+		op.allocated = append(op.allocated, lastPort+1)
+		return lastPort + 1, nil
+	}
+
 	port, err := op.pa.AllocateNext()
 	if err == nil {
 		op.allocated = append(op.allocated, port)

pkg/registry/core/service/portallocator/operation_test.go

+/*
+Copyright 2018 The Kubernetes Authors.
+
+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.
+*/
+
+package portallocator
+
+import (
+	"testing"
+
+	"k8s.io/apimachinery/pkg/util/net"
+)
+
+// TestDryRunAllocate tests the Allocate function in dry run mode
+func TestDryRunAllocate(t *testing.T) {
+	pr, err := net.ParsePortRange("10000-10200")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Allocate some ports before calling
+	previouslyAllocated := []int{10000, 10010, 10020}
+	r := NewPortAllocator(*pr)
+	for _, port := range previouslyAllocated {
+		_ = r.Allocate(port)
+	}
+	freeAtStart := r.Free()
+
+	// Do some allocations with a dry run operation
+	toAllocate := []int{
+		10000,
+		10030,
+		10030,
+		10040,
+	}
+	expectedErrors := []error{
+		ErrAllocated,
+		nil,
+		ErrAllocated,
+		nil,
+	}
+	op := StartOperation(r, true)
+	for i, port := range toAllocate {
+		err := op.Allocate(port)
+		if err != expectedErrors[i] {
+			t.Errorf("%v: expected error %v but got %v", i, expectedErrors[i], err)
+		}
+	}
+
+	// Make sure no port allocations were actually made by the dry run
+	freeAtEnd := r.Free()
+	if freeAtStart != freeAtEnd {
+		t.Errorf("expected %v free ports but got %v", freeAtStart, freeAtEnd)
+	}
+}
+
+// TestDryRunAllocateNext tests the AllocateNext function in dry run mode
+func TestDryRunAllocateNext(t *testing.T) {
+	pr, err := net.ParsePortRange("10000-10200")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Allocate some ports before calling
+	previouslyAllocated := []int{10000, 10010, 10020}
+	r := NewPortAllocator(*pr)
+	for _, port := range previouslyAllocated {
+		_ = r.Allocate(port)
+	}
+	freeAtStart := r.Free()
+
+	// AllocateNext without a previously unused dry run operation
+	op := StartOperation(r, true)
+	port, err := op.AllocateNext()
+	if port == 0 {
+		t.Errorf("expected non zero port but got: %v", port)
+	}
+	if err != nil {
+		t.Errorf("expected no error but got: %v", err)
+	}
+
+	// Try to allocate the returned port using the same operation
+	if e, a := ErrAllocated, op.Allocate(port); e != a {
+		t.Errorf("expected %v but got: %v", e, a)
+	}
+
+	// AllocateNext with a previously used dry run operation
+	op = StartOperation(r, true)
+	_ = op.Allocate(12345)
+	port, err = op.AllocateNext()
+	if port == 0 {
+		t.Errorf("expected non zero port but got: %v", port)
+	}
+	if port == 12345 {
+		t.Errorf("expected port not to be 12345 but got %v", port)
+	}
+	if err != nil {
+		t.Errorf("expected no error but got: %v", err)
+	}
+
+	// Make sure no port allocations were actually made by the dry run
+	freeAtEnd := r.Free()
+	if freeAtStart != freeAtEnd {
+		t.Errorf("expected %v free ports but got %v", freeAtStart, freeAtEnd)
+	}
+}

pkg/registry/core/service/storage/BUILD

@@ -39,6 +39,7 @@ go_test(
         "//staging/src/k8s.io/apiserver/pkg/registry/generic/testing:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/rest:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/storage/etcd/testing:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/dryrun:go_default_library",
     ],
 )
 
@@ -74,6 +75,7 @@ go_library(
         "//staging/src/k8s.io/apiserver/pkg/registry/generic:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/generic/registry:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/registry/rest:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/dryrun:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
     ],

pkg/registry/core/service/storage/rest.go

@@ -37,6 +37,7 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/rest"
+	"k8s.io/apiserver/pkg/util/dryrun"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 
 	apiservice "k8s.io/kubernetes/pkg/api/service"
@@ -170,13 +171,15 @@ func (rs *REST) Create(ctx context.Context, obj runtime.Object, createValidation
 	}()
 
 	var err error
-	if service.Spec.Type != api.ServiceTypeExternalName {
-		if releaseServiceIP, err = initClusterIP(service, rs.serviceIPs); err != nil {
-			return nil, err
+	if !dryrun.IsDryRun(options.DryRun) {
+		if service.Spec.Type != api.ServiceTypeExternalName {
+			if releaseServiceIP, err = initClusterIP(service, rs.serviceIPs); err != nil {
+				return nil, err
+			}
 		}
 	}
 
-	nodePortOp := portallocator.StartOperation(rs.serviceNodePorts)
+	nodePortOp := portallocator.StartOperation(rs.serviceNodePorts, dryrun.IsDryRun(options.DryRun))
 	defer nodePortOp.Finish()
 
 	if service.Spec.Type == api.ServiceTypeNodePort || service.Spec.Type == api.ServiceTypeLoadBalancer {
@@ -222,13 +225,32 @@ func (rs *REST) Delete(ctx context.Context, id string, options *metav1.DeleteOpt
 
 	svc := obj.(*api.Service)
 
-	// TODO: can leave dangling endpoints, and potentially return incorrect
-	// endpoints if a new service is created with the same name
-	_, _, err = rs.endpoints.Delete(ctx, id, &metav1.DeleteOptions{})
-	if err != nil && !errors.IsNotFound(err) {
-		return nil, false, err
+	// Only perform the cleanup if this is a non-dryrun deletion
+	if !dryrun.IsDryRun(options.DryRun) {
+		// TODO: can leave dangling endpoints, and potentially return incorrect
+		// endpoints if a new service is created with the same name
+		_, _, err = rs.endpoints.Delete(ctx, id, &metav1.DeleteOptions{})
+		if err != nil && !errors.IsNotFound(err) {
+			return nil, false, err
+		}
+
+		rs.releaseAllocatedResources(svc)
 	}
 
+	// TODO: this is duplicated from the generic storage, when this wrapper is fully removed we can drop this
+	details := &metav1.StatusDetails{
+		Name: svc.Name,
+		UID:  svc.UID,
+	}
+	if info, ok := genericapirequest.RequestInfoFrom(ctx); ok {
+		details.Group = info.APIGroup
+		details.Kind = info.Resource // legacy behavior
+	}
+	status := &metav1.Status{Status: metav1.StatusSuccess, Details: details}
+	return status, true, nil
+}
+
+func (rs *REST) releaseAllocatedResources(svc *api.Service) {
 	if helper.IsServiceIPSet(svc) {
 		rs.serviceIPs.Release(net.ParseIP(svc.Spec.ClusterIP))
 	}
@@ -251,18 +273,6 @@ func (rs *REST) Delete(ctx context.Context, id string, options *metav1.DeleteOpt
 			}
 		}
 	}
-
-	// TODO: this is duplicated from the generic storage, when this wrapper is fully removed we can drop this
-	details := &metav1.StatusDetails{
-		Name: svc.Name,
-		UID:  svc.UID,
-	}
-	if info, ok := genericapirequest.RequestInfoFrom(ctx); ok {
-		details.Group = info.APIGroup
-		details.Kind = info.Resource // legacy behavior
-	}
-	status := &metav1.Status{Status: metav1.StatusSuccess, Details: details}
-	return status, true, nil
 }
 
 // externalTrafficPolicyUpdate adjusts ExternalTrafficPolicy during service update if needed.
@@ -358,19 +368,21 @@ func (rs *REST) Update(ctx context.Context, name string, objInfo rest.UpdatedObj
 		}
 	}()
 
-	nodePortOp := portallocator.StartOperation(rs.serviceNodePorts)
+	nodePortOp := portallocator.StartOperation(rs.serviceNodePorts, dryrun.IsDryRun(options.DryRun))
 	defer nodePortOp.Finish()
 
-	// Update service from ExternalName to non-ExternalName, should initialize ClusterIP.
-	if oldService.Spec.Type == api.ServiceTypeExternalName && service.Spec.Type != api.ServiceTypeExternalName {
-		if releaseServiceIP, err = initClusterIP(service, rs.serviceIPs); err != nil {
-			return nil, false, err
+	if !dryrun.IsDryRun(options.DryRun) {
+		// Update service from ExternalName to non-ExternalName, should initialize ClusterIP.
+		if oldService.Spec.Type == api.ServiceTypeExternalName && service.Spec.Type != api.ServiceTypeExternalName {
+			if releaseServiceIP, err = initClusterIP(service, rs.serviceIPs); err != nil {
+				return nil, false, err
+			}
 		}
-	}
-	// Update service from non-ExternalName to ExternalName, should release ClusterIP if exists.
-	if oldService.Spec.Type != api.ServiceTypeExternalName && service.Spec.Type == api.ServiceTypeExternalName {
-		if helper.IsServiceIPSet(oldService) {
-			rs.serviceIPs.Release(net.ParseIP(oldService.Spec.ClusterIP))
+		// Update service from non-ExternalName to ExternalName, should release ClusterIP if exists.
+		if oldService.Spec.Type != api.ServiceTypeExternalName && service.Spec.Type == api.ServiceTypeExternalName {
+			if helper.IsServiceIPSet(oldService) {
+				rs.serviceIPs.Release(net.ParseIP(oldService.Spec.ClusterIP))
+			}
 		}
 	}
 	// Update service from NodePort or LoadBalancer to ExternalName or ClusterIP, should release NodePort if exists.