From 9a6e0f204d4bddadfbbb6f90fda35f3f6538d9d4 Mon Sep 17 00:00:00 2001 From: Joseph Callen Date: Wed, 16 Sep 2026 20:58:31 -0400 Subject: [PATCH] vsphere: optimize reconciliation and vCenter API usage Reduce redundant vCenter requests through session, tag, network, and power-state caching; share reconciliation paths; add request metrics; and improve machine recovery, concurrency, and resync behavior. --- cmd/vsphere/main.go | 43 +- cmd/vsphere/main_test.go | 90 +++ pkg/controller/vsphere/actuator.go | 66 +- pkg/controller/vsphere/actuator_test.go | 298 +++++++ pkg/controller/vsphere/reconciler.go | 443 ++++++----- pkg/controller/vsphere/reconciler_test.go | 745 +++++++++++++++--- pkg/controller/vsphere/session/session.go | 119 +-- .../vsphere/session/session_test.go | 44 ++ .../vsphere/session/tag_ids_caching_client.go | 224 +++--- .../session/test_ids_caching_client_test.go | 364 +++++---- .../vsphere/session/transport_metrics.go | 105 +++ .../vsphere/session/transport_metrics_test.go | 266 +++++++ pkg/metrics/metrics.go | 10 +- pkg/metrics/metrics_test.go | 24 + 14 files changed, 2235 insertions(+), 606 deletions(-) create mode 100644 cmd/vsphere/main_test.go create mode 100644 pkg/controller/vsphere/session/transport_metrics.go create mode 100644 pkg/controller/vsphere/session/transport_metrics_test.go create mode 100644 pkg/metrics/metrics_test.go diff --git a/cmd/vsphere/main.go b/cmd/vsphere/main.go index 27d602810c..69146f459a 100644 --- a/cmd/vsphere/main.go +++ b/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) } diff --git a/cmd/vsphere/main_test.go b/cmd/vsphere/main_test.go new file mode 100644 index 0000000000..65de33af11 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/pkg/controller/vsphere/actuator.go b/pkg/controller/vsphere/actuator.go index 6b9aad7461..bb4ce24742 100644 --- a/pkg/controller/vsphere/actuator.go +++ b/pkg/controller/vsphere/actuator.go @@ -5,12 +5,11 @@ package vsphere import ( "context" "fmt" - "time" + "sync" "k8s.io/component-base/featuregate" machinev1 "github.com/openshift/api/machine/v1beta1" - machinecontroller "github.com/openshift/machine-api-operator/pkg/controller/machine" corev1 "k8s.io/api/core/v1" "k8s.io/client-go/tools/events" "k8s.io/klog/v2" @@ -18,13 +17,12 @@ import ( ) const ( - scopeFailFmt = "%s: failed to create scope for machine: %v" - reconcilerFailFmt = "%s: reconciler failed to %s machine: %w" - createEventAction = "Create" - updateEventAction = "Update" - deleteEventAction = "Delete" - noEventAction = "" - requeueAfterSeconds = 20 + scopeFailFmt = "%s: failed to create scope for machine: %v" + reconcilerFailFmt = "%s: reconciler failed to %s machine: %w" + createEventAction = "Create" + updateEventAction = "Update" + deleteEventAction = "Delete" + noEventAction = "" ) // Actuator is responsible for performing machine reconciliation. @@ -33,6 +31,7 @@ type Actuator struct { apiReader runtimeclient.Reader eventRecorder events.EventRecorder TaskIDCache map[string]string + taskIDCacheMu sync.Mutex FeatureGates featuregate.MutableFeatureGate openshiftConfigNamespace string } @@ -59,6 +58,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 { @@ -86,20 +107,25 @@ func (a *Actuator) Create(ctx context.Context, machine *machinev1.Machine) error return a.handleMachineError(machine, fmtErr, createEventAction) } - // 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 != scope.providerStatus.TaskRef { - klog.Errorf("%s: machine object missing expected provider task ID, requeue", machine.GetName()) - return &machinecontroller.RequeueAfterError{RequeueAfter: requeueAfterSeconds * time.Second} - } + // If the task we last submitted for this machine (tracked in the in-memory + // cache) differs from what the Machine object reflects, the status patch + // that would have persisted it may have failed, or the client cache may be + // stale. The cache always holds the most recently submitted task (it is + // updated on every reconcile, before the patch), so recover it and reconcile + // that task instead of requeueing forever (which permanently wedged + // creation) or reprocessing a stale reference (which could submit a + // duplicate clone or power-on). + if cachedTaskRef, ok := a.getTaskID(machine.Name); ok && cachedTaskRef != scope.providerStatus.TaskRef { + klog.Infof("%s: recovering task reference %q from cache; Machine status reflects %q", machine.GetName(), cachedTaskRef, scope.providerStatus.TaskRef) + scope.providerStatus.TaskRef = cachedTaskRef } var retErr error err = newReconciler(scope).create() - // save the taskRef in our cache in case of any error with patch. + // Remember the submitted task reference even if the patch below fails, so a + // retry reconciles the in-flight task instead of submitting a second clone. 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 +160,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 +202,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, diff --git a/pkg/controller/vsphere/actuator_test.go b/pkg/controller/vsphere/actuator_test.go index 38118eed5a..fb1d129c3f 100644 --- a/pkg/controller/vsphere/actuator_test.go +++ b/pkg/controller/vsphere/actuator_test.go @@ -5,6 +5,8 @@ import ( "fmt" "net" "path/filepath" + "strings" + "sync" "testing" "time" @@ -13,14 +15,18 @@ import ( . "github.com/onsi/gomega" configv1 "github.com/openshift/api/config/v1" machinev1 "github.com/openshift/api/machine/v1beta1" + "github.com/vmware/govmomi/object" "github.com/vmware/govmomi/simulator" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/tools/events" ipamv1beta1 "sigs.k8s.io/cluster-api/api/ipam/v1beta1" //nolint:staticcheck "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" "sigs.k8s.io/controller-runtime/pkg/envtest" "sigs.k8s.io/controller-runtime/pkg/manager" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" @@ -422,3 +428,295 @@ 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() +} + +// TestActuatorCreateTaskRefLifecycle verifies the actuator's clone task +// reference bookkeeping: the reference is remembered even when the status patch +// that would persist it is denied, so a retry reconciles the same task instead +// of requeueing forever or submitting a duplicate clone (OCPBUGS-100316). +func TestActuatorCreateTaskRefLifecycle(t *testing.T) { + model, session, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + host, port, err := net.SplitHostPort(server.URL.Host) + if err != nil { + t.Fatal(err) + } + + credentialsSecretUsername := fmt.Sprintf("%s.username", host) + credentialsSecretPassword := fmt.Sprintf("%s.password", host) + password, _ := server.URL.User.Password() + namespace := "test" + + vm := model.Map().Any("VirtualMachine").(*simulator.VirtualMachine) + vm.Config.Version = minimumHWVersionString + + credentialsSecret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "test", Namespace: namespace}, + Data: map[string][]byte{ + credentialsSecretUsername: []byte(server.URL.User.Username()), + credentialsSecretPassword: []byte(password), + }, + } + configMap := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: OpenshiftConfigManagedConfigMap, Namespace: openshiftConfigNamespaceForTest}, + Data: map[string]string{OpenshiftConfigManagedCloudConfigKey: fmt.Sprintf(testConfigFmt, port, "test", namespace)}, + } + userDataSecret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "vsphere-ignition", Namespace: namespace}, + Data: map[string][]byte{userDataSecretKey: []byte("{}")}, + } + + newMachine := func(name string) *machinev1.Machine { + providerSpec, err := RawExtensionFromProviderSpec(&machinev1.VSphereMachineProviderSpec{ + Template: vm.Name, + Workspace: &machinev1.Workspace{Server: host}, + CredentialsSecret: &corev1.LocalObjectReference{Name: "test"}, + UserDataSecret: &corev1.LocalObjectReference{Name: "vsphere-ignition"}, + DiskGiB: 10, + }) + if err != nil { + t.Fatal(err) + } + return &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{machinev1.MachineClusterIDLabel: "CLUSTERID"}, + }, + Spec: machinev1.MachineSpec{ProviderSpec: machinev1.ProviderSpec{Value: providerSpec}}, + Status: machinev1.MachineStatus{}, + } + } + + denyStatusPatch := func(base client.WithWatch) client.WithWatch { + return interceptor.NewClient(base, interceptor.Funcs{ + SubResourcePatch: func(_ context.Context, _ client.Client, _ string, _ client.Object, _ client.Patch, _ ...client.SubResourcePatchOption) error { + return fmt.Errorf("admission webhook denied the request") + }, + }) + } + + gates, err := testutils.NewDefaultMutableFeatureGate() + if err != nil { + t.Fatalf("unexpected error setting up feature gates: %v", err) + } + + waitForCloneTask := func(g *WithT, taskRef string) { + moTask, err := session.GetTask(context.TODO(), taskRef) + g.Expect(err).ToNot(HaveOccurred()) + if moTask != nil { + g.Expect(object.NewTask(session.Client.Client, moTask.Reference()).Wait(context.TODO())).To(Succeed()) + } + } + + t.Run("a denied status patch retains the clone task reference", func(t *testing.T) { + g := NewWithT(t) + machine := newMachine("patch-denied") + + base := fake.NewClientBuilder().WithScheme(scheme.Scheme). + WithStatusSubresource(machine). + WithRuntimeObjects(credentialsSecret, configMap, userDataSecret, machine). + Build() + + taskIDCache := map[string]string{} + actuator := NewActuator(ActuatorParams{ + Client: denyStatusPatch(base), + APIReader: base, + EventRecorder: events.NewFakeRecorder(10), + TaskIDCache: taskIDCache, + OpenshiftConfigNamespace: openshiftConfigNamespaceForTest, + FeatureGates: gates, + }) + + err := actuator.Create(context.Background(), machine) + g.Expect(err).To(HaveOccurred()) + // The clone identity must survive the failed patch so the next reconcile + // reconciles the same task rather than submitting a duplicate clone. + g.Expect(taskIDCache).To(HaveKey(machine.Name)) + g.Expect(taskIDCache[machine.Name]).ToNot(BeEmpty()) + + waitForCloneTask(g, taskIDCache[machine.Name]) + }) + + t.Run("a successful patch caches the clone task reference", func(t *testing.T) { + g := NewWithT(t) + machine := newMachine("patch-ok") + + c := fake.NewClientBuilder().WithScheme(scheme.Scheme). + WithStatusSubresource(machine). + WithRuntimeObjects(credentialsSecret, configMap, userDataSecret, machine). + Build() + + taskIDCache := map[string]string{} + actuator := NewActuator(ActuatorParams{ + Client: c, + APIReader: c, + EventRecorder: events.NewFakeRecorder(10), + TaskIDCache: taskIDCache, + OpenshiftConfigNamespace: openshiftConfigNamespaceForTest, + FeatureGates: gates, + }) + + g.Expect(actuator.Create(context.Background(), machine)).To(Succeed()) + g.Expect(taskIDCache).To(HaveKey(machine.Name)) + g.Expect(taskIDCache[machine.Name]).ToNot(BeEmpty()) + + waitForCloneTask(g, taskIDCache[machine.Name]) + }) + + t.Run("a lost task reference is reconciled on retry without a second clone", func(t *testing.T) { + g := NewWithT(t) + + // Hold the clone task in-flight so the cloned VM is not yet discoverable + // in vCenter - the exact window in which a lost TaskRef previously caused + // a duplicate clone submission. + simulator.TaskDelay.MethodDelay = map[string]int{"CloneVm": 2000, "LockHandoff": 0} + defer func() { simulator.TaskDelay = simulator.DelayConfig{} }() + + machine := newMachine("inflight") + base := fake.NewClientBuilder().WithScheme(scheme.Scheme). + WithStatusSubresource(machine). + WithRuntimeObjects(credentialsSecret, configMap, userDataSecret, machine). + Build() + + taskIDCache := map[string]string{} + actuator := NewActuator(ActuatorParams{ + Client: denyStatusPatch(base), + APIReader: base, + EventRecorder: events.NewFakeRecorder(10), + TaskIDCache: taskIDCache, + OpenshiftConfigNamespace: openshiftConfigNamespaceForTest, + FeatureGates: gates, + }) + + vmCountBefore := len(model.Map().All("VirtualMachine")) + cloneTasksBefore := countTasksMatching(model, cloneVmTaskDescriptionId) + + // First reconcile submits the clone; the status patch is denied so the + // TaskRef lives only in the cache. + g.Expect(actuator.Create(context.Background(), machine)).To(HaveOccurred()) + g.Expect(taskIDCache).To(HaveKey(machine.Name)) + g.Expect(countTasksMatching(model, cloneVmTaskDescriptionId)).To(Equal(cloneTasksBefore+1), "first reconcile must submit exactly one clone") + // The clone is still running, so the VM is not yet discoverable. This + // confirms the in-flight window is actually reproduced. + g.Expect(model.Map().All("VirtualMachine")).To(HaveLen(vmCountBefore), "clone should still be in-flight (VM not yet created)") + + // Retry the way the machine controller would: with a freshly read + // Machine. Because the status patch was denied, the persisted object has + // no TaskRef, so the actuator must recover it from the cache rather than + // submit a second clone. (Reusing the in-memory pointer would hide the + // bug, since PatchMachine mutates it before the failed patch.) + fresh := &machinev1.Machine{} + g.Expect(base.Get(context.Background(), client.ObjectKeyFromObject(machine), fresh)).To(Succeed()) + g.Expect(fresh.Status.ProviderStatus).To(BeNil(), "denied status patch must not have persisted a TaskRef") + g.Expect(actuator.Create(context.Background(), fresh)).To(HaveOccurred()) + g.Expect(countTasksMatching(model, cloneVmTaskDescriptionId)).To(Equal(cloneTasksBefore+1), "retry must not submit a second clone") + + // Let the clone finish before teardown. + waitForCloneTask(g, taskIDCache[machine.Name]) + }) + + t.Run("a stale nonempty task reference is reconciled from the cache on retry", func(t *testing.T) { + g := NewWithT(t) + + machine := newMachine("stale-nonempty") + base := fake.NewClientBuilder().WithScheme(scheme.Scheme). + WithStatusSubresource(machine). + WithRuntimeObjects(credentialsSecret, configMap, userDataSecret, machine). + Build() + + taskIDCache := map[string]string{} + + // Clone successfully first, so the Machine object and the cache both + // track the clone task and the (powered-off) VM exists. + allowActuator := NewActuator(ActuatorParams{ + Client: base, + APIReader: base, + EventRecorder: events.NewFakeRecorder(10), + TaskIDCache: taskIDCache, + OpenshiftConfigNamespace: openshiftConfigNamespaceForTest, + FeatureGates: gates, + }) + g.Expect(allowActuator.Create(context.Background(), machine)).To(Succeed()) + cloneTaskRef := taskIDCache[machine.Name] + g.Expect(cloneTaskRef).ToNot(BeEmpty()) + waitForCloneTask(g, cloneTaskRef) + + // Hold power-on in-flight and deny status patches, so the power-on task + // is submitted but never persisted onto the Machine object. + simulator.TaskDelay.MethodDelay = map[string]int{"PowerOnMultiVM": 2000, "LockHandoff": 0} + defer func() { simulator.TaskDelay = simulator.DelayConfig{} }() + + denyActuator := NewActuator(ActuatorParams{ + Client: denyStatusPatch(base), + APIReader: base, + EventRecorder: events.NewFakeRecorder(10), + TaskIDCache: taskIDCache, + OpenshiftConfigNamespace: openshiftConfigNamespaceForTest, + FeatureGates: gates, + }) + + powerOnBefore := countTasksMatching(model, powerOnTaskDescriptionID) + + // Reconcile the finished clone: submits a power-on task and advances the + // cache, but the denied patch leaves the Machine object on the clone task. + fresh1 := &machinev1.Machine{} + g.Expect(base.Get(context.Background(), client.ObjectKeyFromObject(machine), fresh1)).To(Succeed()) + g.Expect(denyActuator.Create(context.Background(), fresh1)).To(HaveOccurred()) + g.Expect(countTasksMatching(model, powerOnTaskDescriptionID)).To(Equal(powerOnBefore+1), "reconciling the finished clone must submit exactly one power-on") + g.Expect(taskIDCache[machine.Name]).ToNot(Equal(cloneTaskRef), "cache should have advanced to the power-on task") + + // Retry with a freshly read Machine, which still carries the stale clone + // task because the power-on patch was denied. The actuator must recover + // the newer power-on task from the cache instead of reprocessing the + // clone and submitting a second power-on. + fresh2 := &machinev1.Machine{} + g.Expect(base.Get(context.Background(), client.ObjectKeyFromObject(machine), fresh2)).To(Succeed()) + g.Expect(denyActuator.Create(context.Background(), fresh2)).To(HaveOccurred()) + g.Expect(countTasksMatching(model, powerOnTaskDescriptionID)).To(Equal(powerOnBefore+1), "retry must not submit a second power-on") + + // Let the power-on finish before teardown. + moTask, err := session.GetTask(context.TODO(), taskIDCache[machine.Name]) + g.Expect(err).ToNot(HaveOccurred()) + if moTask != nil { + g.Expect(object.NewTask(session.Client.Client, moTask.Reference()).Wait(context.TODO())).To(Succeed()) + } + }) +} + +// powerOnTaskDescriptionID is the DescriptionId of the task issued when powering +// on a VM via Datacenter.PowerOnVM in the simulator. +const powerOnTaskDescriptionID = "powerOnMultiVM" + +// countTasksMatching returns the number of tasks in the simulator inventory +// whose DescriptionId contains the given substring. +func countTasksMatching(model *simulator.Model, descriptionSubstring string) int { + count := 0 + for _, ref := range model.Map().AllReference("") { + if task, ok := ref.(*simulator.Task); ok && strings.Contains(task.Info.DescriptionId, descriptionSubstring) { + count++ + } + } + return count +} diff --git a/pkg/controller/vsphere/reconciler.go b/pkg/controller/vsphere/reconciler.go index 7af805f43a..86f8bdbd9f 100644 --- a/pkg/controller/vsphere/reconciler.go +++ b/pkg/controller/vsphere/reconciler.go @@ -85,6 +85,39 @@ func newReconciler(scope *machineScope) *Reconciler { } } +// addVMGroupAndPowerOn restores VM group membership (if configured) and +// powers the machine on, persisting the power-on task ref in the provider +// status. Shared by the recovered-VM path and the completed-clone path in +// create() so the two cannot drift apart. +func (r *Reconciler) addVMGroupAndPowerOn(kind string) error { + if r.machineScope.providerSpec.Workspace.VMGroup != "" { + klog.Infof("Adding %s machine: %s to vm group: %s", kind, r.machine.Name, r.machineScope.providerSpec.Workspace.VMGroup) + if err := modifyVMGroup(r.machineScope, false); err != nil { + var taskError task.Error + if errors.As(err, &taskError) { + return fmt.Errorf("could not update VM Group membership: %w", taskError) + } + return fmt.Errorf("could not update VM Group membership: %w", err) + } + } + klog.Infof("Powering on %s machine: %v", kind, r.machine.Name) + task, err := powerOn(r.machineScope) + if err != nil { + metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ + Name: r.machine.Name, + Namespace: r.machine.Namespace, + Reason: "PowerOn task finished with error", + }) + conditionFailed := conditionFailed() + conditionFailed.Message = err.Error() + if statusError := setProviderStatus(task, conditionFailed, r.machineScope, nil); statusError != nil { + return fmt.Errorf("failed to set provider status: %w", err) + } + return fmt.Errorf("%v: failed to power on machine: %w", r.machine.GetName(), err) + } + return setProviderStatus(task, conditionSuccess(), r.machineScope, nil) +} + // create creates machine if it does not exists. func (r *Reconciler) create() error { if err := validateMachine(*r.machine); err != nil { @@ -130,62 +163,84 @@ func (r *Reconciler) create() error { return fmt.Errorf("%v: not connected to a vCenter", r.machine.GetName()) } - // Attempt to power on instance in situation where we alredy cloned the instance and lost taskRef. - klog.V(4).Infof("%v: InstanceState is: %q", r.machine.GetName(), ptr.Deref(r.machineScope.providerStatus.InstanceState, "")) - if types.VirtualMachinePowerState(ptr.Deref(r.machineScope.providerStatus.InstanceState, "")) == types.VirtualMachinePowerStatePoweredOff { - klog.Infof("Powering on cloned machine without taskID: %v", r.machine.Name) + // A missing TaskRef usually means the VM has not been cloned yet. It can + // also mean we cloned the VM successfully but lost the TaskRef because + // the status patch that would have persisted it failed (for example, a + // transient admission-webhook denial during install). Look the VM up + // directly in vCenter before cloning so that a lost TaskRef never + // results in a duplicate VM: if the VM already exists we adopt it and + // power it on, otherwise we clone the template. + if _, err := findVM(r.machineScope); err != nil { + if !isNotFound(err) { + metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ + Name: r.machine.Name, + Namespace: r.machine.Namespace, + Reason: "FindVM finished with error", + }) + return err + } - task, err := powerOn(r.machineScope) + klog.Infof("%v: cloning", r.machine.GetName()) + // A new clone has a different identity. Clear values from a VM that + // may have disappeared so the next update records the new VM identity. + r.machine.Spec.ProviderID = nil + r.providerStatus.InstanceID = nil + task, err := clone(r.machineScope) if err != nil { metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ Name: r.machine.Name, Namespace: r.machine.Namespace, - Reason: "PowerOn task finished with error", + Reason: "Clone task finished with error", }) - conditionFailed := conditionFailed() conditionFailed.Message = err.Error() statusError := setProviderStatus(task, conditionFailed, r.machineScope, nil) if statusError != nil { return fmt.Errorf("failed to set provider status: %w", err) } - - return fmt.Errorf("%v: failed to power on machine: %w", r.machine.GetName(), err) + return err } - return setProviderStatus(task, conditionSuccess(), r.machineScope, nil) } - klog.Infof("%v: cloning", r.machine.GetName()) - task, err := clone(r.machineScope) - if err != nil { + // The VM already exists but we have no TaskRef for it: we cloned it + // previously and lost the TaskRef. Complete the post-clone sequence to + // recover — restore VM group membership (if configured) and power the VM + // on, recording the power-on task so subsequent reconciles can track it, + // instead of requeueing forever. This mirrors the completed-clone path + // below so a recovered VM is not left outside its configured VM group. + klog.Infof("%v: VM already exists without a persisted taskRef, recovering", r.machine.GetName()) + return r.addVMGroupAndPowerOn("recovered") + } + + moTask, err := r.session.GetTask(r.Context, r.providerStatus.TaskRef) + if err != nil { + if !isRetrieveMONotFound(r.providerStatus.TaskRef, err) { metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ Name: r.machine.Name, Namespace: r.machine.Namespace, - Reason: "Clone task finished with error", + Reason: "GetTask finished with error", }) - conditionFailed := conditionFailed() - conditionFailed.Message = err.Error() - statusError := setProviderStatus(task, conditionFailed, r.machineScope, nil) - if statusError != nil { - return fmt.Errorf("failed to set provider status: %w", err) - } return err } - return setProviderStatus(task, conditionSuccess(), r.machineScope, nil) - } - - moTask, err := r.session.GetTask(r.Context, r.providerStatus.TaskRef) - if err != nil { - metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ - Name: r.machine.Name, - Namespace: r.machine.Namespace, - Reason: "GetTask finished with error", - }) - return err + // Task history eviction or a session restart can make the clone + // task ref permanently unavailable. Clear it; the moTask == nil + // block below probes for the VM instead of failing forever. + klog.Infof("%v: task %s no longer found, clearing TaskRef", r.machine.GetName(), r.providerStatus.TaskRef) + r.providerStatus.TaskRef = "" } if moTask == nil { + // The clone task is gone from vCenter. If the VM exists the clone + // succeeded; reconcile it into steady state instead of failing on + // the missing task. + if vmRef, err := findVM(r.machineScope); err != nil { + return err + } else if vmRef != (types.ManagedObjectReference{}) { + klog.Infof("%v: clone task gone but VM found, reconciling VM state", r.machine.GetName()) + vm := r.machineScope.newVM(r.machineScope.Context, vmRef) + return r.reconcileMachineWithCloudState(vm, r.providerStatus.TaskRef) + } // Possible eventual consistency problem from vsphere // TODO: change error message here to indicate this might be expected. return fmt.Errorf("unexpected moTask nil") @@ -219,36 +274,7 @@ func (r *Reconciler) create() error { // if clone task finished successfully, power on the vm // The simulator task.Info.DescriptionId is different (VirtualMachine.cloneVM) if strings.Contains(moTask.Info.DescriptionId, cloneVmTaskDescriptionId) { - if r.machineScope.providerSpec.Workspace.VMGroup != "" { - klog.Infof("Adding on cloned machine: %s to vm group: %s", r.machine.Name, r.machineScope.providerSpec.Workspace.VMGroup) - - if err := modifyVMGroup(r.machineScope, false); err != nil { - var taskError task.Error - if errors.As(err, &taskError) { - return fmt.Errorf("could not update VM Group membership: %w", taskError) - } - - return fmt.Errorf("could not update VM Group membership: %w", err) - } - } - - klog.Infof("Powering on cloned machine: %v", r.machine.Name) - task, err := powerOn(r.machineScope) - if err != nil { - metrics.RegisterFailedInstanceCreate(&metrics.MachineLabels{ - Name: r.machine.Name, - Namespace: r.machine.Namespace, - Reason: "PowerOn task finished with error", - }) - conditionFailed := conditionFailed() - conditionFailed.Message = err.Error() - statusError := setProviderStatus(task, conditionFailed, r.machineScope, nil) - if statusError != nil { - return fmt.Errorf("failed to set provider status: %w", err) - } - return err - } - return setProviderStatus(task, conditionSuccess(), r.machineScope, nil) + return r.addVMGroupAndPowerOn("cloned") } // If taskIsFinished then next reconcile should result in update. @@ -272,6 +298,11 @@ func (r *Reconciler) update() error { }) return err } + // Task history eviction or a session restart can make a task + // ref permanently unavailable. Clear it so future resyncs do + // not keep issuing the same GetTask request. + klog.Infof("%v: task %s no longer found, clearing TaskRef", r.machine.GetName(), r.providerStatus.TaskRef) + r.providerStatus.TaskRef = "" } if moTask != nil { if taskIsFinished, err := taskIsFinished(moTask); err != nil { @@ -280,9 +311,19 @@ func (r *Reconciler) update() error { Namespace: r.machine.Namespace, Reason: "Task finished with error", }) + if taskIsFinished { + // A terminally failed task cannot transition to success. Clear + // its ref so retries reconcile the VM instead of polling it forever. + r.providerStatus.TaskRef = "" + } return fmt.Errorf("%v task %v finished with error: %w", moTask.Info.DescriptionId, moTask.Reference().Value, err) } else if !taskIsFinished { return fmt.Errorf("%v task %v has not finished", moTask.Info.DescriptionId, moTask.Reference().Value) + } else { + // A completed task can never transition again. Clear its ref + // so steady-state resyncs skip GetTask entirely. + klog.Infof("%v: task %v has completed, clearing TaskRef", r.machine.GetName(), moTask.Reference().Value) + r.providerStatus.TaskRef = "" } } } @@ -300,13 +341,9 @@ func (r *Reconciler) update() error { return fmt.Errorf("vm not found on update: %w", err) } - vm := &virtualMachine{ - Context: r.machineScope.Context, - Obj: object.NewVirtualMachine(r.machineScope.session.Client.Client, vmRef), - Ref: vmRef, - } + vm := r.machineScope.newVM(r.machineScope.Context, vmRef) - if err := vm.reconcileTags(r.Context, r.session, r.machine, r.providerSpec); err != nil { + if err := vm.reconcileTags(r.Context, r.session.GetCachingTagsManager(), r.machine, r.providerSpec); err != nil { metrics.RegisterFailedInstanceUpdate(&metrics.MachineLabels{ Name: r.machine.Name, Namespace: r.machine.Namespace, @@ -346,11 +383,7 @@ func (r *Reconciler) exists() (bool, error) { // If it is powered off and in "Provisioning" phase, treat machine as non-existed yet and proceed with creation procedure. powerState := types.VirtualMachinePowerState(ptr.Deref(r.machineScope.providerStatus.InstanceState, "")) if powerState == "" || ptr.Deref(r.machine.Status.Phase, "") == machinev1.PhaseProvisioning { - vm := &virtualMachine{ - Context: r.machineScope.Context, - Obj: object.NewVirtualMachine(r.machineScope.session.Client.Client, vmRef), - Ref: vmRef, - } + vm := r.machineScope.newVM(r.machineScope.Context, vmRef) powerState, err = vm.getPowerState() if err != nil { return false, fmt.Errorf("%v: failed checking machine's power state: %w", r.machine.GetName(), err) @@ -423,11 +456,7 @@ func (r *Reconciler) delete() error { return nil } - vm := &virtualMachine{ - Context: r.Context, - Obj: object.NewVirtualMachine(r.machineScope.session.Client.Client, vmRef), - Ref: vmRef, - } + vm := r.machineScope.newVM(r.Context, vmRef) powerState, err := vm.getPowerState() if err != nil { @@ -591,6 +620,14 @@ func (r *Reconciler) reconcileRegionAndZoneLabels(vm *virtualMachine) error { return nil } + // Region/zone come from tags on the VM's ancestry and are immutable + // after provisioning. If the labels are already set, skip the tag + // traversal (HostSystem + Ancestors + N REST tag calls per resync). + if r.machine.Labels[machinecontroller.MachineRegionLabelName] != "" && + r.machine.Labels[machinecontroller.MachineAZLabelName] != "" { + return nil + } + regionLabel := r.vSphereConfig.Labels.Region zoneLabel := r.vSphereConfig.Labels.Zone @@ -613,6 +650,10 @@ func (r *Reconciler) reconcileRegionAndZoneLabels(vm *virtualMachine) error { } func (r *Reconciler) reconcileProviderID(vm *virtualMachine) error { + if r.machine.Spec.ProviderID != nil && *r.machine.Spec.ProviderID != "" { + return nil + } + providerID, err := convertUUIDToProviderID(vm.Obj.UUID(vm.Context)) if err != nil { return err @@ -631,7 +672,8 @@ func convertUUIDToProviderID(UUID string) (string, error) { } func (r *Reconciler) reconcileNetwork(vm *virtualMachine) error { - currentNetworkStatusList, err := vm.getNetworkStatusList(r.session.Client.Client) + // The same property call also seeds the power-state cache. + currentNetworkStatusList, vmName, err := vm.getNetworkAndPowerStatus(r.session.Client.Client) if err != nil { return fmt.Errorf("error getting network status: %v", err) } @@ -653,15 +695,6 @@ func (r *Reconciler) reconcileNetwork(vm *virtualMachine) error { } } - // Using Name() if InventoryPath is empty will return empty name - // see: https://github.com/vmware/govmomi/blob/master/object/common.go#L66-L75 - // Using ObjectName() as it will query from VirtualMachine properties - - vmName, err := vm.Obj.ObjectName(vm.Context) - if err != nil { - return fmt.Errorf("error getting virtual machine name: %v", err) - } - ipAddrs = append(ipAddrs, corev1.NodeAddress{ Type: corev1.NodeInternalDNS, Address: vmName, @@ -856,7 +889,12 @@ func constructKargsFromNetworkConfig(s *machineScope) (string, error) { } func isRetrieveMONotFound(taskRef string, err error) bool { - return err.Error() == fmt.Sprintf("ServerFaultCode: The object 'vim.Task:%v' has already been deleted or has not been completely created", taskRef) + if err == nil { + return false + } + errMessage := err.Error() + return errMessage == fmt.Sprintf("ServerFaultCode: The object 'vim.Task:%v' has already been deleted or has not been completely created", taskRef) || + errMessage == "ServerFaultCode: The object has already been deleted or has not been completely created" } func getHwVersion(ctx context.Context, vm *object.VirtualMachine) (int, error) { @@ -1068,6 +1106,15 @@ func clone(s *machineScope) (string, error) { return taskVal, nil } +// newVM builds a virtualMachine for the given managed object reference. +func (s *machineScope) newVM(ctx context.Context, vmRef types.ManagedObjectReference) *virtualMachine { + return &virtualMachine{ + Context: ctx, + Obj: object.NewVirtualMachine(s.session.Client.Client, vmRef), + Ref: vmRef, + } +} + func modifyVMGroup(s *machineScope, delete bool) error { vmRef, err := findVM(s) if err != nil { @@ -1159,11 +1206,7 @@ func powerOn(s *machineScope) (string, error) { datacenter := s.session.Datacenter if datacenter == nil { // if there is no dataceneter, fallback to old powerOn method via vm object - vm := &virtualMachine{ - Context: s.Context, - Obj: object.NewVirtualMachine(s.session.Client.Client, vmRef), - Ref: vmRef, - } + vm := s.newVM(s.Context, vmRef) return vm.powerOnVM() } @@ -1434,12 +1477,17 @@ func taskIsFinished(task *mo.Task) (bool, error) { } } +// setProviderStatus updates the Machine's ProviderStatus with a task reference, +// a condition, and (optionally) instance metadata. It is called from create(), +// update(), and delete() flows; pass "" for taskRef when there is no active task. func setProviderStatus(taskRef string, condition metav1.Condition, scope *machineScope, vm *virtualMachine) error { klog.Infof("%s: Updating provider status", scope.machine.Name) if vm != nil { - id := vm.Obj.UUID(scope.Context) - scope.providerStatus.InstanceID = &id + if scope.providerStatus.InstanceID == nil || *scope.providerStatus.InstanceID == "" { + id := vm.Obj.UUID(scope.Context) + scope.providerStatus.InstanceID = &id + } // This can return an error if machine is being deleted powerState, err := vm.getPowerState() @@ -1478,6 +1526,12 @@ type virtualMachine struct { context.Context Ref types.ManagedObjectReference Obj *object.VirtualMachine + + // powerState cache: one PowerState call per reconcile pass. + // The struct is built fresh per reconcile (see update()/exists()), + // so the cache never leaks across passes. + ps types.VirtualMachinePowerState + psKnown bool } // getHostSystemAncestors looks up and returns vm's host system ancestors, such as "Cluster" and "Datacenter". @@ -1552,6 +1606,9 @@ func (vm *virtualMachine) getRegionAndZone(tagsMgr *session.CachingTagsManager, } func (vm *virtualMachine) powerOnVM() (string, error) { + // Invalidate the power-state cache so a subsequent getPowerState in + // the same reconcile pass observes the new state, not the stale one. + vm.psKnown = false task, err := vm.Obj.PowerOn(vm.Context) if err != nil { return "", err @@ -1560,6 +1617,9 @@ func (vm *virtualMachine) powerOnVM() (string, error) { } func (vm *virtualMachine) powerOffVM() (string, error) { + // Invalidate the power-state cache so a subsequent getPowerState in + // the same reconcile pass observes the new state, not the stale one. + vm.psKnown = false task, err := vm.Obj.PowerOff(vm.Context) if err != nil { return "", err @@ -1567,128 +1627,111 @@ func (vm *virtualMachine) powerOffVM() (string, error) { return task.Reference().Value, nil } +// resyncPowerState reports whether s is a power state the resync path +// understands (anything else is a fetch bug, not a transient state). +func resyncPowerState(s types.VirtualMachinePowerState) bool { + switch s { + case types.VirtualMachinePowerStatePoweredOn, + types.VirtualMachinePowerStatePoweredOff, + types.VirtualMachinePowerStateSuspended: + return true + default: + return false + } +} + func (vm *virtualMachine) getPowerState() (types.VirtualMachinePowerState, error) { + if vm.psKnown { + return vm.ps, nil + } + powerState, err := vm.Obj.PowerState(vm.Context) if err != nil { return "", err } - switch powerState { - case types.VirtualMachinePowerStatePoweredOn: - return types.VirtualMachinePowerStatePoweredOn, nil - case types.VirtualMachinePowerStatePoweredOff: - return types.VirtualMachinePowerStatePoweredOff, nil - case types.VirtualMachinePowerStateSuspended: - return types.VirtualMachinePowerStateSuspended, nil - default: + if !resyncPowerState(powerState) { return "", fmt.Errorf("unexpected power state %q for vm %v", powerState, vm) } + vm.ps = powerState + vm.psKnown = true + return powerState, nil } // reconcileTags ensures that the required tags are present on the virtual machine, eg the Cluster ID // that is used by the installer on cluster deletion to ensure ther are no leaked resources. -func (vm *virtualMachine) reconcileTags(ctx context.Context, sessionInstance *session.Session, machine *machinev1.Machine, providerSpec *machinev1.VSphereMachineProviderSpec) error { - // Use cached tag manager to avoid creating new REST sessions. - // This eliminates excessive vCenter login/logout cycles. - tagManager := sessionInstance.GetCachingTagsManager() - klog.Infof("%v: Reconciling attached tags", machine.GetName()) - +// The attached-tag list is fetched once per reconcile via the batch +// list-attached-on-objects endpoint (the per-object list-attached action is +// documented as much slower at scale, per the Broadcom vCenter tagging +// performance white paper); every required tag is checked against it in +// memory, and any missing tags are attached in one attach-multiple call. +func (vm *virtualMachine) reconcileTags(ctx context.Context, tagManager *session.CachingTagsManager, machine *machinev1.Machine, providerSpec *machinev1.VSphereMachineProviderSpec) error { clusterID := machine.Labels[machinev1.MachineClusterIDLabel] - tagIDs := []string{clusterID} - tagIDs = append(tagIDs, providerSpec.TagIDs...) + tagIDs := append([]string{clusterID}, providerSpec.TagIDs...) klog.Infof("%v: Reconciling %s tags to vm", machine.GetName(), tagIDs) - for _, tagID := range tagIDs { - attached, err := vm.checkAttachedTag(ctx, tagID, tagManager) - if err != nil { - return err - } - - if !attached { - klog.Infof("%v: Attaching %s tag to vm", machine.GetName(), tagID) - // the tag should already be created by installer or the administrator - if err := tagManager.AttachTag(ctx, tagID, vm.Ref); err != nil { - return err - } - } - } - return nil -} - -// checkAttachedTag returns true if tag is already attached to a vm or tag doesn't exist -func (vm *virtualMachine) checkAttachedTag(ctx context.Context, tagName string, m *session.CachingTagsManager) (bool, error) { - // cluster ID tag doesn't exists in UPI, we should skip tag attachment if it's not found - foundTag, err := vm.foundTag(ctx, tagName, m) - if err != nil { - return false, err - } - if !foundTag { - return true, nil - } + var toAttach []string - tags, err := m.GetAttachedTags(ctx, vm.Ref) + objs, err := tagManager.ListAttachedTagsOnObjects(ctx, []mo.Reference{vm.Ref}) if err != nil { - return false, err + return fmt.Errorf("failed to list attached tags for vm %v: %w", vm.Ref, err) } - - for _, tag := range tags { - if session.IsName(tagName) { - if tag.Name == tagName { - return true, nil - } - } else { - if tag.ID == tagName { - return true, nil - } + // The list may be empty (unrecognized reference) or contain more + // entries than requested; iterate defensively instead of assuming + // objs[0] exists. + attachedIDs := make(map[string]bool) + for _, obj := range objs { + for _, id := range obj.TagIDs { + attachedIDs[id] = true } - } - return false, nil -} - -// tagToCategoryName converts the tag name to the category name based upon the format set up by the installer. -// Note this is only valid in IPI clusters as typically a UPI cluster won't have the cluster ID tag, in which case the -// controller skips tag creation. -// Ref: https://github.com/openshift/installer/blob/f912534f12491721e3874e2bf64f7fa8d44aa7f5/data/data/vsphere/pre-bootstrap/main.tf#L57 -// Ref: https://github.com/openshift/installer/blob/f912534f12491721e3874e2bf64f7fa8d44aa7f5/pkg/destroy/vsphere/vsphere.go#L231 -func tagToCategoryName(tagName string) string { - return fmt.Sprintf("openshift-%s", tagName) -} - -func (vm *virtualMachine) foundTag(ctx context.Context, tagName string, m *session.CachingTagsManager) (bool, error) { - var tags []string - var err error - - if session.IsName(tagName) { - tags, err = m.ListTagsForCategory(ctx, tagToCategoryName(tagName)) - if err != nil { - if isNotFoundErr(err) { - return false, nil - } - return false, err + for _, tagID := range tagIDs { + if tagID == "" { + continue } - } else { - tags = []string{tagName} - } - klog.V(4).Infof("validating the presence of tags: %+v", tags) - for _, id := range tags { - tag, err := m.GetTag(ctx, id) - if err != nil { - return false, err + if attachedIDs[tagID] { + continue } - if session.IsName(tagName) { - if tag.Name == tagName { - return true, nil + + if session.IsName(tagID) { + // Resolve the name to an ID. A missing tag is not an error: + // clusters may run without the cluster-ID tag, and attaching + // would fail anyway. + tag, err := tagManager.GetTag(ctx, tagID) + if err != nil { + if isNotFoundErr(err) { + klog.V(3).Infof("%v: tag %q not found in vCenter, skipping attach", machine.GetName(), tagID) + continue + } + return err } - } else { - if tag.ID == tagName { - return true, nil + if attachedIDs[tag.ID] { + continue } + // Mark the queued ID so duplicate entries (repeated cluster-ID + // or providerSpec.TagIDs) are not attached twice. + attachedIDs[tag.ID] = true + toAttach = append(toAttach, tag.ID) + } else if _, err := tagManager.GetTag(ctx, tagID); err != nil { + // Unknown tag ID: fail loudly, matching previous behavior. + return err + } else { + attachedIDs[tagID] = true + toAttach = append(toAttach, tagID) } } - return false, nil + // One batched attach for everything missing (white paper: + // attach-multiple-tags-to-object has flat latency; per-tag attach() + // scales linearly with tag count). + if len(toAttach) > 0 { + klog.Infof("%v: Attaching %d tag(s) to vm", machine.GetName(), len(toAttach)) + if err := tagManager.AttachMultipleTagsToObject(ctx, toAttach, vm.Ref); err != nil { + return err + } + } + return nil } type NetworkStatus struct { @@ -1706,20 +1749,34 @@ type NetworkStatus struct { NetworkName string } -func (vm *virtualMachine) getNetworkStatusList(client *vim25.Client) ([]NetworkStatus, error) { +// getNetworkAndPowerStatus fetches the VM's network status, name, and power state +// in a single property call, seeding the power-state cache for the reconcile pass. +func (vm *virtualMachine) getNetworkAndPowerStatus(client *vim25.Client) ([]NetworkStatus, string, error) { var obj mo.VirtualMachine var pc = property.DefaultCollector(client) var props = []string{ "config.hardware.device", "guest.net", + "name", + "runtime.powerState", } if err := pc.RetrieveOne(vm.Context, vm.Ref, props, &obj); err != nil { - return nil, fmt.Errorf("unable to fetch props %v for vm %v: %w", props, vm.Ref, err) + return nil, "", fmt.Errorf("unable to fetch props %v for vm %v: %w", props, vm.Ref, err) + } + + // Seed the power-state cache so a subsequent getPowerState in the same + // reconcile pass does not issue a second property call. Invalid or + // missing states are left uncached so getPowerState() still re-fetches + // and errors as before. + if resyncPowerState(obj.Runtime.PowerState) { + vm.ps = obj.Runtime.PowerState + vm.psKnown = true } + klog.V(3).Infof("Getting network status: object reference: %v", obj.Reference().Value) if obj.Config == nil { - return nil, errors.New("config.hardware.device is nil") + return nil, "", errors.New("config.hardware.device is nil") } var networkStatusList []NetworkStatus @@ -1746,7 +1803,7 @@ func (vm *virtualMachine) getNetworkStatusList(client *vim25.Client) ([]NetworkS } } - return networkStatusList, nil + return networkStatusList, obj.Name, nil } type attachedDisk struct { diff --git a/pkg/controller/vsphere/reconciler_test.go b/pkg/controller/vsphere/reconciler_test.go index 3fc67d9936..89696bbad1 100644 --- a/pkg/controller/vsphere/reconciler_test.go +++ b/pkg/controller/vsphere/reconciler_test.go @@ -17,22 +17,31 @@ import ( "context" "crypto/tls" "encoding/base64" + "encoding/json" "errors" "fmt" + "io" "net" + "net/http" + "net/http/httptest" + "net/url" "os" "path" "reflect" + "sort" "strings" + "sync" "testing" . "github.com/onsi/gomega" "github.com/vmware/govmomi/object" "github.com/vmware/govmomi/simulator" + "github.com/vmware/govmomi/vapi/rest" "github.com/vmware/govmomi/vapi/tags" "github.com/vmware/govmomi/vim25" "github.com/vmware/govmomi/vim25/mo" + "github.com/vmware/govmomi/vim25/soap" "github.com/vmware/govmomi/vim25/types" corev1 "k8s.io/api/core/v1" @@ -1466,7 +1475,41 @@ func createDataDiskDefinitions(numOfDataDisks int) []machinev1.VSphereDisk { return disks } -func TestGetNetworkStatusList(t *testing.T) { +func TestSetProviderStatusPreservesExistingInstanceID(t *testing.T) { + model, sess, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + managedObj := model.Map().Any("VirtualMachine").(*simulator.VirtualMachine) + vmRef := managedObj.Reference() + vm := &virtualMachine{ + Context: context.Background(), + Obj: object.NewVirtualMachine(sess.Client.Client, vmRef), + Ref: vmRef, + } + + const existingInstanceID = "existing-instance-id" + scope := &machineScope{ + Context: context.Background(), + machine: &machinev1.Machine{ObjectMeta: metav1.ObjectMeta{Name: "test-machine"}}, + providerStatus: &machinev1.VSphereMachineProviderStatus{ + InstanceID: func() *string { v := existingInstanceID; return &v }(), + }, + } + + if err := setProviderStatus("", conditionSuccess(), scope, vm); err != nil { + t.Fatal(err) + } + got := "" + if scope.providerStatus.InstanceID != nil { + got = *scope.providerStatus.InstanceID + } + if got != existingInstanceID { + t.Errorf("InstanceID changed from %q to %q", existingInstanceID, got) + } +} + +func TestGetNetworkAndPowerStatus(t *testing.T) { model, session, server := initSimulator(t) defer model.Remove() defer server.Close() @@ -1492,7 +1535,7 @@ func TestGetNetworkStatusList(t *testing.T) { } // validations - networkStatusList, err := vm.getNetworkStatusList(session.Client.Client) + networkStatusList, _, err := vm.getNetworkAndPowerStatus(session.Client.Client) if err != nil { t.Fatal(err) } @@ -1512,6 +1555,46 @@ func TestGetNetworkStatusList(t *testing.T) { // TODO: add more cases by adding network devices to the NewVirtualMachine() object } +func TestGetNetworkAndPowerStatusCachesPowerState(t *testing.T) { + model, session, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + // The simulator powers VMs on by default. + simVMObject := object.NewVirtualMachine(session.Client.Client, model.Map().Any("VirtualMachine").Reference()) + + // Fresh virtualMachine: nothing cached yet. + freshVM := &virtualMachine{ + Context: context.TODO(), + Obj: simVMObject, + Ref: simVMObject.Reference(), + } + _, _, err := freshVM.getNetworkAndPowerStatus(session.Client.Client) + if err != nil { + t.Fatal(err) + } + if !freshVM.psKnown { + t.Fatal("expected getNetworkAndPowerStatus to seed the power-state cache (psKnown)") + } + + // A second power-state read must come from the cache: flipping the + // simulated VM's power must not be observed by getPowerState. + task, err := simVMObject.PowerOff(context.TODO()) + if err != nil { + t.Fatal(err) + } + if err := task.Wait(context.TODO()); err != nil { + t.Fatal(err) + } + state, err := freshVM.getPowerState() + if err != nil { + t.Fatal(err) + } + if state != freshVM.ps { + t.Errorf("expected cached power state %v, got %v", freshVM.ps, state) + } +} + func TestReconcileNetwork(t *testing.T) { model, session, server := initSimulator(t) defer model.Remove() @@ -1584,6 +1667,18 @@ func TestReconcileTags(t *testing.T) { Ref: managedObjRef, } + // attachedTagIDs returns the tag URNs currently attached to the sim VM. + attachedTagIDs := func() ([]string, error) { + attached, err := sessionObj.GetCachingTagsManager().ListAttachedTagsOnObjects(context.TODO(), []mo.Reference{managedObjRef}) + if err != nil { + return nil, err + } + ids := make([]string, 0, len(attached[0].TagIDs)) + ids = append(ids, attached[0].TagIDs...) + sort.Strings(ids) + return ids, nil + } + testCases := []struct { name string expectedError bool @@ -1596,6 +1691,19 @@ func TestReconcileTags(t *testing.T) { expectedError: false, tagName: "FOOOOOOOOO", }, + { + // steady-state machine: tag exists in vCenter and is already + // attached, reconcileTags must attach nothing and succeed. + name: "Skip attach when the tag is already attached", + expectedError: false, + tagName: "ALREADYATTACHED", + testCondition: func(tagName string) (string, error) { + if _, err := createTagAndCategory(sessionObj, tagToCategoryName(tagName), tagName); err != nil { + return "", err + } + return "", sessionObj.GetCachingTagsManager().AttachTag(context.TODO(), tagName, managedObjRef) + }, + }, { name: "Successfully attach a tag", expectedError: false, @@ -1648,7 +1756,12 @@ func TestReconcileTags(t *testing.T) { } } - err := vm.reconcileTags(context.TODO(), sessionObj, &machinev1.Machine{ + before, err := attachedTagIDs() + if err != nil { + t.Fatal(err) + } + + err = vm.reconcileTags(context.TODO(), sessionObj.GetCachingTagsManager(), &machinev1.Machine{ ObjectMeta: metav1.ObjectMeta{ Name: "machine", Labels: map[string]string{machinev1.MachineClusterIDLabel: tc.tagName}, @@ -1664,42 +1777,36 @@ func TestReconcileTags(t *testing.T) { t.Fatalf("Not expected error %v", err) } - if tc.attachTag { - tagMgr := sessionObj.GetCachingTagsManager() - - tags, err := tagMgr.GetAttachedTags(context.TODO(), managedObjRef) - if err != nil { - t.Fatal(err) - } - - if len(tags) == 0 { - t.Fatalf("Expected tags to be found") - } + after, err := attachedTagIDs() + if err != nil { + t.Fatal(err) + } + if tc.attachTag { expectedTags := []string{tc.tagName} if len(providerSpec.TagIDs) > 0 { expectedTags = append(expectedTags, providerSpec.TagIDs...) } + tagMgr := sessionObj.GetCachingTagsManager() for _, expectedTag := range expectedTags { + resolved, err := tagMgr.GetTag(context.TODO(), expectedTag) + if err != nil { + t.Fatalf("Expected tag %s to exist: %v", expectedTag, err) + } gotTag := false - for _, attachedTag := range tags { - if session.IsName(expectedTag) { - if attachedTag.Name == expectedTag { - gotTag = true - break - } - } else { - if attachedTag.ID == expectedTag { - gotTag = true - break - } + for _, attachedID := range after { + if attachedID == resolved.ID { + gotTag = true + break } } if !gotTag { t.Fatalf("Expected tag %s to be found", expectedTag) } } + } else if !reflect.DeepEqual(before, after) { + t.Fatalf("Expected attached tags to be unchanged: before %v, after %v", before, after) } } @@ -1707,10 +1814,12 @@ func TestReconcileTags(t *testing.T) { } } -func TestCheckAttachedTag(t *testing.T) { +// TestReconcileTagsListFailure verifies that a failure of the batched +// list-attached-tags-on-objects call surfaces as an error from reconcileTags. +func TestReconcileTagsListFailure(t *testing.T) { model, sessionObj, server := initSimulator(t) defer model.Remove() - defer server.Close() + server.Close() managedObj := model.Map().Any("VirtualMachine").(*simulator.VirtualMachine) managedObjRef := object.NewVirtualMachine(sessionObj.Client.Client, managedObj.Reference()).Reference() @@ -1721,83 +1830,101 @@ func TestCheckAttachedTag(t *testing.T) { Ref: managedObjRef, } - tagName := "CLUSTERID" - nonAttachedTagName := "nonAttachedTag" - - tagsMgr := sessionObj.TagManager - - id, err := tagsMgr.CreateCategory(context.TODO(), &tags.Category{ - AssociableTypes: []string{"VirtualMachine"}, - Cardinality: "SINGLE", - Name: tagToCategoryName(tagName), - }) - if err != nil { - t.Fatal(err) - } + err := vm.reconcileTags(context.TODO(), sessionObj.GetCachingTagsManager(), &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "machine", + Labels: map[string]string{machinev1.MachineClusterIDLabel: "CLUSTERID"}, + }, + }, &machinev1.VSphereMachineProviderSpec{}) - _, err = tagsMgr.CreateTag(context.TODO(), &tags.Tag{ - CategoryID: id, - Name: tagName, - }) - if err != nil { - t.Fatal(err) + if err == nil { + t.Fatal("Expected error when listing attached tags fails") } +} - if err := tagsMgr.AttachTag(context.TODO(), tagName, vm.Ref); err != nil { - t.Fatal(err) - } +// noopCookieJar satisfies http.CookieJar without x/net/cookiejar (not +// vendored); NewServiceClient dereferences the jar unconditionally. +type noopCookieJar struct{} + +func (noopCookieJar) SetCookies(*url.URL, []*http.Cookie) {} +func (noopCookieJar) Cookies(*url.URL) []*http.Cookie { return nil } + +// TestReconcileTagsEmptyAttachedList verifies that an empty +// list-attached-tags-on-objects response is treated as "nothing attached" +// (no panic, missing tags are still attached), and that duplicate tag IDs +// (cluster-ID plus repeated providerSpec.TagIDs) are attached only once. +func TestReconcileTagsEmptyAttachedList(t *testing.T) { + var mu sync.Mutex + var attachBodies []string + + // Responses are wrapped in {"value": ...} because the client requests the + // /rest endpoint (raw bodies are only expected under /api). + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.Contains(r.URL.RawQuery, "action=list-attached-tags-on-objects"): + // Empty response: no attached-tags entries at all. + _, _ = w.Write([]byte(`{"value":[]}`)) + case strings.Contains(r.URL.RawQuery, "action=attach-multiple-tags-to-object"): + body, _ := io.ReadAll(r.Body) + mu.Lock() + attachBodies = append(attachBodies, string(body)) + mu.Unlock() + _, _ = w.Write([]byte(`{"value":{"success":true}}`)) + case r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/tag/"): + _, _ = w.Write([]byte(`{"value":{"id":"urn:vmomi:InventoryTag:42","name":"CLUSTERID"}}`)) + case strings.HasSuffix(r.URL.Path, "/tagging/tag"): + // ListTags: array of tag IDs. + _, _ = w.Write([]byte(`{"value":["urn:vmomi:InventoryTag:42"]}`)) + default: + http.Error(w, "Not Found", http.StatusNotFound) + } + })) + defer server.Close() - nonAttachedCategoryId, err := tagsMgr.CreateCategory(context.TODO(), &tags.Category{ - AssociableTypes: []string{"VirtualMachine"}, - Cardinality: "SINGLE", - Name: tagToCategoryName(nonAttachedTagName), - }) + u, err := url.Parse(server.URL) if err != nil { t.Fatal(err) } + sc := soap.NewClient(u, false) + sc.Client = *server.Client() + sc.Client.Jar = noopCookieJar{} + tagMgr := &session.CachingTagsManager{ + Manager: tags.NewManager(rest.NewClient(&vim25.Client{Client: sc, RoundTripper: sc})), + } - _, err = tagsMgr.CreateTag(context.TODO(), &tags.Tag{ - CategoryID: nonAttachedCategoryId, - Name: nonAttachedTagName, - }) - if err != nil { - t.Fatal(err) + vm := &virtualMachine{ + Context: context.TODO(), + Ref: types.ManagedObjectReference{Type: "VirtualMachine", Value: "vm-100"}, } - testCases := []struct { - name string - findTag bool - tagName string - }{ - { - name: "Successfully find a tag", - findTag: true, - tagName: tagName, - }, - { - name: "Return true if a tag doesn't exist", - tagName: "non existent tag", - findTag: true, - }, - { - name: "Fail to find a tag", - tagName: nonAttachedTagName, + machine := &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "machine", + Labels: map[string]string{machinev1.MachineClusterIDLabel: "CLUSTERID"}, }, } + // Same URN as the cluster-ID tag, repeated twice. + spec := &machinev1.VSphereMachineProviderSpec{ + TagIDs: []string{"urn:vmomi:InventoryTag:42", "urn:vmomi:InventoryTag:42"}, + } - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - c := sessionObj.GetCachingTagsManager() - - attached, err := vm.checkAttachedTag(context.TODO(), tc.tagName, c) - if err != nil { - t.Fatalf("Not expected error %v", err) - } + if err := vm.reconcileTags(context.TODO(), tagMgr, machine, spec); err != nil { + t.Fatalf("Not expected error %v", err) + } - if attached != tc.findTag { - t.Fatalf("Failed to find attached tag: got %v, expected %v", attached, tc.findTag) - } - }) + mu.Lock() + defer mu.Unlock() + if len(attachBodies) != 1 { + t.Fatalf("Expected 1 attach call, got %d", len(attachBodies)) + } + var req struct { + TagIDs []string `json:"tag_ids"` + } + if err := json.Unmarshal([]byte(attachBodies[0]), &req); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(req.TagIDs, []string{"urn:vmomi:InventoryTag:42"}) { + t.Fatalf("Expected a single deduped tag ID, got %v", req.TagIDs) } } @@ -2945,6 +3072,184 @@ func waitForTaskToComplete(session *session.Session, reconciler *Reconciler) err return nil } +// TestCreateRecoversLostTaskRef verifies that create() recovers a VM that was +// cloned but whose TaskRef was never persisted (for example, because the status +// patch was denied by an admission webhook during install). Instead of cloning +// a second VM, create() must find the existing VM and power it on. This is the +// provider-side defense for OCPBUGS-100316. +func TestCreateRecoversLostTaskRef(t *testing.T) { + g := NewWithT(t) + + // Autostart=false leaves the simulator VMs powered off, mimicking a VM that + // was cloned but never powered on. + poweredOff := func(m *simulator.Model) { m.Autostart = false } + model, server := initSimulatorCustom(t, poweredOff) + session := getSimulatorSession(t, server) + defer model.Remove() + defer server.Close() + + host, _, err := net.SplitHostPort(server.URL.Host) + g.Expect(err).ToNot(HaveOccurred()) + + vms := model.Map().All("VirtualMachine") + g.Expect(vms).ToNot(BeEmpty()) + existingVM := vms[0].(*simulator.VirtualMachine) + vmCountBefore := len(vms) + + provisioning := string(machinev1.PhaseProvisioning) + machineObj := &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: existingVM.Name, + Namespace: "test", + Labels: map[string]string{machinev1.MachineClusterIDLabel: "CLUSTERID"}, + // The machine UID matches the VM instance UUID so findVM adopts the + // already-cloned VM instead of cloning a new one. + UID: apimachinerytypes.UID(existingVM.Config.InstanceUuid), + }, + Status: machinev1.MachineStatus{Phase: &provisioning}, + } + + machineScope := &machineScope{ + Context: context.TODO(), + machine: machineObj, + machineToBePatched: runtimeclient.MergeFrom(machineObj.DeepCopy()), + providerSpec: &machinev1.VSphereMachineProviderSpec{ + Template: existingVM.Name, + Workspace: &machinev1.Workspace{Server: host}, + }, + session: session, + // No TaskRef and no InstanceState: the reference to the clone task was lost. + providerStatus: &machinev1.VSphereMachineProviderStatus{}, + client: fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects(machineObj).WithStatusSubresource(machineObj).Build(), + } + + reconciler := newReconciler(machineScope) + + g.Expect(reconciler.create()).To(Succeed()) + + // A recovery task must have been recorded rather than requeueing forever. + g.Expect(reconciler.providerStatus.TaskRef).ToNot(BeEmpty(), "expected a recovery power-on task to be recorded") + + // No new VM must have been cloned. + g.Expect(model.Map().All("VirtualMachine")).To(HaveLen(vmCountBefore), "create() must not clone a duplicate VM when one already exists") + + // The recovery task must be a power-on (not a clone) and must succeed. + g.Expect(waitForTaskToComplete(session, reconciler)).To(Succeed()) + moTask, err := session.GetTask(context.TODO(), reconciler.providerStatus.TaskRef) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(moTask).ToNot(BeNil()) + g.Expect(moTask.Info.DescriptionId).ToNot(ContainSubstring(cloneVmTaskDescriptionId)) + + // The existing VM must now be powered on. + vmObj := &virtualMachine{ + Context: context.TODO(), + Obj: object.NewVirtualMachine(session.Client.Client, existingVM.Reference()), + Ref: existingVM.Reference(), + } + powerState, err := vmObj.getPowerState() + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(powerState).To(Equal(types.VirtualMachinePowerStatePoweredOn)) +} + +// TestCreateRecoveryRestoresVMGroup verifies that when create() recovers a VM +// whose TaskRef was lost, it restores the configured VM-group membership before +// powering the VM on, matching the normal completed-clone path. Otherwise a +// recovered VM would be left outside its DRS host-affinity group +// (OCPBUGS-100316). +func TestCreateRecoveryRestoresVMGroup(t *testing.T) { + g := NewWithT(t) + + poweredOff := func(m *simulator.Model) { m.Autostart = false } + model, server := initSimulatorCustom(t, poweredOff) + session := getSimulatorSession(t, server) + defer model.Remove() + defer server.Close() + + host, _, err := net.SplitHostPort(server.URL.Host) + g.Expect(err).ToNot(HaveOccurred()) + + ctx := context.Background() + ccr, err := session.Finder.ClusterComputeResourceOrDefault(ctx, "/...") + g.Expect(err).ToNot(HaveOccurred()) + resourcePool := path.Join(ccr.InventoryPath, "Resources") + + vmGroup := "recovery-vm-group" + g.Expect(createVMGroup(ctx, session, ccr.Name(), vmGroup)).To(Succeed()) + + // Pick a powered-off VM that belongs to the cluster, standing in for a VM we + // cloned but whose TaskRef we lost. + var existingVM *simulator.VirtualMachine + for _, obj := range model.Map().All("VirtualMachine") { + candidate := obj.(*simulator.VirtualMachine) + if candidate.Runtime.PowerState == types.VirtualMachinePowerStatePoweredOff && candidate.ResourcePool != nil { + existingVM = candidate + break + } + } + g.Expect(existingVM).ToNot(BeNil()) + vmCountBefore := len(model.Map().All("VirtualMachine")) + + gates, err := testutils.NewDefaultMutableFeatureGate() + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(gates.SetFromMap(map[string]bool{string(features.FeatureGateVSphereHostVMGroupZonal): true})).To(Succeed()) + + provisioning := string(machinev1.PhaseProvisioning) + machineObj := &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: existingVM.Name, + Namespace: "test", + Labels: map[string]string{machinev1.MachineClusterIDLabel: "CLUSTERID"}, + UID: apimachinerytypes.UID(existingVM.Config.InstanceUuid), + }, + Status: machinev1.MachineStatus{Phase: &provisioning}, + } + + machineScope := &machineScope{ + Context: ctx, + machine: machineObj, + machineToBePatched: runtimeclient.MergeFrom(machineObj.DeepCopy()), + providerSpec: &machinev1.VSphereMachineProviderSpec{ + Template: existingVM.Name, + Workspace: &machinev1.Workspace{ + Server: host, + VMGroup: vmGroup, + ResourcePool: resourcePool, + }, + }, + session: session, + providerStatus: &machinev1.VSphereMachineProviderStatus{}, + featureGates: gates, + client: fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects(machineObj).WithStatusSubresource(machineObj).Build(), + } + + reconciler := newReconciler(machineScope) + + g.Expect(reconciler.create()).To(Succeed()) + + // No duplicate VM was cloned. + g.Expect(model.Map().All("VirtualMachine")).To(HaveLen(vmCountBefore)) + + // The recovered VM must have been added to the configured VM group before + // power-on. + clusterConfig, err := ccr.Configuration(ctx) + g.Expect(err).ToNot(HaveOccurred()) + memberFound := false + for _, grp := range clusterConfig.Group { + if vmg, ok := grp.(*types.ClusterVmGroup); ok && vmg.Name == vmGroup { + for _, ref := range vmg.Vm { + if ref.Value == existingVM.Reference().Value { + memberFound = true + } + } + } + } + g.Expect(memberFound).To(BeTrue(), "recovered VM must be a member of its configured VM group") + + // A power-on task must have been recorded for the recovered VM. + g.Expect(reconciler.providerStatus.TaskRef).ToNot(BeEmpty()) + g.Expect(waitForTaskToComplete(session, reconciler)).To(Succeed()) +} + func TestUpdate(t *testing.T) { model, session, server := initSimulator(t) defer model.Remove() @@ -3345,6 +3650,15 @@ func TestReconcileMachineWithCloudState(t *testing.T) { } } +// tagToCategoryName converts the tag name to the category name based upon the format set up by the installer. +// Note this is only valid in IPI clusters as typically a UPI cluster won't have the cluster ID tag, in which case the +// controller skips tag creation. +// Ref: https://github.com/openshift/installer/blob/f912534f12491721e3874e2bf64f7fa8d44aa7f5/data/data/vsphere/pre-bootstrap/main.tf#L57 +// Ref: https://github.com/openshift/installer/blob/f912534f12491721e3874e2bf64f7fa8d44aa7f5/pkg/destroy/vsphere/vsphere.go#L231 +func tagToCategoryName(tagName string) string { + return fmt.Sprintf("openshift-%s", tagName) +} + func createTagAndCategory(session *session.Session, categoryName, tagName string) (string, error) { tagsMgr := session.TagManager @@ -3585,3 +3899,254 @@ func TestReconcilePowerStateAnnontation(t *testing.T) { } // See https://github.com/vmware/govmomi/blob/master/simulator/example_extend_test.go#L33:6 for extending behaviour example + +func TestUpdateClearsFinishedTaskRef(t *testing.T) { + model, sess, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + host, port, err := net.SplitHostPort(server.URL.Host) + if err != nil { + t.Fatal(err) + } + password, _ := server.URL.User.Password() + namespace := "test" + credentialsSecretName := "test" + credentialsSecret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{ + Name: credentialsSecretName, + Namespace: namespace, + }, + Data: map[string][]byte{ + fmt.Sprintf("%s.username", host): []byte(server.URL.User.Username()), + fmt.Sprintf("%s.password", host): []byte(password), + }, + } + configMap := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: OpenshiftConfigManagedConfigMap, + Namespace: openshiftConfigNamespaceForTest, + }, + Data: map[string]string{ + OpenshiftConfigManagedCloudConfigKey: fmt.Sprintf(testConfigFmt, port, credentialsSecretName, namespace), + }, + } + if _, err := createTagAndCategory(sess, tagToCategoryName("CLUSTERID"), "CLUSTERID"); err != nil { + t.Fatalf("cannot create tag and category: %v", err) + } + + vm := model.Map().Any("VirtualMachine").(*simulator.VirtualMachine) + vm.Config.InstanceUuid = "a5764857-ae35-34dc-8f25-a9c9e73aa898" + vmObj := object.NewVirtualMachine(sess.Client.Client, vm.Reference()) + powerOffTask, err := vmObj.PowerOff(context.Background()) + if err != nil { + t.Fatal(err) + } + if err := object.NewTask(sess.Client.Client, powerOffTask.Reference()).Wait(context.Background()); err != nil { + t.Fatal(err) + } + task, err := vmObj.PowerOn(context.Background()) + if err != nil { + t.Fatal(err) + } + if err := object.NewTask(sess.Client.Client, task.Reference()).Wait(context.Background()); err != nil { + t.Fatal(err) + } + + failedTask := simulator.CreateTask(vm, "failedTask", func(*simulator.Task) (types.AnyType, types.BaseMethodFault) { + return nil, &types.InvalidArgument{} + }) + failedTaskRef := failedTask.Run(model.Service.Context) + failedTask.Wait() + + rawProviderSpec, err := RawExtensionFromProviderSpec(&machinev1.VSphereMachineProviderSpec{ + Workspace: &machinev1.Workspace{Server: host}, + CredentialsSecret: &corev1.LocalObjectReference{ + Name: credentialsSecretName, + }, + Template: vm.Name, + Network: machinev1.NetworkSpec{ + Devices: []machinev1.NetworkDeviceSpec{{NetworkName: "test"}}, + }, + }) + if err != nil { + t.Fatal(err) + } + + for _, tc := range []struct { + name string + taskRef string + expectError bool + }{ + {name: "finished task", taskRef: task.Reference().Value}, + {name: "stale missing task", taskRef: "task-99999"}, + {name: "failed task", taskRef: failedTaskRef.Value, expectError: true}, + } { + t.Run(tc.name, func(t *testing.T) { + machineObj := &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-" + strings.ReplaceAll(tc.name, " ", "-"), + Namespace: namespace, + Labels: map[string]string{ + machinev1.MachineClusterIDLabel: "CLUSTERID", + }, + UID: apimachinerytypes.UID(vm.Config.InstanceUuid), + }, + Spec: machinev1.MachineSpec{ + ProviderSpec: machinev1.ProviderSpec{Value: rawProviderSpec}, + }, + } + client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects( + credentialsSecret, configMap).Build() + scope, err := newMachineScope(machineScopeParams{ + client: client, + Context: context.Background(), + machine: machineObj, + apiReader: client, + openshiftConfigNameSpace: openshiftConfigNamespaceForTest, + }) + if err != nil { + t.Fatal(err) + } + scope.providerStatus.TaskRef = tc.taskRef + + err = newReconciler(scope).update() + if tc.expectError { + if err == nil { + t.Fatal("update() succeeded for failed task") + } + } else if err != nil { + t.Fatalf("update() error: %v", err) + } + if scope.providerStatus.TaskRef != "" { + t.Errorf("TaskRef not cleared after finished/stale task, got %q", scope.providerStatus.TaskRef) + } + }) + } +} + +func TestReconcileRegionAndZoneLabelsSkipsWhenSet(t *testing.T) { + machine := &machinev1.Machine{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test", + Labels: map[string]string{ + machinecontroller.MachineRegionLabelName: "east", + machinecontroller.MachineAZLabelName: "a", + }, + }, + } + r := &Reconciler{ + machineScope: &machineScope{ + machine: machine, + providerStatus: &machinev1.VSphereMachineProviderStatus{}, + vSphereConfig: &vsphere.Config{ + Labels: vsphere.Labels{Region: "region", Zone: "zone"}, + }, + }, + } + // No session: if the function touches the session it panics; the + // guard must return before any vCenter call. + if err := r.reconcileRegionAndZoneLabels(nil); err != nil { + t.Fatalf("expected nil, got %v", err) + } + if machine.Labels[machinecontroller.MachineRegionLabelName] != "east" || + machine.Labels[machinecontroller.MachineAZLabelName] != "a" { + t.Errorf("labels were modified: %v", machine.Labels) + } +} + +func TestReconcileProviderIDSkipsWhenSet(t *testing.T) { + pid := "vsphere://564d...c7f6" + machine := &machinev1.Machine{} + machine.Spec.ProviderID = &pid + r := &Reconciler{ + machineScope: &machineScope{machine: machine, providerStatus: &machinev1.VSphereMachineProviderStatus{}}, + } + // vm == nil: if the function calls into the VM client it panics; + // the guard must return first. + if err := r.reconcileProviderID(nil); err != nil { + t.Fatalf("expected nil, got %v", err) + } +} + +func TestGetPowerStateCachedWithinPass(t *testing.T) { + model, sess, server := initSimulator(t) + defer model.Remove() + defer server.Close() + ctx := context.Background() + + simVM := model.Map().Any("VirtualMachine").(*simulator.VirtualMachine) + vmObj := object.NewVirtualMachine(sess.Client.Client, simVM.Reference()) + vm := &virtualMachine{Context: ctx, Obj: vmObj, Ref: simVM.Reference()} + + first, err := vm.getPowerState() + if err != nil { + t.Fatal(err) + } + + // Mutate the simulator's power state in-process so a non-caching + // implementation would observe a different value on the next call. + simVM.Runtime.PowerState = types.VirtualMachinePowerStateSuspended + + // The second call must still return the cached value, proving the + // cache is used instead of re-querying vCenter within the pass. + second, err := vm.getPowerState() + if err != nil { + t.Fatal(err) + } + if second != first { + t.Errorf("cached power state = %s, want %s (simulator state changed to suspended)", second, first) + } +} + +func TestIsRetrieveMONotFound(t *testing.T) { + taskRef := "task-12345" + expectedErr := fmt.Sprintf("ServerFaultCode: The object 'vim.Task:%v' has already been deleted or has not been completely created", taskRef) + + tests := []struct { + name string + taskRef string + err error + want bool + }{ + { + name: "nil error returns false", + taskRef: taskRef, + err: nil, + want: false, + }, + { + name: "RetrieveMO NotFound with full message returns true", + taskRef: taskRef, + err: errors.New(expectedErr), + want: true, + }, + { + name: "RetrieveMO NotFound with generic message returns true", + taskRef: taskRef, + err: errors.New("ServerFaultCode: The object has already been deleted or has not been completely created"), + want: true, + }, + { + name: "other error returns false", + taskRef: taskRef, + err: errors.New("some other error"), + want: false, + }, + { + name: "different task ref in message returns false", + taskRef: taskRef, + err: errors.New("ServerFaultCode: The object 'vim.Task:task-99999' has already been deleted or has not been completely created"), + want: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := isRetrieveMONotFound(tc.taskRef, tc.err) + if got != tc.want { + t.Errorf("isRetrieveMONotFound(%q, %v) = %v, want %v", tc.taskRef, tc.err, got, tc.want) + } + }) + } +} diff --git a/pkg/controller/vsphere/session/session.go b/pkg/controller/vsphere/session/session.go index 6a42d2ba20..dfc0686c6b 100644 --- a/pkg/controller/vsphere/session/session.go +++ b/pkg/controller/vsphere/session/session.go @@ -24,12 +24,14 @@ import ( "sync" "time" + "github.com/prometheus/client_golang/prometheus" "github.com/vmware/govmomi/vapi/rest" "github.com/vmware/govmomi/vapi/tags" "github.com/vmware/govmomi/vim25/mo" "github.com/vmware/govmomi/vim25/types" "github.com/google/uuid" + maometrics "github.com/openshift/machine-api-operator/pkg/metrics" "github.com/vmware/govmomi" "github.com/vmware/govmomi/find" "github.com/vmware/govmomi/object" @@ -37,7 +39,21 @@ import ( "k8s.io/klog/v2" ) -var sessionCache = map[string]Session{} +// sessionValidationTTL is how long we trust a cached session without +// re-running the SOAP SessionIsActive + REST session check pair. +// vCenter SOAP/REST sessions live far longer than this (hours), and +// every operation on a dead session returns an auth error, forcing a +// re-login on the next GetOrCreate anyway. +var sessionValidationTTL = 5 * time.Minute + +// sessionEntry wraps a cached Session with the time it was last validated. +type sessionEntry struct { + session Session + lastValidated time.Time +} + +// sessionCache is a cache of sessions keyed by server, username, and datacenter. +var sessionCache = map[string]sessionEntry{} var sessionMU sync.Mutex const ( @@ -45,6 +61,32 @@ const ( clientTimeout = 15 * time.Second ) +func instrumentSOAPClient(client *govmomi.Client, histogram *prometheus.HistogramVec) { + if client == nil || client.RoundTripper == nil { + return + } + if _, ok := client.RoundTripper.(*metricRoundTripper); ok { + return + } + client.RoundTripper = &metricRoundTripper{ + roundTripper: client.RoundTripper, + histogram: histogram, + } +} + +func instrumentRESTClient(client *rest.Client, histogram *prometheus.HistogramVec) { + if client == nil { + return + } + if _, ok := client.Transport.(*metricHTTPTransport); ok { + return + } + client.Transport = &metricHTTPTransport{ + roundTripper: client.Transport, + histogram: histogram, + } +} + // Session is a vSphere session with a configured Finder. // This implementation is inspired by cluster-api-provider-vsphere's session caching pattern // to avoid excessive vCenter login/logout cycles for REST API operations. @@ -55,6 +97,8 @@ type Session struct { Datacenter *object.Datacenter TagManager *tags.Manager + cachingTagManager *CachingTagsManager // per-session caching wrapper around TagManager + username string password string @@ -83,7 +127,13 @@ func GetOrCreate( defer sessionMU.Unlock() sessionKey := server + username + datacenter - if session, ok := sessionCache[sessionKey]; ok { + if entry, ok := sessionCache[sessionKey]; ok { + if time.Since(entry.lastValidated) < sessionValidationTTL { + klog.V(4).Infof("Reusing cached vSphere session within validation TTL") + return &entry.session, nil + } + session := entry.session + // Check both SOAP and REST session validity before reusing cached session. // This prevents reusing sessions where one connection type has expired. // Pattern adapted from cluster-api-provider-vsphere: @@ -104,6 +154,8 @@ func GetOrCreate( if sessionActive && restSessionActive { klog.V(3).Infof("Found active cached vSphere session with valid SOAP and REST connections") + entry.lastValidated = time.Now() + sessionCache[sessionKey] = entry return &session, nil } @@ -136,6 +188,7 @@ func GetOrCreate( if err != nil { return nil, fmt.Errorf("error setting up new vSphere SOAP client: %w", err) } + instrumentSOAPClient(client, maometrics.VsphereRequestDurationSeconds) // Set up user agent before login for being able to track mapi component in vcenter sessions list client.UserAgent = "machineAPIvSphereProvider" if err := client.Login(ctx, url.UserPassword(username, password)); err != nil { @@ -163,6 +216,7 @@ func GetOrCreate( // Pattern adapted from cluster-api-provider-vsphere: // https://github.com/kubernetes-sigs/cluster-api-provider-vsphere/blob/main/pkg/session/session.go#L196-L205 restClient := rest.NewClient(session.Client.Client) + instrumentRESTClient(restClient, maometrics.VsphereRequestDurationSeconds) if err := restClient.Login(ctx, url.UserPassword(username, password)); err != nil { // Cleanup SOAP session on REST login failure if logoutErr := client.Logout(ctx); logoutErr != nil { @@ -171,9 +225,10 @@ func GetOrCreate( return nil, fmt.Errorf("unable to login REST client to vCenter: %w", err) } session.TagManager = tags.NewManager(restClient) + session.cachingTagManager = newTagsCachingClient(session.TagManager, sessionKey) // Cache the session. - sessionCache[sessionKey] = session + sessionCache[sessionKey] = sessionEntry{session: session, lastValidated: time.Now()} return &session, nil } @@ -251,59 +306,9 @@ func (s *Session) GetTask(ctx context.Context, taskRef string) (*mo.Task, error) return &obj, nil } -// GetCachingTagsManager returns a CachingTagsManager that wraps the cached TagManager. -// This replaces the previous WithCachingTagsManager pattern which created new sessions -// on every call. The returned manager uses the session's cached REST client. +// GetCachingTagsManager returns the per-session CachingTagsManager that wraps +// the cached TagManager. It is created once when the session is created, +// so no new vCenter login/logout happens on access. func (s *Session) GetCachingTagsManager() *CachingTagsManager { - return newTagsCachingClient(s.TagManager, s.sessionKey) -} - -// WithRestClient is deprecated. Use s.TagManager directly instead. -// This function is maintained for backward compatibility but creates excessive -// vCenter login/logout cycles. Migration path: replace callback pattern with -// direct access to s.TagManager. -// -// Deprecated: Use s.TagManager for direct REST client access. -func (s *Session) WithRestClient(ctx context.Context, f func(c *rest.Client) error) error { - klog.Warning("WithRestClient is deprecated and causes excessive vCenter logouts. Use s.TagManager directly instead.") - c := rest.NewClient(s.Client.Client) - - user := url.UserPassword(s.username, s.password) - if err := c.Login(ctx, user); err != nil { - return err - } - - defer func() { - if err := c.Logout(ctx); err != nil { - klog.Errorf("Failed to logout: %v", err) - } - }() - - return f(c) -} - -// WithCachingTagsManager is deprecated. Use s.GetCachingTagsManager() instead. -// This function is maintained for backward compatibility but creates excessive -// vCenter login/logout cycles. Migration path: replace callback pattern with -// direct call to s.GetCachingTagsManager(). -// -// Deprecated: Use s.GetCachingTagsManager() for cached tag manager access. -func (s *Session) WithCachingTagsManager(ctx context.Context, f func(m *CachingTagsManager) error) error { - klog.Warning("WithCachingTagsManager is deprecated and causes excessive vCenter logouts. Use s.GetCachingTagsManager() instead.") - c := rest.NewClient(s.Client.Client) - - user := url.UserPassword(s.username, s.password) - if err := c.Login(ctx, user); err != nil { - return err - } - - defer func() { - if err := c.Logout(ctx); err != nil { - klog.Errorf("Failed to logout: %v", err) - } - }() - - m := newTagsCachingClient(tags.NewManager(c), s.sessionKey) - - return f(m) + return s.cachingTagManager } diff --git a/pkg/controller/vsphere/session/session_test.go b/pkg/controller/vsphere/session/session_test.go index a350ea93b4..aad8a54c1a 100644 --- a/pkg/controller/vsphere/session/session_test.go +++ b/pkg/controller/vsphere/session/session_test.go @@ -63,6 +63,50 @@ func initSimulator(t *testing.T) (*simulator.Model, *Session, *simulator.Server) return model, authSession, server } +func TestGetOrCreateSkipsValidationWithinTTL(t *testing.T) { + model, session, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + oldTTL := sessionValidationTTL + sessionValidationTTL = time.Hour + defer func() { sessionValidationTTL = oldTTL }() + + pass, _ := server.URL.User.Password() + key := server.URL.Host + server.URL.User.Username() + + // Simulate an old validation so the first call must validate (and pass). + sessionMU.Lock() + entry := sessionCache[key] + entry.lastValidated = time.Now().Add(-time.Hour - time.Second) + sessionCache[key] = entry + sessionMU.Unlock() + + s2, err := GetOrCreate(context.TODO(), server.URL.Host, "", + server.URL.User.Username(), pass, true) + if err != nil { + t.Fatalf("expected cached session reuse, got error: %v", err) + } + if s2.Client.Client != session.Client.Client { + t.Fatal("expected the same underlying SOAP client to be reused") + } + + // Second call inside the TTL: no validation path taken at all. + sessionMU.Lock() + validated := sessionCache[key].lastValidated + sessionMU.Unlock() + _, err = GetOrCreate(context.TODO(), server.URL.Host, "", + server.URL.User.Username(), pass, true) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + sessionMU.Lock() + if sessionCache[key].lastValidated != validated { + t.Fatal("lastValidated changed: validation ran inside the TTL") + } + sessionMU.Unlock() +} + func TestFindVMByName(t *testing.T) { model, session, server := initSimulator(t) defer model.Remove() diff --git a/pkg/controller/vsphere/session/tag_ids_caching_client.go b/pkg/controller/vsphere/session/tag_ids_caching_client.go index d6a053d0bc..bf95e1cbf2 100644 --- a/pkg/controller/vsphere/session/tag_ids_caching_client.go +++ b/pkg/controller/vsphere/session/tag_ids_caching_client.go @@ -63,29 +63,47 @@ func (c *cacheMap) Delete(key string) { c.internalMap.Delete(key) } -type tagsAndCategoriesCache struct { - tags cacheMap - categories cacheMap +type cachedObject struct { + obj any + expiresAt int64 +} + +func (co cachedObject) expired() bool { + return time.Now().UnixNano() > co.expiresAt +} + +type objectCacheMap struct { + internalMap sync.Map } -var sessionAnnotatedCache = map[string]*tagsAndCategoriesCache{} -var sessionCacheMU sync.Mutex +// SetWithTTL stores the object by the given key with lifetime specified in ttl parameter. +func (c *objectCacheMap) SetWithTTL(key string, obj any, ttl time.Duration) { + c.internalMap.Store(key, cachedObject{obj: obj, expiresAt: time.Now().Add(ttl).UnixNano()}) +} -func getOrCreateSessionCache(sessionKey string) *tagsAndCategoriesCache { - sessionCacheMU.Lock() - defer sessionCacheMU.Unlock() +// Set stores the object with the default 12h lifetime. +func (c *objectCacheMap) Set(key string, obj any) { + c.SetWithTTL(key, obj, defaultCacheTTL) +} - cache, found := sessionAnnotatedCache[sessionKey] +// Get returns the cached object. Second return value indicates a hit. +func (c *objectCacheMap) Get(key string) (any, bool) { + item, found := c.internalMap.Load(key) if !found { - cache = &tagsAndCategoriesCache{} - sessionAnnotatedCache[sessionKey] = cache - return cache + return nil, false } - return cache + co := item.(cachedObject) + if co.expired() { + klog.V(4).Infof("object cache item with key %s expired, invalidating", key) + c.internalMap.Delete(key) + return nil, false + } + return co.obj, true } // CachingTagsManager wraps tags.Manager from vSphere SDK for -// cache mapping between tags or categories name and their ids. +// cache mapping between tags or categories name and their ids, +// and for caching the full tag/category objects themselves. // Reasoning behind this is the implementation details of tags/categories lookup by name, // to find a tag/category by name vSphere SDK gets a list of ids and then makes an additional request // for every object till it will not find matched names. Such peculiarity causes a huge performance degradation @@ -94,17 +112,18 @@ func getOrCreateSessionCache(sessionKey string) *tagsAndCategoriesCache { // // See tags.Manager methods for more details: https://github.com/vmware/govmomi/blob/a2fb82dc55a8eb00932233aa8028ce97140df784/vapi/tags/tags.go#L172 // -// This structure is intended to be used from a Session instance (Session.WithCachingTagsManager method specifically) presented in this module. +// This structure is intended to be used from a Session instance (Session.GetCachingTagsManager method specifically) presented in this module. type CachingTagsManager struct { *tags.Manager - sessionKey string // different vCenters might have a different tags in it + tags cacheMap // name -> ID + categories cacheMap // name -> ID + tagObjects objectCacheMap // name (and ID) -> *tags.Tag + categoryObjects objectCacheMap // name (and ID) -> *tags.Category } -func newTagsCachingClient(tagsManager *tags.Manager, sessionKey string) *CachingTagsManager { - return &CachingTagsManager{ - tagsManager, sessionKey, - } +func newTagsCachingClient(tagsManager *tags.Manager, _ string) *CachingTagsManager { + return &CachingTagsManager{Manager: tagsManager} } // IsName returns true if the id is not an urn. @@ -121,109 +140,108 @@ func isObjectNotFoundErr(err error) bool { return err != nil && strings.HasSuffix(err.Error(), http.StatusText(http.StatusNotFound)) } -// GetTag fetches the tag information for the given identifier. -// The id parameter can be a Tag ID or Tag Name. -// This method shadows original tags.Manager method and caches mapping between -// tag name and its id. In case ID was passed, the original method from tags.Manager would be used. +// lookupObject implements the lookup flow shared by GetTag and GetCategory: +// object-cache hit -> name->id cache (notFoundValue sentinel, invalidation +// fallback when a cached id goes stale) -> fetch by id or name, refilling +// both caches on success. // -// In case if a tag was not found in vCenter, this would be cached for 12 hours and lookup won't happen till cache expiration. -func (t *CachingTagsManager) GetTag(ctx context.Context, id string) (*tags.Tag, error) { - if !IsName(id) { // if id is passed no cache check needed, use original GetTag method instantly - return t.Manager.GetTag(ctx, id) +// Not-found results are cached for 12 hours (defaultCacheTTL) because vCenter +// REST by-name lookups are expensive: for every missing object the number of +// requests equals the number of categories/tags. See govmomi vapi/tags and +// the vCenter REST API docs referenced in the original implementation. +func lookupObject[T any]( + ctx context.Context, + kind string, + objCache *objectCacheMap, + idCache *cacheMap, + fetchByID func(ctx context.Context, id string) (T, error), + nameOf func(T) string, + idOf func(T) string, + id string, +) (T, error) { + var zero T + + if obj, ok := objCache.Get(id); ok { + if o, ok := obj.(T); ok { + return o, nil + } } - cache := getOrCreateSessionCache(t.sessionKey) - cachedTagID, found := cache.tags.Get(id) - if found { - klog.V(4).Infof("tag %s: found cached tag id value", id) - if cachedTagID == notFoundValue { - klog.V(4).Infof("tag %s: cache contains special value indicates that tag was not found when cache was filled, treating as non existed tag", id) - return nil, fmt.Errorf("%s", notFoundErrMessage) + if !IsName(id) { + obj, err := fetchByID(ctx, id) + if err == nil { + objCache.Set(id, obj) } + return obj, err + } - tag, err := t.Manager.GetTag(ctx, cachedTagID) - if err != nil { - if isObjectNotFoundErr(err) { - klog.V(3).Infof("tag %s: tag was not found in vCenter by cached id, invalidating cache", id) - // if tag not found, invalidate the cache and fallback to the default search method - cache.tags.Delete(id) - return t.Manager.GetTag(ctx, id) + cachedID, found := idCache.Get(id) + if found { + if cachedID == notFoundValue { + klog.V(4).Infof("%s %s: cache indicates %s does not exist", kind, id, kind) + return zero, fmt.Errorf("%s", notFoundErrMessage) + } + obj, err := fetchByID(ctx, cachedID) + if err != nil && isObjectNotFoundErr(err) { + klog.V(3).Infof("%s %s: %s was not found in vCenter by cached id, invalidating cache", kind, id, kind) + // if not found, invalidate the name->id cache and fallback to the by-name lookup + idCache.Delete(id) + obj, err = fetchByID(ctx, id) + if err == nil { + // fallback found it by name, refill both caches so the next call hits + idCache.Set(nameOf(obj), idOf(obj)) + objCache.Set(nameOf(obj), obj) + objCache.Set(idOf(obj), obj) } - return tag, err + return obj, err + } + if err == nil { + objCache.Set(id, obj) } - return tag, nil + return obj, err } - klog.V(3).Infof("tag %s: tags cache miss, trying to find tag by name, it might take time", id) - tag, err := t.Manager.GetTag(ctx, id) + klog.V(3).Infof("%s %s: %s cache miss, trying to find %s by name", kind, id, kind, kind) + obj, err := fetchByID(ctx, id) if err != nil { if isObjectNotFoundErr(err) { - klog.V(3).Infof("tag %s not found in vCenter, caching", id) - // Caching the fact that tag was not found due to performance issues with lookup tag by name. - // In case when object does not exist amount of requests equals the amount of categories should happen, - // because of vCenter rest api design. - // For more context see vcenter rest api documentation and original method implementation: - // https://developer.vmware.com/apis/vsphere-automation/v7.0U1/cis/rest/com/vmware/cis/tagging/tag/idtag_id/get/ - // https://developer.vmware.com/apis/vsphere-automation/v7.0U1/cis/rest/com/vmware/cis/tagging/tag/get/ - // https://github.com/vmware/govmomi/blob/a2fb82dc55a8eb00932233aa8028ce97140df784/vapi/tags/tags.go#L121 - cache.tags.Set(id, notFoundValue) + klog.V(3).Infof("%s %s not found in vCenter, caching", kind, id) + idCache.Set(id, notFoundValue) } - return tag, err + return obj, err } - cache.tags.Set(tag.Name, tag.ID) - return tag, err + idCache.Set(nameOf(obj), idOf(obj)) + objCache.Set(nameOf(obj), obj) + objCache.Set(idOf(obj), obj) + return obj, err +} + +// GetTag fetches the tag information for the given identifier. +// The id parameter can be a Tag ID or Tag Name. +// This method shadows original tags.Manager method and caches mapping between +// tag name and its id as well as the full tag object, so any hit path +// (object cache or name->id cache) returns without REST calls. +// +// In case if a tag was not found in vCenter, this would be cached for 12 hours and lookup won't happen till cache expiration. +func (t *CachingTagsManager) GetTag(ctx context.Context, id string) (*tags.Tag, error) { + return lookupObject(ctx, "tag", &t.tagObjects, &t.tags, t.Manager.GetTag, + func(tag *tags.Tag) string { return tag.Name }, + func(tag *tags.Tag) string { return tag.ID }, + id) } // GetCategory fetches the category information for the given identifier. // The id parameter can be a Category ID or Category Name. // This method shadows original tags.Manager method and caches mapping between -// tag name and its id. In case ID was passed, the original method from tags.Manager would be used. +// category name and its id as well as the full category object, so any hit path +// (object cache or name->id cache) returns without REST calls. // -// In case if a tag was not found in vCenter, this would be cached for 12 hours and lookup won't happen till cache expiration. +// In case if a category was not found in vCenter, this would be cached for 12 hours and lookup won't happen till cache expiration. func (t *CachingTagsManager) GetCategory(ctx context.Context, id string) (*tags.Category, error) { - if !IsName(id) { // if id is passed no cache check needed, use original GetTag method instantly - return t.Manager.GetCategory(ctx, id) - } - - cache := getOrCreateSessionCache(t.sessionKey) - cachedCategoryID, found := cache.categories.Get(id) - if found { - klog.V(4).Infof("category %s: found cached category id value", id) - if cachedCategoryID == notFoundValue { - klog.V(4).Infof("category %s: cache contains special value indicates that tag was not found when cache was filled, treating as non existing category", id) - return nil, fmt.Errorf("%s", notFoundErrMessage) - } - - category, err := t.Manager.GetCategory(ctx, cachedCategoryID) - if err != nil { - if isObjectNotFoundErr(err) { - klog.V(3).Infof("category %s: category was not found in vCenter by cached id, invalidating id cache", id) - cache.categories.Delete(id) // if category not found, invalidate the cache and fallback to the default search method - return t.Manager.GetCategory(ctx, id) - } - return category, err - } - return category, nil - } - - klog.V(3).Infof("category %s: categories cache miss, trying to find category by name, it might take time", id) - category, err := t.Manager.GetCategory(ctx, id) - if err != nil { - if isObjectNotFoundErr(err) { - klog.V(3).Infof("category %s not found in vCenter, caching", id) - // Caching the fact that category was not found due to performance issues with lookup category by name. - // In case when object does not exist amount of requests equals the amount of categories should happen, - // because of vCenter rest api design. - // For more context see vcenter rest api documentation and original method implementation: - // https://developer.vmware.com/apis/vsphere-automation/v7.0U1/cis/rest/com/vmware/cis/tagging/category/idcategory_id/get/ - // https://developer.vmware.com/apis/vsphere-automation/v7.0U1/cis/rest/com/vmware/cis/tagging/category/get/ - // https://github.com/vmware/govmomi/blob/a2fb82dc55a8eb00932233aa8028ce97140df784/vapi/tags/categories.go#L122 - cache.categories.Set(id, notFoundValue) - } - return category, err - } - cache.categories.Set(category.Name, category.ID) - return category, err + return lookupObject(ctx, "category", &t.categoryObjects, &t.categories, t.Manager.GetCategory, + func(category *tags.Category) string { return category.Name }, + func(category *tags.Category) string { return category.ID }, + id) } // ListTagsForCategory tag ids for the given category. diff --git a/pkg/controller/vsphere/session/test_ids_caching_client_test.go b/pkg/controller/vsphere/session/test_ids_caching_client_test.go index 19c8b05ede..f9ea8f08a8 100644 --- a/pkg/controller/vsphere/session/test_ids_caching_client_test.go +++ b/pkg/controller/vsphere/session/test_ids_caching_client_test.go @@ -10,6 +10,13 @@ import ( "github.com/vmware/govmomi/vapi/tags" ) +func requireNoErr(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } +} + func createTagsAndCategories(ctx context.Context, tagNames []string, categoryNames []string, m *CachingTagsManager, g Gomega) { testCategoryID, err := m.CreateCategory(ctx, &tags.Category{ AssociableTypes: []string{"VirtualMachine"}, @@ -44,10 +51,6 @@ func cleanupTagsAndCategories(ctx context.Context, m *CachingTagsManager, g Gome } } -func purgeCache() { - sessionAnnotatedCache = map[string]*tagsAndCategoriesCache{} -} - func TestGetTag(t *testing.T) { model, sessionObj, server := initSimulator(t) defer model.Remove() @@ -60,86 +63,103 @@ func TestGetTag(t *testing.T) { tagsToCreate := []string{"fooo", "bar", "baz", "fizz"} categoriesToCreate := []string{"fizz"} - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - defer cleanupTagsAndCategories(ctx, m, g) - defer purgeCache() - - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + m := newTagsCachingClient(sessionObj.TagManager, "") - tag, err := m.GetTag(ctx, "baz") - g.Expect(err).To(Succeed()) - g.Expect(tag).NotTo(BeNil()) + createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) + defer cleanupTagsAndCategories(ctx, m, g) - // check cache filled - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) - cachedTagId, found := getOrCreateSessionCache(sessionObj.sessionKey).tags.Get("baz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedTagId).Should(ContainSubstring("urn:")) + tag, err := m.GetTag(ctx, "baz") + g.Expect(err).To(Succeed()) + g.Expect(tag).NotTo(BeNil()) - return nil - }) - g.Expect(err).Should(Succeed()) + // check cache filled + cachedTagId, found := m.tags.Get("baz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedTagId).Should(ContainSubstring("urn:")) }) t.Run("Tag deleted from vCenter after being cached", func(t *testing.T) { g := NewWithT(t) - tagsToCreate := []string{"fooo", "bar", "baz", "fizz"} + tagsToCreate := []string{"fizz"} categoriesToCreate := []string{"fizz"} - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - defer purgeCache() + m := newTagsCachingClient(sessionObj.TagManager, "") - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - tag, err := m.GetTag(ctx, "fizz") - g.Expect(err).To(Succeed()) - g.Expect(tag).NotTo(BeNil()) + tag, err := m.GetTag(ctx, "fizz") + g.Expect(err).To(Succeed()) + g.Expect(tag).NotTo(BeNil()) + + // check cache filled + cachedTagId, found := m.tags.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedTagId).Should(ContainSubstring("urn:")) + + cleanupTagsAndCategories(ctx, m, g) + + // A fresh manager simulates an expired object cache: the name->id cache hit + // resolves to a stale id, the id lookup 404s, the name cache entry is + // invalidated and lookup falls back to the default by-name search. + m2 := newTagsCachingClient(sessionObj.TagManager, "") + m2.tags.Set("fizz", tag.ID) + + _, err = m2.GetTag(ctx, "fizz") + // Cache should be invalidated and not found err returned + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + _, found = m2.tags.Get("fizz") + g.Expect(found).To(BeFalse()) + + _, err = m2.GetTag(ctx, "fizz") + // Not found value should be landed to the cache after next call + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + cachedTagId, found = m2.tags.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) + }) - // check cache filled - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) - cachedTagId, found := getOrCreateSessionCache(sessionObj.sessionKey).tags.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedTagId).Should(ContainSubstring("urn:")) + t.Run("Tag 404 fallback success refills cache", func(t *testing.T) { + g := NewWithT(t) + m := newTagsCachingClient(sessionObj.TagManager, "") - cleanupTagsAndCategories(ctx, m, g) + createTagsAndCategories(ctx, []string{"fizz"}, []string{"fizz"}, m, g) + defer cleanupTagsAndCategories(ctx, m, g) - _, err = m.GetTag(ctx, "fizz") - // Cache should be invalidated and not found err returned - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - _, found = getOrCreateSessionCache(sessionObj.sessionKey).tags.Get("fizz") - g.Expect(found).To(BeFalse()) + tag, err := m.GetTag(ctx, "fizz") + g.Expect(err).To(Succeed()) - _, err = m.GetTag(ctx, "fizz") - // Not found value should be landed to the cache after next call - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - cachedTagId, found = getOrCreateSessionCache(sessionObj.sessionKey).tags.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) + // A fresh manager simulates an expired object cache holding a stale id: + // the id lookup 404s and the by-name fallback finds the tag again. + m2 := newTagsCachingClient(sessionObj.TagManager, "") + m2.tags.Set("fizz", "urn:stale") - return nil - }) - g.Expect(err).Should(Succeed()) + tag2, err := m2.GetTag(ctx, "fizz") + g.Expect(err).To(Succeed()) + g.Expect(tag2.ID).To(BeEquivalentTo(tag.ID)) + + // Fallback success must refill both caches so the next call hits + cachedID, found := m2.tags.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedID).To(BeEquivalentTo(tag.ID)) + obj, found := m2.tagObjects.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(obj.(*tags.Tag).ID).To(BeEquivalentTo(tag.ID)) }) t.Run("Tag not found", func(t *testing.T) { g := NewWithT(t) - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - defer purgeCache() - - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + m := newTagsCachingClient(sessionObj.TagManager, "") - _, err := m.GetTag(ctx, "fizz") - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - cachedTagId, found := getOrCreateSessionCache(sessionObj.sessionKey).tags.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) + _, err := m.GetTag(ctx, "fizz") + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + cachedTagId, found := m.tags.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) - return nil - }) - g.Expect(err).Should(Succeed()) + // Second lookup is served from the not-found cache value + _, err = m.GetTag(ctx, "fizz") + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) }) } @@ -156,128 +176,196 @@ func TestGetCategory(t *testing.T) { tagsToCreate := []string{"fooo"} categoriesToCreate := []string{"fizz", "bazz", "eggz"} - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - defer cleanupTagsAndCategories(ctx, m, g) - defer purgeCache() - - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + m := newTagsCachingClient(sessionObj.TagManager, "") - cat, err := m.GetCategory(ctx, "fizz") - g.Expect(err).To(Succeed()) - g.Expect(cat).NotTo(BeNil()) + createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) + defer cleanupTagsAndCategories(ctx, m, g) - // check cache filled - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) - cachedCategoryId, found := getOrCreateSessionCache(sessionObj.sessionKey).categories.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedCategoryId).Should(ContainSubstring("urn:")) + cat, err := m.GetCategory(ctx, "fizz") + g.Expect(err).To(Succeed()) + g.Expect(cat).NotTo(BeNil()) - return nil - }) - g.Expect(err).Should(Succeed()) + // check cache filled + cachedCategoryId, found := m.categories.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedCategoryId).Should(ContainSubstring("urn:")) }) t.Run("Category deleted from vCenter after being cached", func(t *testing.T) { g := NewWithT(t) - tagsToCreate := []string{"fooo", "bar"} + tagsToCreate := []string{"fizz"} categoriesToCreate := []string{"fizz", "bazz", "eggz"} - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - defer purgeCache() + m := newTagsCachingClient(sessionObj.TagManager, "") - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + createTagsAndCategories(ctx, tagsToCreate, categoriesToCreate, m, g) - cat, err := m.GetCategory(ctx, "fizz") - g.Expect(err).To(Succeed()) - g.Expect(cat).NotTo(BeNil()) + cat, err := m.GetCategory(ctx, "fizz") + g.Expect(err).To(Succeed()) + g.Expect(cat).NotTo(BeNil()) + + // check cache filled + cachedCatId, found := m.categories.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedCatId).Should(ContainSubstring("urn:")) + + cleanupTagsAndCategories(ctx, m, g) + + // A fresh manager simulates an expired object cache: the name->id cache hit + // resolves to a stale id, the id lookup 404s, the name cache entry is + // invalidated and lookup falls back to the default by-name search. + m2 := newTagsCachingClient(sessionObj.TagManager, "") + m2.categories.Set("fizz", cat.ID) + + _, err = m2.GetCategory(ctx, "fizz") + // Cache should be invalidated and not found err returned + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + _, found = m2.categories.Get("fizz") + g.Expect(found).To(BeFalse()) + + _, err = m2.GetCategory(ctx, "fizz") + // Not found value should be landed to the cache after next call + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + cachedCatId, found = m2.categories.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedCatId).To(BeEquivalentTo(notFoundValue)) + }) - // check cache filled - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) - cachedCatId, found := getOrCreateSessionCache(sessionObj.sessionKey).categories.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedCatId).Should(ContainSubstring("urn:")) + t.Run("Category 404 fallback success refills cache", func(t *testing.T) { + g := NewWithT(t) + m := newTagsCachingClient(sessionObj.TagManager, "") - cleanupTagsAndCategories(ctx, m, g) + createTagsAndCategories(ctx, []string{"fizz"}, []string{"fizz", "bazz"}, m, g) + defer cleanupTagsAndCategories(ctx, m, g) - _, err = m.GetCategory(ctx, "fizz") - // Cache should be invalidated and not found err returned - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - _, found = getOrCreateSessionCache(sessionObj.sessionKey).categories.Get("fizz") - g.Expect(found).To(BeFalse()) + cat, err := m.GetCategory(ctx, "fizz") + g.Expect(err).To(Succeed()) - _, err = m.GetCategory(ctx, "fizz") - // Not found value should be landed to the cache after next call - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - cachedCatId, found = getOrCreateSessionCache(sessionObj.sessionKey).categories.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedCatId).To(BeEquivalentTo(notFoundValue)) + // A fresh manager simulates an expired object cache holding a stale id: + // the id lookup 404s and the by-name fallback finds the category again. + m2 := newTagsCachingClient(sessionObj.TagManager, "") + m2.categories.Set("fizz", "urn:stale") - return nil - }) - g.Expect(err).Should(Succeed()) + cat2, err := m2.GetCategory(ctx, "fizz") + g.Expect(err).To(Succeed()) + g.Expect(cat2.ID).To(BeEquivalentTo(cat.ID)) + + // Fallback success must refill both caches so the next call hits + cachedID, found := m2.categories.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedID).To(BeEquivalentTo(cat.ID)) + obj, found := m2.categoryObjects.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(obj.(*tags.Category).ID).To(BeEquivalentTo(cat.ID)) }) t.Run("Category not found", func(t *testing.T) { g := NewWithT(t) - err := sessionObj.WithCachingTagsManager(context.TODO(), func(m *CachingTagsManager) error { - defer purgeCache() + m := newTagsCachingClient(sessionObj.TagManager, "") - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + _, err := m.GetCategory(ctx, "fizz") + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) + cachedTagId, found := m.categories.Get("fizz") + g.Expect(found).To(BeTrue()) + g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) - _, err := m.GetCategory(ctx, "fizz") - g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) - cachedTagId, found := getOrCreateSessionCache(sessionObj.sessionKey).categories.Get("fizz") - g.Expect(found).To(BeTrue()) - g.Expect(cachedTagId).To(BeEquivalentTo(notFoundValue)) - - return nil - }) - g.Expect(err).Should(Succeed()) + // Second lookup is served from the not-found cache value + _, err = m.GetCategory(ctx, "fizz") + g.Expect(err.Error()).To(ContainSubstring("404 Not Found")) }) } -func TestSessionCacheGetter(t *testing.T) { - defer purgeCache() +// objectCacheHit: a second fetch by the same ID (name or urn) must not +// re-fetch from the REST manager. We assert via pointer equality of the +// returned object, which only holds if the cached *tags.Tag is reused. +func TestGetTagObjectCache(t *testing.T) { + model, sessionObj, server := initSimulator(t) + defer model.Remove() + defer server.Close() g := NewWithT(t) - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) + m := newTagsCachingClient(sessionObj.TagManager, "") + ctx := context.Background() - getOrCreateSessionCache("foo") - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) - getOrCreateSessionCache("foo") - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(1)) + createTagsAndCategories(ctx, []string{"cache-tag"}, []string{"cache-cat"}, m, g) + defer cleanupTagsAndCategories(ctx, m, g) - getOrCreateSessionCache("bar") - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(2)) - getOrCreateSessionCache("bar") - g.Expect(len(sessionAnnotatedCache)).To(BeEquivalentTo(2)) + byName, err := m.GetTag(ctx, "cache-tag") + requireNoErr(t, err) + + byID, err := m.GetTag(ctx, byName.ID) + requireNoErr(t, err) + + if byName != byID { + t.Fatalf("expected cached object reuse, got %p vs %p", byName, byID) + } } -func TestValuesExpiration(t *testing.T) { - defer purgeCache() +func TestGetCategoryObjectCache(t *testing.T) { + model, sessionObj, server := initSimulator(t) + defer model.Remove() + defer server.Close() + + g := NewWithT(t) + m := newTagsCachingClient(sessionObj.TagManager, "") + ctx := context.Background() + + createTagsAndCategories(ctx, []string{"cache-tag"}, []string{"cache-cat"}, m, g) + defer cleanupTagsAndCategories(ctx, m, g) + + byName, err := m.GetCategory(ctx, "cache-cat") + requireNoErr(t, err) + + byID, err := m.GetCategory(ctx, byName.ID) + requireNoErr(t, err) + + if byName != byID { + t.Fatalf("expected cached object reuse, got %p vs %p", byName, byID) + } +} +func TestValuesExpiration(t *testing.T) { g := NewWithT(t) - g.Expect(len(sessionAnnotatedCache)).To(BeZero()) - cache := getOrCreateSessionCache("foo") - cache.tags.SetWithTTL("foo", "bar", time.Millisecond*15) - cache.tags.SetWithTTL("baz", "eggz", time.Second*15) + cache := &cacheMap{} + cache.SetWithTTL("foo", "bar", time.Millisecond*15) + cache.SetWithTTL("baz", "eggz", time.Second*15) - value, found := cache.tags.Get("foo") + value, found := cache.Get("foo") g.Expect(found).To(BeTrue()) g.Expect(value).To(BeEquivalentTo("bar")) g.Eventually(func() (found bool) { - _, found = cache.tags.Get("foo") + _, found = cache.Get("foo") + return found + }, "20ms", "5ms").Should(BeFalse()) + + g.Consistently(func() (found bool) { + _, found = cache.Get("baz") + return found + }, "20ms", "5ms").Should(BeTrue()) +} + +func TestObjectValuesExpiration(t *testing.T) { + g := NewWithT(t) + + cache := &objectCacheMap{} + cache.SetWithTTL("foo", "bar", time.Millisecond*15) + cache.SetWithTTL("baz", "eggz", time.Second*15) + + obj, found := cache.Get("foo") + g.Expect(found).To(BeTrue()) + g.Expect(obj).To(BeEquivalentTo("bar")) + + g.Eventually(func() (found bool) { + _, found = cache.Get("foo") return found }, "20ms", "5ms").Should(BeFalse()) g.Consistently(func() (found bool) { - _, found = cache.tags.Get("baz") + _, found = cache.Get("baz") return found }, "20ms", "5ms").Should(BeTrue()) } diff --git a/pkg/controller/vsphere/session/transport_metrics.go b/pkg/controller/vsphere/session/transport_metrics.go new file mode 100644 index 0000000000..b3b7446d39 --- /dev/null +++ b/pkg/controller/vsphere/session/transport_metrics.go @@ -0,0 +1,105 @@ +package session + +import ( + "context" + "net/http" + "reflect" + "regexp" + "strconv" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/vmware/govmomi/vim25/soap" +) + +const unknownOperation = "unknown" + +var uuidPathSegment = regexp.MustCompile(`^[[:xdigit:]]{8}-[[:xdigit:]]{4}-[1-5][[:xdigit:]]{3}-[89abAB][[:xdigit:]]{3}-[[:xdigit:]]{12}$`) + +type metricRoundTripper struct { + roundTripper soap.RoundTripper + histogram *prometheus.HistogramVec +} + +func (t *metricRoundTripper) RoundTrip(ctx context.Context, req, res soap.HasFault) error { + operation := soapOperation(req) + start := time.Now() + err := t.roundTripper.RoundTrip(ctx, req, res) + status := "success" + if err != nil { + status = "error" + } + t.histogram.WithLabelValues("soap", operation, status).Observe(time.Since(start).Seconds()) + return err +} + +func soapOperation(req soap.HasFault) string { + if req == nil { + return unknownOperation + } + + typ := reflect.TypeOf(req) + value := reflect.ValueOf(req) + if typ.Kind() != reflect.Ptr || value.IsNil() || typ.Elem().Kind() != reflect.Struct { + return unknownOperation + } + + name := typ.Elem().Name() + if !strings.HasSuffix(name, "Body") || len(name) == len("Body") { + return unknownOperation + } + return strings.TrimSuffix(name, "Body") +} + +type metricHTTPTransport struct { + roundTripper http.RoundTripper + histogram *prometheus.HistogramVec +} + +func (t *metricHTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) { + start := time.Now() + response, err := t.transport().RoundTrip(req) + status := "success" + if err != nil || response == nil || response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + status = "error" + } + t.histogram.WithLabelValues("rest", restOperation(req), status).Observe(time.Since(start).Seconds()) + return response, err +} + +func (t *metricHTTPTransport) transport() http.RoundTripper { + if t.roundTripper == nil { + return http.DefaultTransport + } + return t.roundTripper +} + +func restOperation(req *http.Request) string { + if req == nil || req.URL == nil { + return "UNKNOWN " + unknownOperation + } + return req.Method + " " + normalizeRESTPath(req.URL.Path) +} + +func normalizeRESTPath(path string) string { + segments := strings.Split(path, "/") + for i, segment := range segments { + if isIDPathSegment(segment) { + segments[i] = "{id}" + } + } + return strings.Join(segments, "/") +} + +func isIDPathSegment(segment string) bool { + if segment == "" { + return false + } + lower := strings.ToLower(segment) + if strings.HasPrefix(lower, "urn:") || strings.HasPrefix(lower, "id:") || uuidPathSegment.MatchString(segment) { + return true + } + _, err := strconv.ParseUint(segment, 10, 64) + return err == nil +} diff --git a/pkg/controller/vsphere/session/transport_metrics_test.go b/pkg/controller/vsphere/session/transport_metrics_test.go new file mode 100644 index 0000000000..5e5b099e65 --- /dev/null +++ b/pkg/controller/vsphere/session/transport_metrics_test.go @@ -0,0 +1,266 @@ +package session + +import ( + "context" + "errors" + "io" + "net/http" + "net/url" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/vmware/govmomi" + "github.com/vmware/govmomi/vapi/rest" + "github.com/vmware/govmomi/vim25" + "github.com/vmware/govmomi/vim25/soap" +) + +type fakeSOAPRoundTripper struct { + calls int + req soap.HasFault + err error +} + +func (f *fakeSOAPRoundTripper) RoundTrip(_ context.Context, req, _ soap.HasFault) error { + f.calls++ + f.req = req + return f.err +} + +type metricSOAPRequestBody struct{} + +func (*metricSOAPRequestBody) Fault() *soap.Fault { return nil } + +type unexpectedSOAPRequest struct{} + +func (unexpectedSOAPRequest) Fault() *soap.Fault { return nil } + +type fakeHTTPRoundTripper struct { + calls int + response *http.Response + err error +} + +func (f *fakeHTTPRoundTripper) RoundTrip(*http.Request) (*http.Response, error) { + f.calls++ + return f.response, f.err +} + +func newTestHistogram(t *testing.T) (*prometheus.Registry, *prometheus.HistogramVec) { + t.Helper() + registry := prometheus.NewRegistry() + histogram := prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "test_vsphere_request_duration_seconds", + Help: "Test vSphere request duration.", + Buckets: []float64{0.1, 1}, + }, []string{"client", "operation", "status"}) + registry.MustRegister(histogram) + return registry, histogram +} + +func histogramSampleCount(t *testing.T, registry *prometheus.Registry, labels map[string]string) uint64 { + t.Helper() + families, err := registry.Gather() + if err != nil { + t.Fatal(err) + } + for _, family := range families { + for _, metric := range family.GetMetric() { + observed := make(map[string]string, len(metric.GetLabel())) + for _, label := range metric.GetLabel() { + observed[label.GetName()] = label.GetValue() + } + matches := len(observed) == len(labels) + for key, value := range labels { + if observed[key] != value { + matches = false + } + } + if matches && metric.GetHistogram() != nil { + return metric.GetHistogram().GetSampleCount() + } + } + } + return 0 +} + +func TestSessionInstrumentation(t *testing.T) { + _, histogram := newTestHistogram(t) + + soapInner := &fakeSOAPRoundTripper{} + vimClient := &vim25.Client{RoundTripper: soapInner} + client := &govmomi.Client{Client: vimClient} + instrumentSOAPClient(client, histogram) + if _, ok := client.RoundTripper.(*metricRoundTripper); !ok { + t.Fatalf("SOAP RoundTripper = %T, want *metricRoundTripper", client.RoundTripper) + } + soapTransport := client.RoundTripper + instrumentSOAPClient(client, histogram) + if client.RoundTripper != soapTransport { + t.Fatal("instrumentSOAPClient wrapped the SOAP transport twice") + } + + restClient := &rest.Client{} + restClient.Client = &soap.Client{} + instrumentRESTClient(restClient, histogram) + if _, ok := restClient.Transport.(*metricHTTPTransport); !ok { + t.Fatalf("REST Transport = %T, want *metricHTTPTransport", restClient.Transport) + } + restTransport := restClient.Transport + instrumentRESTClient(restClient, histogram) + if restClient.Transport != restTransport { + t.Fatal("instrumentRESTClient wrapped the REST transport twice") + } +} + +func TestMetricRoundTripper(t *testing.T) { + registry, histogram := newTestHistogram(t) + inner := &fakeSOAPRoundTripper{} + transport := &metricRoundTripper{roundTripper: inner, histogram: histogram} + + if err := transport.RoundTrip(context.Background(), &metricSOAPRequestBody{}, &metricSOAPRequestBody{}); err != nil { + t.Fatalf("RoundTrip() error = %v", err) + } + if inner.calls != 1 { + t.Fatalf("inner RoundTrip() calls = %d, want 1", inner.calls) + } + if inner.req == nil { + t.Fatal("inner RoundTrip() did not receive request") + } + if got := histogramSampleCount(t, registry, map[string]string{ + "client": "soap", "operation": "metricSOAPRequest", "status": "success", + }); got != 1 { + t.Fatalf("success sample count = %d, want 1", got) + } + + expectedErr := errors.New("soap request failed") + inner.err = expectedErr + if err := transport.RoundTrip(context.Background(), &metricSOAPRequestBody{}, &metricSOAPRequestBody{}); !errors.Is(err, expectedErr) { + t.Fatalf("RoundTrip() error = %v, want %v", err, expectedErr) + } + if got := histogramSampleCount(t, registry, map[string]string{ + "client": "soap", "operation": "metricSOAPRequest", "status": "error", + }); got != 1 { + t.Fatalf("error sample count = %d, want 1", got) + } +} + +func TestMetricRoundTripperUnknownRequestDoesNotPanic(t *testing.T) { + registry, histogram := newTestHistogram(t) + inner := &fakeSOAPRoundTripper{} + transport := &metricRoundTripper{roundTripper: inner, histogram: histogram} + + if err := transport.RoundTrip(context.Background(), nil, &metricSOAPRequestBody{}); err != nil { + t.Fatalf("RoundTrip(nil) error = %v", err) + } + if err := transport.RoundTrip(context.Background(), unexpectedSOAPRequest{}, &metricSOAPRequestBody{}); err != nil { + t.Fatalf("RoundTrip(unexpected) error = %v", err) + } + if inner.calls != 2 { + t.Fatalf("inner RoundTrip() calls = %d, want 2", inner.calls) + } + if got := histogramSampleCount(t, registry, map[string]string{ + "client": "soap", "operation": "unknown", "status": "success", + }); got != 2 { + t.Fatalf("unknown request sample count = %d, want 2", got) + } +} + +func TestMetricHTTPTransport(t *testing.T) { + tests := []struct { + name string + requestURL string + response *http.Response + err error + status string + }{ + { + name: "success", + requestURL: "https://vcenter.example/rest/com/vmware/cis/tagging/tag?foo=bar", + response: &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(nil)}, + status: "success", + }, + { + name: "http error", + requestURL: "https://vcenter.example/rest/com/vmware/cis/tagging/tag/id/42", + response: &http.Response{StatusCode: http.StatusInternalServerError, Body: io.NopCloser(nil)}, + status: "error", + }, + { + name: "transport error", + requestURL: "https://vcenter.example/rest/com/vmware/cis/tagging/tag", + err: errors.New("transport failed"), + status: "error", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + registry, histogram := newTestHistogram(t) + inner := &fakeHTTPRoundTripper{response: test.response, err: test.err} + transport := &metricHTTPTransport{roundTripper: inner, histogram: histogram} + req, err := http.NewRequest(http.MethodGet, test.requestURL, nil) + if err != nil { + t.Fatal(err) + } + + response, gotErr := transport.RoundTrip(req) + if response != test.response { + t.Fatalf("response = %p, want %p", response, test.response) + } + if !errors.Is(gotErr, test.err) { + t.Fatalf("error = %v, want %v", gotErr, test.err) + } + if inner.calls != 1 { + t.Fatalf("inner RoundTrip() calls = %d, want 1", inner.calls) + } + operation := "GET /rest/com/vmware/cis/tagging/tag" + if test.name == "http error" { + operation += "/id/{id}" + } + if got := histogramSampleCount(t, registry, map[string]string{ + "client": "rest", "operation": operation, "status": test.status, + }); got != 1 { + t.Fatalf("sample count = %d, want 1", got) + } + }) + } +} + +func TestNormalizeRESTPath(t *testing.T) { + tests := []struct { + path string + want string + }{ + { + path: "/rest/com/vmware/cis/tagging/tag/id/urn:vmomi:InventoryServiceTag:abc", + want: "/rest/com/vmware/cis/tagging/tag/id/{id}", + }, + { + path: "/rest/com/vmware/cis/tagging/tag/id:urn:vmomi:InventoryServiceTag:abc", + want: "/rest/com/vmware/cis/tagging/tag/{id}", + }, + { + path: "/rest/com/vmware/cis/tagging/tag/id/550e8400-e29b-41d4-a716-446655440000", + want: "/rest/com/vmware/cis/tagging/tag/id/{id}", + }, + { + path: "/rest/com/vmware/cis/tagging/tag/id/42", + want: "/rest/com/vmware/cis/tagging/tag/id/{id}", + }, + } + for _, test := range tests { + t.Run(test.path, func(t *testing.T) { + if got := normalizeRESTPath(test.path); got != test.want { + t.Fatalf("normalizeRESTPath() = %q, want %q", got, test.want) + } + }) + } +} + +func TestRESTOperationUsesURLPath(t *testing.T) { + req := &http.Request{Method: http.MethodPost, URL: &url.URL{Path: "/rest/com/vmware/cis/tagging/tag", RawQuery: "foo=bar"}} + if got, want := restOperation(req), "POST /rest/com/vmware/cis/tagging/tag"; got != want { + t.Fatalf("restOperation() = %q, want %q", got, want) + } +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index ecd71a416f..78d2a9fff4 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -74,11 +74,19 @@ var ( Buckets: []float64{5, 10, 20, 30, 60, 90, 120, 180, 240, 300, 360, 480, 600}, }, []string{"phase"}, ) + + VsphereRequestDurationSeconds = prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "mapi_vsphere_request_duration_seconds", + Help: "Duration of requests made by the Machine API Operator to vSphere.", + Buckets: []float64{0.1, 0.25, 0.5, 1, 2, 5, 10, 20, 30, 60, 120, 180}, + }, []string{"client", "operation", "status"}, + ) ) func init() { prometheus.MustRegister(MachineCollectorUp) - metrics.Registry.MustRegister(MachinePhaseTransitionSeconds) + metrics.Registry.MustRegister(MachinePhaseTransitionSeconds, VsphereRequestDurationSeconds) metrics.Registry.MustRegister( failedInstanceCreateCount, failedInstanceUpdateCount, diff --git a/pkg/metrics/metrics_test.go b/pkg/metrics/metrics_test.go new file mode 100644 index 0000000000..8eb9a4eebe --- /dev/null +++ b/pkg/metrics/metrics_test.go @@ -0,0 +1,24 @@ +package metrics + +import ( + "testing" + + crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" +) + +func TestVsphereRequestDurationSecondsRegistered(t *testing.T) { + VsphereRequestDurationSeconds.WithLabelValues("soap", "RetrieveProperties", "success").Observe(1) + + families, err := crmetrics.Registry.Gather() + if err != nil { + t.Fatalf("gather metrics: %v", err) + } + + for _, family := range families { + if family.GetName() == "mapi_vsphere_request_duration_seconds" { + return + } + } + + t.Fatal("mapi_vsphere_request_duration_seconds was not registered") +}