Revert "feat: official support for multiple controller server replicas (

#47)"

This reverts commit 8b4e410.

cmd/wekafsplugin/main.go

@@ -83,7 +83,6 @@ var (
 	allowInsecureHttps            = flag.Bool("allowinsecurehttps", false, "Allow insecure HTTPS connection without cert validation")
 	alwaysAllowSnapshotVolumes    = flag.Bool("alwaysallowsnapshotvolumes", false, "Allow snapshot-backed volumes even when Weka cluster doesn't support capacity enforcement")
 	usejsonlogging                = flag.Bool("usejsonlogging", false, "Use structured JSON logging rather than human-readable console log formatting")
-	maxRandomWaitIntervalSecs     = flag.Int("maxrandomwaitintervalsecs", 0, "Use random wait on prolonged controller operations to minimize risk of races. Used only if number of replicas is >=2")
 	maxConcurrentRequests         = flag.Int64("maxconcurrentrequests", 10, "Do not allow more than X requests in parallel")
 	grpcRequestTimeoutSeconds     = flag.Int("grpcrequesttimeoutseconds", 30, "Time out requests waiting in queue after X seconds")
 	// Set by the build process
@@ -196,7 +195,7 @@ func handle() {
 		*allowAutoFsCreation, *allowAutoFsExpansion,
 		*allowAutoSeedSnapshotCreation, *allowSnapshotsOfLegacyVolumes,
 		*suppressSnapshotsCapability, *suppressVolumeCloneCapability,
-		*allowInsecureHttps, *alwaysAllowSnapshotVolumes, *maxRandomWaitIntervalSecs, mutuallyExclusiveMountOptionsStrings,
+		*allowInsecureHttps, *alwaysAllowSnapshotVolumes, mutuallyExclusiveMountOptionsStrings,
 		*maxConcurrentRequests, *grpcRequestTimeoutSeconds)
 	driver, err := wekafs.NewWekaFsDriver(
 		*driverName, *nodeID, *endpoint, *maxVolumesPerNode, version, *debugPath, csiMode, *selinuxSupport, config)

deploy/helm/csi-wekafsplugin/templates/controllerserver-statefulset.yaml

@@ -133,9 +133,6 @@ spec:
           {{- if .Values.useJsonLogging }}
             - "--usejsonlogging"
           {{- end }}
-          {{- if ge (int (default 1 .Values.controller.replicas)) 2 }}
-            - "--maxrandomwaitintervalsecs={{ .Values.controller.maxRandomWaitIntervalSecs | default 0 }}"
-          {{- end}}
           {{- if .Values.controller.grpcRequestTimeoutSeconds }}
             - "--grpcrequesttimeoutseconds={{ .Values.controller.grpcRequestTimeoutSeconds | default "5" }}"
           {{- end }}

deploy/helm/csi-wekafsplugin/values.yaml

@@ -38,9 +38,6 @@ nodeSelector: {}
 controller:
   # -- Controller number of replicas
   replicas: 2
-  # -- when multiple controller replicas are triggered in parallel, each controller will issue a random wait of up to
-  #    `maxRandomWaitIntervalSecs` to avoid multiple API calls. Can be increased on very busy / large Weka clusters
-  maxRandomWaitIntervalSecs: 5
   # -- maximum concurrent operations per operation type (to avoid API starvation)
   maxConcurrentRequests: 5
   # -- Return GRPC Unavailable if request waits in queue for that long time (seconds)

pkg/wekafs/controllerserver.go

@@ -227,9 +227,6 @@ func (cs *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	// Check for maximum available capacity
 	capacity := req.GetCapacityRange().GetRequiredBytes()
 
-	// randomBackoff to minimize race conditions when multiple CSI controllers are run in parallel
-	waitRandomTime(ctx, cs.config.maxRandomWaitIntervalSecs)
-
 	// IDEMPOTENCE FLOW: If directory already exists, return the createResponse if size matches, or error
 	volExists, volMatchesCapacity, err := volumeExistsAndMatchesCapacity(ctx, volume, capacity)
 
@@ -346,9 +343,6 @@ func (cs *ControllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 		return DeleteVolumeError(ctx, codes.Internal, err.Error())
 	}
 
-	// randomBackoff to minimize race conditions when multiple CSI controllers are run in parallel
-	waitRandomTime(ctx, cs.config.maxRandomWaitIntervalSecs)
-
 	err = volume.Trash(ctx)
 	if os.IsNotExist(err) {
 		logger.Debug().Str("volume_id", volume.GetId()).Msg("Volume not found, but returning success for idempotence")
@@ -424,9 +418,6 @@ func (cs *ControllerServer) ControllerExpandVolume(ctx context.Context, req *csi
 		return ExpandVolumeError(ctx, codes.NotFound, fmt.Sprintf("Volume with id %s does not exist", req.GetVolumeId()))
 	}
 
-	// randomBackoff to minimize race conditions when multiple CSI controllers are run in parallel
-	waitRandomTime(ctx, cs.config.maxRandomWaitIntervalSecs)
-
 	maxStorageCapacity, err := volume.getMaxCapacity(ctx)
 	if err != nil {
 		return ExpandVolumeError(ctx, codes.Unknown, fmt.Sprintf("ExpandVolume: Cannot obtain free capacity for volume %s", volume.GetId()))
@@ -521,9 +512,6 @@ func (cs *ControllerServer) CreateSnapshot(ctx context.Context, req *csi.CreateS
 		return CreateSnapshotError(ctx, codes.FailedPrecondition, fmt.Sprintf("Could not find source volume %s", srcVolume.GetId()))
 	}
 
-	// randomBackoff to minimize race conditions when multiple CSI controllers are run in parallel
-	waitRandomTime(ctx, cs.config.maxRandomWaitIntervalSecs)
-
 	s, err := srcVolume.CreateSnapshot(ctx, snapName)
 	if err != nil {
 		return &csi.CreateSnapshotResponse{}, err
@@ -588,10 +576,6 @@ func (cs *ControllerServer) DeleteSnapshot(ctx context.Context, req *csi.DeleteS
 	if err != nil {
 		return DeleteSnapshotError(ctx, codes.Internal, fmt.Sprintln("Failed to initialize snapshot from ID", snapshotID, err.Error()))
 	}
-
-	// randomBackoff to minimize race conditions when multiple CSI controllers are run in parallel
-	waitRandomTime(ctx, cs.config.maxRandomWaitIntervalSecs)
-
 	err = existingSnap.Delete(ctx)
 	if err != nil {
 		return DeleteSnapshotError(ctx, codes.Internal, fmt.Sprintln("Failed to delete snapshot", snapshotID, err))

pkg/wekafs/driverconfig.go

@@ -30,7 +30,6 @@ type DriverConfig struct {
 	debugPath                     string
 	allowInsecureHttps            bool
 	alwaysAllowSnapshotVolumes    bool
-	maxRandomWaitIntervalSecs     int
 	mutuallyExclusiveOptions      []mutuallyExclusiveMountOptionSet
 	maxConcurrentRequests         int64
 	grpcRequestTimeout            time.Duration
@@ -48,7 +47,7 @@ func (dc *DriverConfig) Log() {
 func NewDriverConfig(dynamicVolPath, VolumePrefix, SnapshotPrefix, SeedSnapshotPrefix, debugPath string,
 	allowAutoFsCreation, allowAutoFsExpansion, allowAutoSeedSnapshotCreation, allowSnapshotsOfLegacyVolumes bool,
 	suppressnapshotSupport, suppressVolumeCloneSupport,
-	allowInsecureHttps, alwaysAllowSnapshotVolumes bool, maxRandomWaitIntervalSecs int,
+	allowInsecureHttps, alwaysAllowSnapshotVolumes bool,
 	mutuallyExclusiveMountOptions MutuallyExclusiveMountOptsStrings, maxConcurrentRequests int64,
 	grpcRequestTimeoutSeconds int) *DriverConfig {
 
@@ -77,7 +76,6 @@ func NewDriverConfig(dynamicVolPath, VolumePrefix, SnapshotPrefix, SeedSnapshotP
 		debugPath:                     debugPath,
 		allowInsecureHttps:            allowInsecureHttps,
 		alwaysAllowSnapshotVolumes:    alwaysAllowSnapshotVolumes,
-		maxRandomWaitIntervalSecs:     maxRandomWaitIntervalSecs,
 		mutuallyExclusiveOptions:      MutuallyExclusiveMountOptions,
 		maxConcurrentRequests:         maxConcurrentRequests,
 		grpcRequestTimeout:            grpcRequestTimeout,

pkg/wekafs/utilities.go

@@ -15,7 +15,6 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	timestamp "google.golang.org/protobuf/types/known/timestamppb"
-	"math/rand"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -470,15 +469,3 @@ func volumeExistsAndMatchesCapacity(ctx context.Context, v *Volume, capacity int
 	matches := reportedCapacity == capacity
 	return exists, matches, err
 }
-
-func waitRandomTime(ctx context.Context, maxInterval int) {
-	if maxInterval == 0 {
-		return
-	}
-	logger := log.Ctx(ctx)
-	interval := rand.Intn(maxInterval)
-	if interval > 0 {
-		logger.Trace().Int("wait_seconds", interval).Msg("Issuing a random backoff before performing operation")
-		time.Sleep(time.Second * time.Duration(interval))
-	}
-}