Use diff prop collectors for two WaitForUpdates calls (#3182)

This patch updates the storagepool listner and listview
used to watch tasks so the two, distinct calls to
WaitForUpdatesEx use distinct PropertyCollectors.

This is necessary because the vSphere API docs do state
that WaitForUpdates and WaitForUpdatesEx do not support
concurrent access to the same PropertyCollector.

Without this patch, when running at scale, CSI can timeout
when waiting for updates because the session's default
PropertyCollector ends up in a blocking state on the
backend.

Author: akutz

pkg/common/cns-lib/volume/listview.go

@@ -339,7 +339,12 @@ func (l *ListViewImpl) listenToTaskUpdates() {
 
 		log.Info("Starting listening for task updates...")
 		log.Infof("waitForUpdatesContext %v", l.waitForUpdatesContext)
-		pc := property.DefaultCollector(l.virtualCenter.Client.Client)
+		pc, pcErr := property.DefaultCollector(l.virtualCenter.Client.Client).Create(l.ctx)
+		if pcErr != nil {
+			log.Errorf("failed to create PropertyCollector for WaitForUpdatesEx for ListView. error: %+v", pcErr)
+			l.mu.Unlock()
+			continue
+		}
 		l.isReady = true
 		log.Infof("listview ready state is %v", l.isReady)
 		l.mu.Unlock()
@@ -357,6 +362,12 @@ func (l *ListViewImpl) listenToTaskUpdates() {
 			// we only want this true while running the unit tests so the test can finish
 			return l.shouldStopListening
 		})
+
+		// Attempt to clean up the property collector using a new context to
+		// ensure it goes through. This call *might* fail if the session's
+		// auth has expired, but it is worth trying.
+		_ = pc.Destroy(context.Background())
+
 		// if property collector returns any errors,
 		// we want to immediately return a fault for all the pending tasks in the map
 		// note: this is not a task error but an error from the vc

pkg/syncer/storagepool/listener.go

@@ -118,8 +118,14 @@ func initListener(ctx context.Context, scWatchCntlr *StorageClassWatch,
 
 		for {
 			log.Infof("Starting property collector...")
-			p := property.DefaultCollector(spController.vc.Client.Client)
-			err := property.WaitForUpdatesEx(ctx, p, filter, func(updates []types.ObjectUpdate) bool {
+
+			pc, pcErr := property.DefaultCollector(spController.vc.Client.Client).Create(ctx)
+			if err != nil {
+				log.Errorf("Not able to create property collector. Restarting property collector. error: %+v", pcErr)
+				return
+			}
+
+			err := property.WaitForUpdatesEx(ctx, pc, filter, func(updates []types.ObjectUpdate) bool {
 				ctx := logger.NewContextWithLogger(ctx)
 				log = logger.GetLogger(ctx)
 				log.Debugf("Got %d property collector update(s)", len(updates))
@@ -159,6 +165,12 @@ func initListener(ctx context.Context, scWatchCntlr *StorageClassWatch,
 				log.Debugf("Done processing %d property collector update(s)", len(updates))
 				return false
 			})
+
+			// Attempt to clean up the property collector using a new context to
+			// ensure it goes through. This call *might* fail if the session's
+			// auth has expired, but it is worth trying.
+			_ = pc.Destroy(context.Background())
+
 			if err != nil {
 				log.Infof("Property collector session needs to be reestablished due to err: %v", err)
 				err = spController.vc.Connect(ctx)