ws-mananger: refactor volume snapshot watcher

Signed-off-by: JenTing Hsiao <[email protected]>

Author: jenting

components/ws-manager/cmd/run.go

@@ -33,7 +33,9 @@ import (
 	imgbldr "github.com/gitpod-io/gitpod/image-builder/api"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/manager"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/proxy"
+
 	volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
+	volumesnapshotclientv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 )
 
 // serveCmd represents the serve command
@@ -77,9 +79,6 @@ var runCmd = &cobra.Command{
 		}
 
 		kubeConfig, err := ctrl.GetConfig()
-		if err != nil {
-			log.WithError(err).Fatal("unable to create a Kubernetes API Client configuration")
-		}
 		if err != nil {
 			log.WithError(err).Fatal("unable to getting Kubernetes client config")
 		}
@@ -89,6 +88,11 @@ var runCmd = &cobra.Command{
 			log.WithError(err).Fatal("constructing Kubernetes client")
 		}
 
+		volumesnapshotclientset, err := volumesnapshotclientv1.NewForConfig(kubeConfig)
+		if err != nil {
+			log.WithError(err).Fatal("constructing volume snapshot client")
+		}
+
 		if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
 			log.WithError(err).Fatal("unable to set up health check")
 		}
@@ -106,7 +110,7 @@ var runCmd = &cobra.Command{
 			log.WithError(err).Fatal("cannot register Kubernetes volumesnapshotv1 schema - this should never happen")
 		}
 
-		mgmt, err := manager.New(cfg.Manager, mgr.GetClient(), clientset, cp)
+		mgmt, err := manager.New(cfg.Manager, mgr.GetClient(), clientset, volumesnapshotclientset, cp)
 		if err != nil {
 			log.WithError(err).Fatal("cannot create manager")
 		}
@@ -200,20 +204,6 @@ var runCmd = &cobra.Command{
 			log.WithError(err).Fatal("unable to create controller", "controller", "Pod")
 		}
 
-		// enable the volume snapshot controller when the VolumeSnapshot CRD exists
-		_, err = clientset.DiscoveryClient.ServerResourcesForGroupVersion(volumesnapshotv1.SchemeGroupVersion.String())
-		if err == nil {
-			err = (&manager.VolumeSnapshotReconciler{
-				Monitor: monitor,
-				Client:  mgr.GetClient(),
-				Log:     ctrl.Log.WithName("controllers").WithName("VolumeSnapshot"),
-				Scheme:  mgr.GetScheme(),
-			}).SetupWithManager(mgr)
-			if err != nil {
-				log.WithError(err).Fatal("unable to create controller", "controller", "VolumeSnapshot")
-			}
-		}
-
 		if cfg.PProf.Addr != "" {
 			go pprof.Serve(cfg.PProf.Addr)
 		}

components/ws-manager/pkg/manager/integration_test.go

@@ -45,6 +45,8 @@ import (
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/manager/internal/grpcpool"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/test"
+
+	volumesnapshotclientv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 )
 
 var integrationFlag = flag.String("integration-test", "disabled", "configures integration tests. Valid values are disabled, local or a path to a kubeconfig file")
@@ -119,13 +121,19 @@ func forIntegrationTestGetManager(t *testing.T) *Manager {
 		return nil
 	}
 
+	volumesnapshotclientset, err := volumesnapshotclientv1.NewForConfig(cfg)
+	if err != nil {
+		t.Errorf("cannt create test environment: %v", err)
+		return nil
+	}
+
 	ctrlClient, err := ctrler_client.New(cfg, ctrler_client.Options{Scheme: scheme})
 	if err != nil {
 		t.Errorf("cannot create test environment: %v", err)
 		return nil
 	}
 
-	m, err := New(config, ctrlClient, clientset, &layer.Provider{Storage: &storage.PresignedNoopStorage{}})
+	m, err := New(config, ctrlClient, clientset, volumesnapshotclientset, &layer.Provider{Storage: &storage.PresignedNoopStorage{}})
 	if err != nil {
 		t.Fatalf("cannot create manager: %s", err.Error())
 	}

components/ws-manager/pkg/manager/manager.go

