OCPBUGS-122165: Reduce vCenter API load in the vSphere machine controller

cmd/vsphere/main.go

@@ -34,7 +34,32 @@ import (
 	"github.com/openshift/machine-api-operator/pkg/version"
 )
 
-const timeout = 10 * time.Minute
+// registerControllerFlags registers machine controller tuning flags on fs.
+func registerControllerFlags(fs *flag.FlagSet) (*int, *time.Duration) {
+	maxConcurrent := fs.Int("max-concurrent-reconciles", 10,
+		"Maximum number of parallel Machine reconciles. Higher values drain a "+
+			"cluster faster but issue the same vCenter calls faster; keep 10 for "+
+			"shared vCenter environments.")
+	sync := fs.Duration("sync-period", 30*time.Minute,
+		"Resync period for the machine controller cache. Larger values reduce steady-state "+
+			"vCenter API load; in-progress machines are requeued every 20s and do not depend "+
+			"on this. Values below 10m multiply vCenter load with no latency benefit.")
+	return maxConcurrent, sync
+}
+
+func validateMaxConcurrentReconciles(n int) error {
+	if n < 1 || n > 100 {
+		return fmt.Errorf("--max-concurrent-reconciles must be in [1, 100]; got %d", n)
+	}
+	return nil
+}
+
+func validateSyncPeriod(d time.Duration) error {
+	if d < time.Minute || d > time.Hour {
+		return fmt.Errorf("--sync-period must be in [1m, 1h]; got %s", d)
+	}
+	return nil
+}
 
 func main() {
 	var printVersion bool
@@ -99,6 +124,8 @@ func main() {
 		"The address for health checking.",
 	)
 
+	maxConcurrentReconciles, syncPeriod := registerControllerFlags(flag.CommandLine)
+
 	majorVersion := version.Version.Major
 
 	if majorVersion == 0 {
@@ -117,13 +144,20 @@ func main() {
 
 	flag.Parse()
 
+	if err := validateMaxConcurrentReconciles(*maxConcurrentReconciles); err != nil {
+		klog.Fatalf("%v", err)
+	}
+	if err := validateSyncPeriod(*syncPeriod); err != nil {
+		klog.Fatalf("%v", err)
+	}
+
 	if printVersion {
 		fmt.Println(version.String)
 		os.Exit(0)
 	}
 
 	cfg := config.GetConfigOrDie()
-	syncPeriod := timeout
+	syncPeriodRef := *syncPeriod
 
 	le := util.GetLeaderElectionConfig(cfg, configv1.LeaderElection{
 		Disable:       !*leaderElect,
@@ -136,7 +170,7 @@ func main() {
 		},
 		HealthProbeBindAddress: *healthAddr,
 		Cache: cache.Options{
-			SyncPeriod: &syncPeriod,
+			SyncPeriod: &syncPeriodRef,
 		},
 		LeaderElection:          *leaderElect,
 		LeaderElectionNamespace: *leaderElectResourceNamespace,
@@ -203,7 +237,8 @@ func main() {
 		klog.Fatalf("unable to add ipamv1beta1 to scheme: %v", err)
 	}
 
-	if err := capimachine.AddWithActuator(mgr, machineActuator, defaultMutableGate); err != nil {
+	if err := capimachine.AddWithActuatorOpts(mgr, machineActuator,
+		controller.Options{MaxConcurrentReconciles: *maxConcurrentReconciles}, defaultMutableGate); err != nil {
 		klog.Fatal(err)
 	}
 

cmd/vsphere/main_test.go

@@ -0,0 +1,90 @@
+package main
+
+import (
+	"flag"
+	"testing"
+	"time"
+)
+
+func TestSyncPeriodDefault(t *testing.T) {
+	// The flag is registered in main(); register it in a test flagset
+	// by calling the helper that wires flags.
+	fs := flag.NewFlagSet("test", flag.ContinueOnError)
+	_, syncPeriod := registerControllerFlags(fs)
+	if *syncPeriod != 30*time.Minute {
+		t.Errorf("default sync-period = %s, want 30m", *syncPeriod)
+	}
+}
+
+func TestSyncPeriodCustom(t *testing.T) {
+	fs := flag.NewFlagSet("test", flag.ContinueOnError)
+	_, syncPeriod := registerControllerFlags(fs)
+	if err := fs.Parse([]string{"--sync-period=45m"}); err != nil {
+		t.Fatalf("unexpected error parsing flags: %v", err)
+	}
+	if *syncPeriod != 45*time.Minute {
+		t.Errorf("expected sync-period = 45m, got %s", *syncPeriod)
+	}
+}
+
+func TestValidateSyncPeriod(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		val     time.Duration
+		wantErr bool
+	}{
+		{name: "default", val: 30 * time.Minute},
+		{name: "min", val: time.Minute},
+		{name: "max", val: time.Hour},
+		{name: "below min", val: 30 * time.Second, wantErr: true},
+		{name: "over max", val: 2 * time.Hour, wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateSyncPeriod(tc.val)
+			if (err != nil) != tc.wantErr {
+				t.Errorf("validateSyncPeriod(%s) err = %v, wantErr %v", tc.val, err, tc.wantErr)
+			}
+		})
+	}
+}
+
+func TestMaxConcurrentReconcilesDefault(t *testing.T) {
+	fs := flag.NewFlagSet("test", flag.ContinueOnError)
+	maxConcurrent, _ := registerControllerFlags(fs)
+	if *maxConcurrent != 10 {
+		t.Errorf("default max-concurrent-reconciles = %d, want 10", *maxConcurrent)
+	}
+}
+
+func TestMaxConcurrentReconcilesCustom(t *testing.T) {
+	fs := flag.NewFlagSet("test", flag.ContinueOnError)
+	maxConcurrent, _ := registerControllerFlags(fs)
+	if err := fs.Parse([]string{"--max-concurrent-reconciles=5"}); err != nil {
+		t.Fatalf("unexpected error parsing flags: %v", err)
+	}
+	if *maxConcurrent != 5 {
+		t.Errorf("expected max-concurrent-reconciles = 5, got %d", *maxConcurrent)
+	}
+}
+
+func TestValidateMaxConcurrentReconciles(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		val     int
+		wantErr bool
+	}{
+		{name: "default", val: 10},
+		{name: "min", val: 1},
+		{name: "max", val: 100},
+		{name: "zero", val: 0, wantErr: true},
+		{name: "negative", val: -1, wantErr: true},
+		{name: "over limit", val: 101, wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateMaxConcurrentReconciles(tc.val)
+			if (err != nil) != tc.wantErr {
+				t.Errorf("validateMaxConcurrentReconciles(%d) err = %v, wantErr %v", tc.val, err, tc.wantErr)
+			}
+		})
+	}
+}

pkg/controller/vsphere/actuator.go

@@ -5,6 +5,7 @@ package vsphere
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"k8s.io/component-base/featuregate"
@@ -33,6 +34,7 @@ type Actuator struct {
 	apiReader                runtimeclient.Reader
 	eventRecorder            events.EventRecorder
 	TaskIDCache              map[string]string
+	taskIDCacheMu            sync.Mutex
 	FeatureGates             featuregate.MutableFeatureGate
 	openshiftConfigNamespace string
 }
@@ -59,6 +61,28 @@ func NewActuator(params ActuatorParams) *Actuator {
 	}
 }
 
+func (a *Actuator) getTaskID(machineName string) (string, bool) {
+	a.taskIDCacheMu.Lock()
+	defer a.taskIDCacheMu.Unlock()
+	value, ok := a.TaskIDCache[machineName]
+	return value, ok
+}
+
+func (a *Actuator) setTaskID(machineName, taskID string) {
+	a.taskIDCacheMu.Lock()
+	defer a.taskIDCacheMu.Unlock()
+	if a.TaskIDCache == nil {
+		a.TaskIDCache = make(map[string]string)
+	}
+	a.TaskIDCache[machineName] = taskID
+}
+
+func (a *Actuator) clearTaskID(machineName string) {
+	a.taskIDCacheMu.Lock()
+	defer a.taskIDCacheMu.Unlock()
+	delete(a.TaskIDCache, machineName)
+}
+
 // Set corresponding event based on error. It also returns the original error
 // for convenience, so callers can do "return handleMachineError(...)".
 func (a *Actuator) handleMachineError(machine *machinev1.Machine, err error, eventAction string) error {
@@ -88,7 +112,7 @@ func (a *Actuator) Create(ctx context.Context, machine *machinev1.Machine) error
 
 	// Ensure we're not reconciling a stale machine by checking our task-id.
 	// This is a workaround for a cache race condition.
-	if val, ok := a.TaskIDCache[machine.Name]; ok {
+	if val, ok := a.getTaskID(machine.Name); ok {
 		if val != scope.providerStatus.TaskRef {
 			klog.Errorf("%s: machine object missing expected provider task ID, requeue", machine.GetName())
 			return &machinecontroller.RequeueAfterError{RequeueAfter: requeueAfterSeconds * time.Second}
@@ -99,7 +123,7 @@ func (a *Actuator) Create(ctx context.Context, machine *machinev1.Machine) error
 	err = newReconciler(scope).create()
 	// save the taskRef in our cache in case of any error with patch.
 	if scope.providerStatus.TaskRef != "" {
-		a.TaskIDCache[machine.Name] = scope.providerStatus.TaskRef
+		a.setTaskID(machine.Name, scope.providerStatus.TaskRef)
 	}
 	if err != nil {
 		fmtErr := fmt.Errorf(reconcilerFailFmt, machine.GetName(), createEventAction, err)
@@ -134,7 +158,7 @@ func (a *Actuator) Exists(ctx context.Context, machine *machinev1.Machine) (bool
 func (a *Actuator) Update(ctx context.Context, machine *machinev1.Machine) error {
 	klog.Infof("%s: actuator updating machine", machine.GetName())
 	// Cleanup TaskIDCache so we don't continually grow
-	delete(a.TaskIDCache, machine.Name)
+	a.clearTaskID(machine.Name)
 
 	scope, err := newMachineScope(machineScopeParams{
 		Context:                  ctx,
@@ -176,7 +200,7 @@ func (a *Actuator) Delete(ctx context.Context, machine *machinev1.Machine) error
 	klog.Infof("%s: actuator deleting machine", machine.GetName())
 	// Cleanup TaskIDCache so we don't continually grow
 	// Cleanup here as well in case Update() was never successfully called.
-	delete(a.TaskIDCache, machine.Name)
+	a.clearTaskID(machine.Name)
 
 	scope, err := newMachineScope(machineScopeParams{
 		Context:                  ctx,

pkg/controller/vsphere/actuator_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -422,3 +423,23 @@ func TestMachineEvents(t *testing.T) {
 		})
 	}
 }
+
+func TestTaskIDCacheConcurrentAccess(t *testing.T) {
+	actuator := &Actuator{TaskIDCache: make(map[string]string)}
+
+	const workers = 100
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			machineName := fmt.Sprintf("machine-%d", i)
+			actuator.setTaskID(machineName, "task")
+			if taskID, ok := actuator.getTaskID(machineName); !ok || taskID != "task" {
+				t.Errorf("getTaskID(%q) = %q, %t; want task, true", machineName, taskID, ok)
+			}
+			actuator.clearTaskID(machineName)
+		}(i)
+	}
+	wg.Wait()
+}