Default use random port allocator for runtime controllers (fluid-clou…

…dnative#2580) * Validate port allocating policy before setting up Signed-off-by: dongyun.xzh <[email protected]> * Expose configurable option for port allocating policy Signed-off-by: dongyun.xzh <[email protected]> * Fix unit tests Signed-off-by: dongyun.xzh <[email protected]> * Fix static check Signed-off-by: dongyun.xzh <[email protected]> * Fix static check Signed-off-by: dongyun.xzh <[email protected]> * Fix comments for `SetupRuntimePortAllocator()` Signed-off-by: dongyun.xzh <[email protected]> * Support configurable port allocate policy Signed-off-by: dongyun.xzh <[email protected]> --------- Signed-off-by: dongyun.xzh <[email protected]>
cheyang · Feb 14, 2023 · ff850f6 · ff850f6
1 parent c69d31e
commit ff850f6
Show file tree

Hide file tree

Showing 21 changed files with 160 additions and 51 deletions.
diff --git a/charts/fluid/fluid/templates/controller/goosefsruntime_controller.yaml b/charts/fluid/fluid/templates/controller/goosefsruntime_controller.yaml
@@ -52,6 +52,7 @@ spec:
           - --runtime-workers={{ .Values.runtime.goosefs.runtimeWorkers }}
           - --pprof-addr=:6060
           - --enable-leader-election
+          - --port-allocate-policy={{ .Values.runtime.goosefs.portAllocatePolicy }}
         env:
           {{- if .Values.workdir }}
           - name: FLUID_WORKDIR

diff --git a/charts/fluid/fluid/templates/controller/jindoruntime_controller.yaml b/charts/fluid/fluid/templates/controller/jindoruntime_controller.yaml
@@ -50,6 +50,7 @@ spec:
           - --runtime-workers={{ .Values.runtime.jindo.runtimeWorkers }}
           - --pprof-addr=:6060
           - --enable-leader-election
+          - --port-allocate-policy={{ .Values.runtime.jindo.portAllocatePolicy }}
         env:
         {{- if .Values.workdir }}
         - name: FLUID_WORKDIR

diff --git a/charts/fluid/fluid/values.yaml b/charts/fluid/fluid/values.yaml
@@ -34,7 +34,7 @@ runtime:
     replicas: 1
     runtimeWorkers: 3
     portRange: 20000-26000
-    portAllocatePolicy: bitmap
+    portAllocatePolicy: random
     enabled: false
     init:
       image: fluidcloudnative/init-users:v0.9.0-fd767b9
@@ -48,6 +48,7 @@ runtime:
     replicas: 1
     runtimeWorkers: 3
     portRange: 18000-19999
+    portAllocatePolicy: random
     enabled: false
     engine: jindofsx
     queryUfsTotal: true
@@ -65,6 +66,7 @@ runtime:
     replicas: 1
     runtimeWorkers: 3
     portRange: 26000-32000
+    portAllocatePolicy: random
     enabled: false
     init:
       image: fluidcloudnative/init-users:v0.9.0-fd767b9

diff --git a/cmd/alluxio/app/alluxio.go b/cmd/alluxio/app/alluxio.go
@@ -74,7 +74,7 @@ func init() {
 	alluxioCmd.Flags().StringVar(&portRange, "runtime-node-port-range", "20000-25000", "Set available port range for Alluxio")
 	alluxioCmd.Flags().IntVar(&maxConcurrentReconciles, "runtime-workers", 3, "Set max concurrent workers for AlluxioRuntime controller")
 	alluxioCmd.Flags().StringVarP(&pprofAddr, "pprof-addr", "", "", "The address for pprof to use while exporting profiling results")
-	alluxioCmd.Flags().StringVar(&portAllocatePolicy, "port-allocate-policy", "bitmap", "Set port allocating policy, available choice is bitmap or random(default bitmap).")
+	alluxioCmd.Flags().StringVar(&portAllocatePolicy, "port-allocate-policy", "random", "Set port allocating policy, available choice is bitmap or random(default random).")
 }
 
 func handle() {
@@ -128,7 +128,12 @@ func handle() {
 	}
 	setupLog.Info("port range parsed", "port range", pr.String())
 
-	portallocator.SetupRuntimePortAllocatorWithType(mgr.GetClient(), pr, portallocator.AllocatePolicy(portAllocatePolicy), alluxio.GetReservedPorts)
+	err = portallocator.SetupRuntimePortAllocator(mgr.GetClient(), pr, portAllocatePolicy, alluxio.GetReservedPorts)
+	if err != nil {
+		setupLog.Error(err, "failed to setup runtime port allocator")
+		os.Exit(1)
+	}
+	setupLog.Info("Set up runtime port allocator", "policy", portAllocatePolicy)
 
 	setupLog.Info("starting alluxioruntime-controller")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {

diff --git a/cmd/goosefs/app/goosefs.go b/cmd/goosefs/app/goosefs.go
@@ -49,6 +49,7 @@ var (
 	portRange               string
 	maxConcurrentReconciles int
 	pprofAddr               string
+	portAllocatePolicy      string
 )
 
 var cmd = &cobra.Command{
@@ -73,6 +74,7 @@ func init() {
 	startCmd.Flags().StringVarP(&leaderElectionNamespace, "leader-election-namespace", "", "fluid-system", "The namespace in which the leader election resource will be created.")
 	startCmd.Flags().BoolVarP(&development, "development", "", true, "Enable development mode for fluid controller.")
 	startCmd.Flags().StringVar(&portRange, "runtime-node-port-range", "20000-25000", "Set available port range for GooseFS")
+	startCmd.Flags().StringVar(&portAllocatePolicy, "port-allocate-policy", "random", "Set port allocating policy, available choice is bitmap or random(default random).")
 	startCmd.Flags().StringVarP(&pprofAddr, "pprof-addr", "", "", "The address for pprof to use while exporting profiling results")
 	startCmd.Flags().IntVar(&maxConcurrentReconciles, "runtime-workers", 3, "Set max concurrent workers for GooseFSRuntime controller")
 	cmd.AddCommand(startCmd)
@@ -129,7 +131,12 @@ func handle() {
 	}
 	setupLog.Info("port range parsed", "port range", pr.String())
 
-	portallocator.SetupRuntimePortAllocator(mgr.GetClient(), pr, goosefs.GetReservedPorts)
+	err = portallocator.SetupRuntimePortAllocator(mgr.GetClient(), pr, portAllocatePolicy, goosefs.GetReservedPorts)
+	if err != nil {
+		setupLog.Error(err, "failed to setup runtime port allocator")
+		os.Exit(1)
+	}
+	setupLog.Info("Set up runtime port allocator", "policy", portAllocatePolicy)
 
 	setupLog.Info("starting goosefsruntime-controller")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {

diff --git a/cmd/jindo/app/jindo.go b/cmd/jindo/app/jindo.go
@@ -52,9 +52,10 @@ var (
 	development             bool
 	// The new mode
 	eventDriven             bool
-	portRange               string
 	maxConcurrentReconciles int
 	pprofAddr               string
+	portRange               string
+	portAllocatePolicy      string
 )
 
 var jindoCmd = &cobra.Command{
@@ -74,6 +75,7 @@ func init() {
 	jindoCmd.Flags().StringVarP(&leaderElectionNamespace, "leader-election-namespace", "", "fluid-system", "The namespace in which the leader election resource will be created.")
 	jindoCmd.Flags().BoolVarP(&development, "development", "", true, "Enable development mode for fluid controller.")
 	jindoCmd.Flags().StringVar(&portRange, "runtime-node-port-range", "18000-19999", "Set available port range for Jindo")
+	jindoCmd.Flags().StringVar(&portAllocatePolicy, "port-allocate-policy", "random", "Set port allocating policy, available choice is bitmap or random(default random).")
 	jindoCmd.Flags().IntVar(&maxConcurrentReconciles, "runtime-workers", 3, "Set max concurrent workers for JindoRuntime controller")
 	jindoCmd.Flags().BoolVar(&eventDriven, "event-driven", true, "The reconciler's loop strategy. if it's false, it indicates period driven.")
 	jindoCmd.Flags().StringVarP(&pprofAddr, "pprof-addr", "", "", "The address for pprof to use while exporting profiling results")
@@ -130,7 +132,12 @@ func handle() {
 	}
 	setupLog.Info("port range parsed", "port range", pr.String())
 
-	portallocator.SetupRuntimePortAllocator(mgr.GetClient(), pr, jindo.GetReservedPorts)
+	err = portallocator.SetupRuntimePortAllocator(mgr.GetClient(), pr, portAllocatePolicy, jindo.GetReservedPorts)
+	if err != nil {
+		setupLog.Error(err, "failed to setup runtime port allocator")
+		os.Exit(1)
+	}
+	setupLog.Info("Set up runtime port allocator", "policy", portAllocatePolicy)
 
 	setupLog.Info("starting jindoruntime-controller")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {

diff --git a/pkg/ddc/alluxio/master_internal_test.go b/pkg/ddc/alluxio/master_internal_test.go
@@ -113,8 +113,11 @@ func TestSetupMasterInternal(t *testing.T) {
 			},
 		},
 	}
-	portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-	err := gohook.Hook(kubectl.CreateConfigMapFromFile, mockExecCreateConfigMapFromFileErr, nil)
+	err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+	err = gohook.Hook(kubectl.CreateConfigMapFromFile, mockExecCreateConfigMapFromFileErr, nil)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -239,8 +242,11 @@ func TestGenerateAlluxioValueFile(t *testing.T) {
 		},
 	}
 
-	portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 50}, GetReservedPorts)
-	err := gohook.Hook(kubectl.CreateConfigMapFromFile, mockExecCreateConfigMapFromFileErr, nil)
+	err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 50}, "bitmap", GetReservedPorts)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
+	err = gohook.Hook(kubectl.CreateConfigMapFromFile, mockExecCreateConfigMapFromFileErr, nil)
 	if err != nil {
 		t.Fatal(err.Error())
 	}

diff --git a/pkg/ddc/alluxio/shutdown_engine_test.go b/pkg/ddc/alluxio/shutdown_engine_test.go
@@ -127,7 +127,10 @@ func TestShutdown(t *testing.T) {
 
 	pr := net.ParsePortRangeOrDie("20000-21000")
 
-	portallocator.SetupRuntimePortAllocator(nil, pr, dummy)
+	err = portallocator.SetupRuntimePortAllocator(nil, pr, "bitmap", dummy)
+	if err != nil {
+		t.Fatalf("failed to set up runtime port allocator due to %v", err)
+	}
 
 	var testCase = []struct {
 		expectedWorkers  int32

diff --git a/pkg/ddc/alluxio/shutdown_test.go b/pkg/ddc/alluxio/shutdown_test.go
@@ -335,7 +335,10 @@ func TestAlluxioEngineReleasePorts(t *testing.T) {
 				Log:       fake.NullLogger(),
 			}
 
-			portallocator.SetupRuntimePortAllocator(client, pr, GetReservedPorts)
+			err := portallocator.SetupRuntimePortAllocator(client, pr, "bitmap", GetReservedPorts)
+			if err != nil {
+				t.Fatal(err.Error())
+			}
 			allocator, _ := portallocator.GetRuntimePortAllocator()
 			patch1 := ApplyMethod(reflect.TypeOf(allocator), "ReleaseReservedPorts",
 				func(_ *portallocator.RuntimePortAllocator, ports []int) {

diff --git a/pkg/ddc/alluxio/transform_test.go b/pkg/ddc/alluxio/transform_test.go
@@ -647,7 +647,10 @@ func TestAlluxioEngine_allocateSinglePort(t *testing.T) {
 
 func TestAlluxioEngine_allocatePorts(t *testing.T) {
 	pr := net.ParsePortRangeOrDie("20000-21000")
-	portallocator.SetupRuntimePortAllocator(nil, pr, dummy)
+	err := portallocator.SetupRuntimePortAllocator(nil, pr, "bitmap", dummy)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
 	type fields struct {
 		runtime            *datav1alpha1.AlluxioRuntime
 		name               string

diff --git a/pkg/ddc/base/portallocator/port_allocator.go b/pkg/ddc/base/portallocator/port_allocator.go
@@ -17,6 +17,8 @@ limitations under the License.
 package portallocator
 
 import (
+	"fmt"
+
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/util/net"
@@ -31,6 +33,15 @@ const (
 	BitMap AllocatePolicy = "bitmap"
 )
 
+func ValidateEnum(allocatePolicyStr string) (AllocatePolicy, error) {
+	switch AllocatePolicy(allocatePolicyStr) {
+	case Random, BitMap:
+		return AllocatePolicy(allocatePolicyStr), nil
+	default:
+		return AllocatePolicy(allocatePolicyStr), fmt.Errorf("runtime-port-allocator can only be random or bitmap")
+	}
+}
+
 type BatchAllocatorInterface interface {
 	Allocate(int) error
 
@@ -60,9 +71,15 @@ type RuntimePortAllocator struct {
 // rpa is a global singleton of type RuntimePortAllocator
 var rpa *RuntimePortAllocator
 
-// SetupRuntimePortAllocator instantiates the global singleton rpa, use BitMap port allocating policy
-func SetupRuntimePortAllocator(client client.Client, pr *net.PortRange, getReservedPorts func(client client.Client) (ports []int, err error)) {
-	SetupRuntimePortAllocatorWithType(client, pr, BitMap, getReservedPorts)
+// SetupRuntimePortAllocator instantiates the global singleton rpa, set up port allocating policy according to the given allocatePolicyStr.
+// Currently the valid policies are either "random" or "bitmap".
+func SetupRuntimePortAllocator(client client.Client, pr *net.PortRange, allocatePolicyStr string, getReservedPorts func(client client.Client) (ports []int, err error)) error {
+	policy, err := ValidateEnum(allocatePolicyStr)
+	if err != nil {
+		return err
+	}
+	SetupRuntimePortAllocatorWithType(client, pr, policy, getReservedPorts)
+	return nil
 }
 
 // SetupRuntimePortAllocatorWithType instantiates the global singleton rpa with specified port allocating policy
@@ -90,7 +107,7 @@ func (alloc *RuntimePortAllocator) createAndRestorePortAllocator() (err error) {
 	case BitMap:
 		alloc.pa, err = newBitMapAllocator(alloc.pr, alloc.log)
 	default:
-		err = errors.New("allocate-port-policy can only be random or bitmap")
+		err = errors.New("runtime-port-allocator can only be random or bitmap")
 	}
 
 	if err != nil {

diff --git a/pkg/ddc/base/portallocator/port_allocator_test.go b/pkg/ddc/base/portallocator/port_allocator_test.go
@@ -33,17 +33,24 @@ var errDummy = func(client client.Client) (ports []int, err error) {
 
 func TestRuntimePortAllocatorWithError(t *testing.T) {
 	pr := net.ParsePortRangeOrDie("20000-21000")
-	SetupRuntimePortAllocator(nil, pr, errDummy)
+	err := SetupRuntimePortAllocator(nil, pr, "bitmap", errDummy)
+	if err != nil {
+		t.Fatalf("failed to setup runtime port allocator due to %v", err)
+	}
 
-	_, err := GetRuntimePortAllocator()
+	_, err = GetRuntimePortAllocator()
 	if err == nil {
 		t.Errorf("Expecetd error when GetRuntimePortAllocator")
 	}
 }
 
 func TestRuntimePortAllocator(t *testing.T) {
 	pr := net.ParsePortRangeOrDie("20000-21000")
-	SetupRuntimePortAllocator(nil, pr, dummy)
+	err := SetupRuntimePortAllocator(nil, pr, "bitmap", dummy)
+	if err != nil {
+		t.Errorf("get non-nil err when GetRuntimePortAllocator")
+		return
+	}
 
 	allocator, err := GetRuntimePortAllocator()
 	if err != nil {
@@ -71,7 +78,11 @@ func TestRuntimePortAllocator(t *testing.T) {
 
 func TestRuntimePortAllocatorRelease(t *testing.T) {
 	pr := net.ParsePortRangeOrDie("20000-20010")
-	SetupRuntimePortAllocator(nil, pr, dummy)
+	err := SetupRuntimePortAllocator(nil, pr, "bitmap", dummy)
+	if err != nil {
+		t.Errorf("get non-nil err when GetRuntimePortAllocator")
+		return
+	}
 
 	preservedPorts, _ := dummy(nil)
 

diff --git a/pkg/ddc/eac/master_internal_test.go b/pkg/ddc/eac/master_internal_test.go
@@ -117,10 +117,12 @@ func TestSetupMasterInternal(t *testing.T) {
 		runtime:   eacruntime,
 	}
 
-	portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-
+	err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
 	// check release found
-	err := gohook.Hook(helm.CheckRelease, mockExecCheckReleaseCommonFound, nil)
+	err = gohook.Hook(helm.CheckRelease, mockExecCheckReleaseCommonFound, nil)
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -244,9 +246,12 @@ func TestGenerateEACValueFile(t *testing.T) {
 		runtime:   eacruntime,
 	}
 
-	portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
+	err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+	if err != nil {
+		t.Fatal(err.Error())
+	}
 
-	err := gohook.Hook(kubectl.CreateConfigMapFromFile, mockCreateConfigMap, nil)
+	err = gohook.Hook(kubectl.CreateConfigMapFromFile, mockCreateConfigMap, nil)
 	if err != nil {
 		t.Fatal(err.Error())
 	}

diff --git a/pkg/ddc/eac/shutdown_test.go b/pkg/ddc/eac/shutdown_test.go
@@ -311,7 +311,10 @@ func TestEACEngineReleasePorts(t *testing.T) {
 				Log:       fake.NullLogger(),
 			}
 
-			portallocator.SetupRuntimePortAllocator(client, pr, GetReservedPorts)
+			err := portallocator.SetupRuntimePortAllocator(client, pr, "bitmap", GetReservedPorts)
+			if err != nil {
+				t.Fatalf("failed to set up runtime port allocator due to %v", err)
+			}
 			allocator, _ := portallocator.GetRuntimePortAllocator()
 			patch1 := ApplyMethod(reflect.TypeOf(allocator), "ReleaseReservedPorts",
 				func(_ *portallocator.RuntimePortAllocator, ports []int) {

diff --git a/pkg/ddc/eac/transform_port_test.go b/pkg/ddc/eac/transform_port_test.go
@@ -35,8 +35,11 @@ func TestTransformPortForMaster(t *testing.T) {
 	for _, test := range tests {
 		client := fake.NewFakeClientWithScheme(testScheme, test.runtime.DeepCopy())
 		engine := &EACEngine{Log: fake.NullLogger()}
-		portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-		err := engine.transformPortForMaster(test.runtime, test.eacValue)
+		err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+		if err != nil {
+			t.Fatalf("failed to set up runtime port allocator due to %v", err)
+		}
+		err = engine.transformPortForMaster(test.runtime, test.eacValue)
 		if err != nil {
 			t.Errorf("unexpected err %v", err)
 		}
@@ -70,8 +73,11 @@ func TestTransformPortForFuse(t *testing.T) {
 	for _, test := range tests {
 		client := fake.NewFakeClientWithScheme(testScheme, test.runtime.DeepCopy())
 		engine := &EACEngine{Log: fake.NullLogger()}
-		portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-		err := engine.transformPortForFuse(test.runtime, test.eacValue)
+		err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+		if err != nil {
+			t.Fatal(err.Error())
+		}
+		err = engine.transformPortForFuse(test.runtime, test.eacValue)
 		if err != nil {
 			t.Errorf("unexpected err %v", err)
 		}
@@ -105,8 +111,11 @@ func TestTransformPortForWorker(t *testing.T) {
 	for _, test := range tests {
 		client := fake.NewFakeClientWithScheme(testScheme, test.runtime.DeepCopy())
 		engine := &EACEngine{Log: fake.NullLogger()}
-		portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-		err := engine.transformPortForWorker(test.runtime, test.eacValue)
+		err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+		if err != nil {
+			t.Fatal(err.Error())
+		}
+		err = engine.transformPortForWorker(test.runtime, test.eacValue)
 		if err != nil {
 			t.Errorf("unexpected err %v", err)
 		}

diff --git a/pkg/ddc/eac/transform_test.go b/pkg/ddc/eac/transform_test.go
@@ -59,8 +59,11 @@ func TestEACEngine_transform(t *testing.T) {
 		ctrl.SetLogger(zap.New(func(o *zap.Options) {
 			o.Development = true
 		}))
-		portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, GetReservedPorts)
-		_, err := engine.transform(test.runtime)
+		err := portallocator.SetupRuntimePortAllocator(client, &net.PortRange{Base: 10, Size: 100}, "bitmap", GetReservedPorts)
+		if err != nil {
+			t.Fatal(err.Error())
+		}
+		_, err = engine.transform(test.runtime)
 		if err != nil {
 			t.Errorf("error %v", err)
 		}