Add processNextItem()

Author: xing-yang

pkg/sidecar-controller/snapshot_controller_base.go

@@ -149,6 +149,99 @@ func (ctrl *csiSnapshotSideCarController) enqueueContentWork(obj interface{}) {
 // contentWorker processes items from contentQueue. It must run only once,
 // syncContent is not assured to be reentrant.
 func (ctrl *csiSnapshotSideCarController) contentWorker() {
+	for ctrl.processNextItem() {
+	}
+}
+
+func (ctrl *csiSnapshotSideCarController) processNextItem() bool {
+	keyObj, quit := ctrl.contentQueue.Get()
+	if quit {
+		return false
+	}
+	defer ctrl.contentQueue.Done(keyObj)
+
+	if err := ctrl.syncContentByKey(keyObj.(string)); err != nil {
+		// Rather than wait for a full resync, re-add the key to the
+		// queue to be processed.
+		ctrl.contentQueue.AddRateLimited(keyObj)
+		klog.V(4).Infof("Failed to sync content %q, will retry again: %v", keyObj.(string), err)
+		return true
+	}
+
+	// Finally, if no error occurs we Forget this item so it does not
+	// get queued again until another change happens.
+	ctrl.contentQueue.Forget(keyObj)
+	return true
+}
+
+func (ctrl *csiSnapshotSideCarController) syncContentByKey(key string) error {
+	klog.V(5).Infof("syncContentByKey[%s]", key)
+	//key := keyObj.(string)
+	//klog.V(5).Infof("contentWorker[%s]", key)
+
+	_, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		klog.V(4).Infof("error getting name of snapshotContent %q to get snapshotContent from informer: %v", key, err)
+		return nil
+	}
+	content, err := ctrl.contentLister.Get(name)
+	// The content still exists in informer cache, the event must have
+	// been add/update/sync
+	if err == nil {
+		if ctrl.isDriverMatch(content) {
+			err = ctrl.updateContentInCacheStore(content)
+		}
+		if err != nil {
+			// If error occurs we add this item back to the queue
+			//ctrl.contentQueue.AddRateLimited(key)
+			return err
+		} /* else {
+			// If no error occurs we Forget this item so it does not
+			// get queued again until another change happens
+			klog.V(4).Infof("Forget snapshotContent %q so it does not get queued again until another change happens", key)
+			ctrl.contentQueue.Forget(key)
+		}*/
+		return nil
+	}
+	if !errors.IsNotFound(err) {
+		klog.V(2).Infof("error getting content %q from informer: %v", key, err)
+		return nil
+	}
+
+	// The content is not in informer cache, the event must have been
+	// "delete"
+	contentObj, found, err := ctrl.contentStore.GetByKey(key)
+	if err != nil {
+		klog.V(2).Infof("error getting content %q from cache: %v", key, err)
+		return nil
+	}
+	if !found {
+		// The controller has already processed the delete event and
+		// deleted the content from its cache
+		klog.V(2).Infof("deletion of content %q was already processed", key)
+		return nil
+	}
+	content, ok := contentObj.(*crdv1.VolumeSnapshotContent)
+	if !ok {
+		klog.Errorf("expected content, got %+v", content)
+		return nil
+	}
+	ctrl.deleteContentInCacheStore(content)
+	return nil
+	//}
+
+	/*for {
+		if quit := workFunc(); quit {
+			klog.Infof("content worker queue shutting down")
+			return
+		}
+	}*/
+	//return nil
+}
+
+// contentWorker processes items from contentQueue. It must run only once,
+// syncContent is not assured to be reentrant.
+/*func (ctrl *csiSnapshotSideCarController) contentWorker() {
 	workFunc := func() bool {
 		keyObj, quit := ctrl.contentQueue.Get()
 		if quit {
@@ -214,7 +307,7 @@ func (ctrl *csiSnapshotSideCarController) contentWorker() {
 			return
 		}
 	}
-}
+}*/
 
 // verify whether the driver specified in VolumeSnapshotContent matches the controller's driver name
 func (ctrl *csiSnapshotSideCarController) isDriverMatch(content *crdv1.VolumeSnapshotContent) bool {