@@ -52,16 +52,19 @@ import (
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/clock"
 	"github.com/gitpod-io/gitpod/ws-manager/pkg/manager/internal/grpcpool"
+
 	volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
+	volumesnapshotclientv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 )
 
 // Manager is a kubernetes backed implementation of a workspace manager
 type Manager struct {
-	Config    config.Configuration
-	Clientset client.Client
-	RawClient kubernetes.Interface
-	Content   *layer.Provider
-	OnChange  func(context.Context, *api.WorkspaceStatus)
+	Config               config.Configuration
+	Clientset            client.Client
+	RawClient            kubernetes.Interface
+	VolumeSnapshotClient volumesnapshotclientv1.Interface
+	Content              *layer.Provider
+	OnChange             func(context.Context, *api.WorkspaceStatus)
 
 	activity sync.Map
 	clock    *clock.HLC
@@ -127,7 +130,7 @@ const (
 )
 
 // New creates a new workspace manager
-func New(config config.Configuration, client client.Client, rawClient kubernetes.Interface, cp *layer.Provider) (*Manager, error) {
+func New(config config.Configuration, client client.Client, rawClient kubernetes.Interface, volumesnapshotClient volumesnapshotclientv1.Interface, cp *layer.Provider) (*Manager, error) {
 	wsdaemonConnfactory, err := newWssyncConnectionFactory(config)
 	if err != nil {
 		return nil, err
@@ -138,14 +141,15 @@ func New(config config.Configuration, client client.Client, rawClient kubernetes
 	eventRecorder := broadcaster.NewRecorder(runtime.NewScheme(), corev1.EventSource{Component: "ws-manager"})
 
 	m := &Manager{
-		Config:        config,
-		Clientset:     client,
-		RawClient:     rawClient,
-		Content:       cp,
-		clock:         clock.System(),
-		subscribers:   make(map[string]chan *api.SubscribeResponse),
-		wsdaemonPool:  grpcpool.New(wsdaemonConnfactory, checkWSDaemonEndpoint(config.Namespace, client)),
-		eventRecorder: eventRecorder,
+		Config:               config,
+		Clientset:            client,
+		RawClient:            rawClient,
+		VolumeSnapshotClient: volumesnapshotClient,
+		Content:              cp,
+		clock:                clock.System(),
+		subscribers:          make(map[string]chan *api.SubscribeResponse),
+		wsdaemonPool:         grpcpool.New(wsdaemonConnfactory, checkWSDaemonEndpoint(config.Namespace, client)),
+		eventRecorder:        eventRecorder,
 	}
 	m.metrics = newMetrics(m)
 	m.OnChange = m.onChange

components/ws-manager/pkg/manager/monitor.go

@@ -27,10 +27,14 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	k8serr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
+	watchtools "k8s.io/client-go/tools/watch"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/gitpod-io/gitpod/common-go/kubernetes"
@@ -84,9 +88,6 @@ type Monitor struct {
 
 	OnError func(error)
 
-	notifyPod        map[string]chan string
-	notifyPodMapLock sync.Mutex
-
 	eventRecorder record.EventRecorder
 }
 
@@ -107,8 +108,6 @@ func (m *Manager) CreateMonitor() (*Monitor, error) {
 			log.WithError(err).Error("workspace monitor error")
 		},
 
-		notifyPod: make(map[string]chan string),
-
 		eventRecorder: m.eventRecorder,
 	}
 	res.eventpool = workpool.NewEventWorkerPool(res.handleEvent)
@@ -152,62 +151,13 @@ func (m *Monitor) handleEvent(evt watch.Event) {
 	switch evt.Object.(type) {
 	case *corev1.Pod:
 		err = m.onPodEvent(evt)
-	case *volumesnapshotv1.VolumeSnapshot:
-		err = m.onVolumesnapshotEvent(evt)
 	}
 
 	if err != nil {
 		m.OnError(err)
 	}
 }
 
-func (m *Monitor) onVolumesnapshotEvent(evt watch.Event) error {
-	vs, ok := evt.Object.(*volumesnapshotv1.VolumeSnapshot)
-	if !ok {
-		return xerrors.Errorf("received non-volume-snapshot event")
-	}
-
-	log := log.WithField("volumesnapshot", vs.Name)
-
-	if vs.Spec.Source.PersistentVolumeClaimName == nil {
-		// there is no pvc name within the VolumeSnapshot object
-		log.Warn("the spec.source.persistentVolumeClaimName is empty")
-		return nil
-	}
-
-	// the pod name is 1:1 mapping to pvc name
-	podName := *vs.Spec.Source.PersistentVolumeClaimName
-	log = log.WithField("pod", podName)
-
-	// get the pod resource
-	var pod corev1.Pod
-	err := m.manager.Clientset.Get(context.Background(), types.NamespacedName{Namespace: vs.Namespace, Name: podName}, &pod)
-	if err != nil && !k8serr.IsNotFound(err) {
-		log.WithError(err).Warnf("cannot get pod")
-	}
-
-	if vs.Status == nil || vs.Status.ReadyToUse == nil || !*vs.Status.ReadyToUse || vs.Status.BoundVolumeSnapshotContentName == nil {
-		if !pod.CreationTimestamp.IsZero() {
-			m.eventRecorder.Eventf(&pod, corev1.EventTypeNormal, "VolumeSnapshot", "Volume snapshot %q is in progress", vs.Name)
-		}
-		return nil
-	}
-
-	vsc := *vs.Status.BoundVolumeSnapshotContentName
-	log.Debugf("the vsc %s is ready to use", vsc)
-	if !pod.CreationTimestamp.IsZero() {
-		m.eventRecorder.Eventf(&pod, corev1.EventTypeNormal, "VolumeSnapshot", "Volume snapshot %q is ready to use", vs.Name)
-	}
-
-	m.notifyPodMapLock.Lock()
-	if m.notifyPod[podName] != nil {
-		m.notifyPod[podName] <- vsc
-	}
-	m.notifyPodMapLock.Unlock()
-
-	return nil
-}
-
 // onPodEvent interpretes Kubernetes events, translates and broadcasts them, and acts based on them
 func (m *Monitor) onPodEvent(evt watch.Event) error {
 	// Beware: we patch running pods to add annotations. At the moment this is not a problem as do not attach
@@ -1150,31 +1100,65 @@ func (m *Monitor) finalizeWorkspaceContent(ctx context.Context, wso *workspaceOb
 				volumeSnapshotTime = time.Now()
 			}
 			if createdVolumeSnapshot {
-				m.notifyPodMapLock.Lock()
-				if m.notifyPod[wso.Pod.Name] == nil {
-					m.notifyPod[wso.Pod.Name] = make(chan string)
-				}
-				m.notifyPodMapLock.Unlock()
-
-				select {
-				case pvcVolumeSnapshotContentName = <-m.notifyPod[wso.Pod.Name]:
+				log = log.WithField("VolumeSnapshot.Name", pvcVolumeSnapshotName)
+
+				var volumeSnapshotWatcher *watchtools.RetryWatcher
+				volumeSnapshotWatcher, err = watchtools.NewRetryWatcher("1", &cache.ListWatch{
+					WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+						return m.manager.VolumeSnapshotClient.SnapshotV1().VolumeSnapshots(m.manager.Config.Namespace).Watch(ctx, metav1.ListOptions{
+							FieldSelector: fields.OneTermEqualSelector("metadata.name", pvcVolumeSnapshotName).String(),
+						})
+					},
+				})
+				if err != nil {
+					log.WithError(err).Info("fall back to exponential backoff retry")
+					// we can not create a retry watcher, we fall back to exponential backoff retry
+					backoff := wait.Backoff{
+						Steps:    30,
+						Duration: 100 * time.Millisecond,
+						Factor:   1.5,
+						Jitter:   0.1,
+						Cap:      10 * time.Minute,
+					}
+					err = wait.ExponentialBackoff(backoff, func() (bool, error) {
+						var vs volumesnapshotv1.VolumeSnapshot
+						err := m.manager.Clientset.Get(ctx, types.NamespacedName{Namespace: m.manager.Config.Namespace, Name: pvcVolumeSnapshotName}, &vs)
+						if err != nil {
+							if k8serr.IsNotFound(err) {
+								// volumesnapshot doesn't exist yet, retry again
+								return false, nil
+							}
+							log.WithError(err).Error("was unable to get volume snapshot")
+							return false, err
+						}
+						if vs.Status != nil && vs.Status.ReadyToUse != nil && *vs.Status.ReadyToUse && vs.Status.BoundVolumeSnapshotContentName != nil {
+							pvcVolumeSnapshotContentName = *vs.Status.BoundVolumeSnapshotContentName
+							return true, nil
+						}
+						return false, nil
+					})
+					if err != nil {
+						log.WithError(err).Errorf("failed while waiting for volume snapshot to get ready")
+						return nil, err
+					}
 					readyVolumeSnapshot = true
-				case <-ctx.Done():
-					// There might be a chance that the VolumeSnapshot is ready but somehow
-					// we did not receive the notification.
-					// For example, the ws-manager restarts before the VolumeSnapshot becomes ready.
-					// Let's give it the last chance to check the VolumeSnapshot is ready.
-					var vs volumesnapshotv1.VolumeSnapshot
-					err := m.manager.Clientset.Get(ctx, types.NamespacedName{Namespace: m.manager.Config.Namespace, Name: pvcVolumeSnapshotName}, &vs)
-					if err == nil && vs.Status != nil && vs.Status.ReadyToUse != nil && *vs.Status.ReadyToUse && vs.Status.BoundVolumeSnapshotContentName != nil {
-						pvcVolumeSnapshotContentName = *vs.Status.BoundVolumeSnapshotContentName
-						readyVolumeSnapshot = true
-						break
+				} else {
+					for event := range volumeSnapshotWatcher.ResultChan() {
+						vs, ok := event.Object.(*volumesnapshotv1.VolumeSnapshot)
+						if !ok {
+							log.Errorf("unexpected type assertion %T", event.Object)
+							continue
+						}
+
+						if vs != nil && vs.Status != nil && vs.Status.ReadyToUse != nil && *vs.Status.ReadyToUse && vs.Status.BoundVolumeSnapshotContentName != nil {
+							pvcVolumeSnapshotContentName = *vs.Status.BoundVolumeSnapshotContentName
+							readyVolumeSnapshot = true
+							break
+						}
 					}
 
-					err = xerrors.Errorf("%s timed out while waiting for volume snapshot to get ready", m.manager.Config.Timeouts.ContentFinalization.String())
-					log.Error(err.Error())
-					return nil, err
+					// stop the volume snapshot retry watcher
+					volumeSnapshotWatcher.Stop()
 				}
 
 				hist, err := m.manager.metrics.volumeSnapshotTimeHistVec.GetMetricWithLabelValues(wsType, wso.Pod.Labels[workspaceClassLabel])

components/ws-manager/pkg/manager/testing_test.go

@@ -22,6 +22,8 @@ import (
 	"github.com/gitpod-io/gitpod/content-service/pkg/storage"
 	"github.com/gitpod-io/gitpod/ws-manager/api"
 	config "github.com/gitpod-io/gitpod/ws-manager/api/config"
+
+	volumesnapshotclientv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
 )
 
 // This file contains test infrastructure for this package. No function in here is meant for consumption outside of tests.
@@ -96,6 +98,12 @@ func forTestingOnlyGetManager(t *testing.T, objects ...client.Object) *Manager {
 		return nil
 	}
 
+	volumesnapshotclientset, err := volumesnapshotclientv1.NewForConfig(cfg)
+	if err != nil {
+		t.Errorf("cannt create test environment: %v", err)
+		return nil
+	}
+
 	ctrlClient, err := client.New(cfg, client.Options{Scheme: scheme})
 	if err != nil {
 		t.Errorf("cannot create test environment: %v", err)
@@ -122,7 +130,7 @@ func forTestingOnlyGetManager(t *testing.T, objects ...client.Object) *Manager {
 		}
 	}
 
-	m, err := New(config, ctrlClient, clientset, &layer.Provider{Storage: &storage.PresignedNoopStorage{}})
+	m, err := New(config, ctrlClient, clientset, volumesnapshotclientset, &layer.Provider{Storage: &storage.PresignedNoopStorage{}})
 	if err != nil {
 		t.Fatalf("cannot create manager: %s", err.Error())
 	}