Add start and stop volumeWatcher (libopenstorage#2320)

* Add start and stop volumeWatcher - Adds two function in Watcher interface and fix tests JIRA: PWX-32359 Signed-off-by: dahuang <[email protected]>
twang-ps · Aug 17, 2023 · 8015fc0 · 8015fc0
1 parent fa4a9b2
commit 8015fc0
Show file tree

Hide file tree

Showing 12 changed files with 85 additions and 2 deletions.
diff --git a/api/client/volume/client.go b/api/client/volume/client.go
@@ -75,10 +75,18 @@ func (v *volumeClient) GraphDriverRemove(id string) error {
  return nil
 }
 
+func (v *volumeClient) StartVolumeWatcher() {
+ return
+}
+
 func (v *volumeClient) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  return nil, nil
 }
 
+func (v *volumeClient) StopVolumeWatcher() {
+ return
+}
+
 func (v *volumeClient) GraphDriverGet(id string, mountLabel string) (string, error) {
  response := ""
  if err := v.c.Get().Resource(graphPath + "/inspect").Instance(id).Do().Unmarshal(&response); err != nil {

diff --git a/api/server/sdk/sdk_test.go b/api/server/sdk/sdk_test.go
@@ -147,6 +147,7 @@ func newTestServer(t *testing.T) *testServer {
 
  assert.Nil(t, err)
 
+ tester.m.EXPECT().StartVolumeWatcher().Return().Times(1)
  tester.m.EXPECT().GetVolumeWatcher(&api.VolumeLocator{}, make(map[string]string)).DoAndReturn(func(a *api.VolumeLocator, l map[string]string) (chan *api.Volume, error) {
  ch := make(chan *api.Volume, 1)
  tester.server.watcherCtxCancel()
@@ -233,7 +234,7 @@ func newTestServerAuth(t *testing.T) *testServer {
  },
  })
  assert.Nil(t, err)
-
+ tester.m.EXPECT().StartVolumeWatcher().Return().Times(1)
  tester.m.EXPECT().GetVolumeWatcher(&api.VolumeLocator{}, make(map[string]string)).DoAndReturn(func(a *api.VolumeLocator, l map[string]string) (chan *api.Volume, error) {
  ch := make(chan *api.Volume, 1)
  tester.server.watcherCtxCancel()
@@ -294,6 +295,7 @@ func (s *testServer) Stop() {
 
  // Shutdown servers
  s.conn.Close()
+ s.m.EXPECT().StopVolumeWatcher().Return().AnyTimes()
  s.server.Stop()
  s.gw.Close()
 

diff --git a/api/server/sdk/server.go b/api/server/sdk/server.go
@@ -314,6 +314,7 @@ func (s *Server) Stop() {
  s.netServer.Stop()
  s.udsServer.Stop()
  s.restGateway.Stop()
+ s.netServer.watcherServer.stopWatcher(s.watcherCtx)
  s.watcherCtxCancel()
 
  if s.accessLog != nil {

diff --git a/api/server/sdk/watcher.go b/api/server/sdk/watcher.go
@@ -84,6 +84,10 @@ func (s *WatcherServer) removeWatcher(name string, eventType string) {
  s.watchConnections[eventType] = newWatchers
 }
 
+func (s *WatcherServer) stopWatcher(ctx context.Context) {
+ s.volumeServer.driver(ctx).StopVolumeWatcher()
+}
+
 func (s *WatcherServer) startWatcher(ctx context.Context) error {
  group, _ := errgroup.WithContext(ctx)
  errChan := make(chan error)
@@ -121,7 +125,7 @@ func (s *WatcherServer) startVolumeWatcher(ctx context.Context) error {
  time.Sleep(2 * time.Second)
  continue
  }
-
+ s.volumeServer.driver(ctx).StartVolumeWatcher()
  volumeChannel, err := s.volumeServer.driver(ctx).GetVolumeWatcher(&api.VolumeLocator{}, make(map[string]string))
  if err != nil {
  logrus.Warnf("Error getting volume watcher %v", err)

diff --git a/csi/csi_test.go b/csi/csi_test.go
@@ -270,6 +270,7 @@ func (s *testServer) Stop() {
 
  // Shutdown servers
  s.conn.Close()
+ s.m.EXPECT().StopVolumeWatcher().Return().AnyTimes()
  s.server.Stop()
  s.sdk.Stop()
 

diff --git a/volume/drivers/buse/buse.go b/volume/drivers/buse/buse.go
@@ -146,10 +146,18 @@ func Init(params map[string]string) (volume.VolumeDriver, error) {
 // These functions below implement the volume driver interface.
 //
 
+func (d *driver) StartVolumeWatcher() {
+ return
+}
+
 func (d *driver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  return nil, nil
 }
 
+func (d *driver) StopVolumeWatcher() {
+ return
+}
+
 func (d *driver) String() string {
  return Name
 }

diff --git a/volume/drivers/fake/fake.go b/volume/drivers/fake/fake.go
@@ -142,6 +142,10 @@ func volumeGenerator(d *driver) {
  }
 }
 
+func (d *driver) StartVolumeWatcher() {
+ return
+}
+
 func (d *driver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  go volumeGenerator(d)
  if d.volumeChannel == nil {
@@ -150,6 +154,10 @@ func (d *driver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]
  return d.volumeChannel, nil
 }
 
+func (d *driver) StopVolumeWatcher() {
+ return
+}
+
 func (d *driver) Name() string {
  return Name
 }

diff --git a/volume/drivers/fuse/volume_driver.go b/volume/drivers/fuse/volume_driver.go
@@ -59,11 +59,18 @@ func newVolumeDriver(
  provider,
  }
 }
+func (v *volumeDriver) StartVolumeWatcher() {
+ return
+}
 
 func (v *volumeDriver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  return nil, nil
 }
 
+func (v *volumeDriver) StopVolumeWatcher() {
+ return
+}
+
 func (v *volumeDriver) Name() string {
  return v.name
 }

diff --git a/volume/drivers/mock/driver.mock.go b/volume/drivers/mock/driver.mock.go
diff --git a/volume/drivers/nfs/nfs.go b/volume/drivers/nfs/nfs.go
@@ -183,10 +183,18 @@ func Init(params map[string]string) (volume.VolumeDriver, error) {
  return inst, nil
 }
 
+func (d *driver) StartVolumeWatcher() {
+ return
+}
+
 func (d *driver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  return nil, nil
 }
 
+func (d *driver) StopVolumeWatcher() {
+ return
+}
+
 func (d *driver) Name() string {
  return Name
 }

diff --git a/volume/drivers/vfs/vfs.go b/volume/drivers/vfs/vfs.go
@@ -57,10 +57,18 @@ func Init(params map[string]string) (volume.VolumeDriver, error) {
  }, nil
 }
 
+func (d *driver) StartVolumeWatcher() {
+ return
+}
+
 func (d *driver) GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error) {
  return nil, nil
 }
 
+func (d *driver) StopVolumeWatcher() {
+ return
+}
+
 func (d *driver) Name() string {
  return Name
 }

diff --git a/volume/volume.go b/volume/volume.go
@@ -311,8 +311,12 @@ type Enumerator interface {
 
 // Water provides a set of function to get volume
 type Watcher interface {
+ // Stop Volume notifier
+ StartVolumeWatcher()
  // Gets Volume notifier
  GetVolumeWatcher(locator *api.VolumeLocator, labels map[string]string) (chan *api.Volume, error)
+ // Stop Volume notifier
+ StopVolumeWatcher()
 }
 
 // StoreEnumerator combines Store and Enumerator capabilities