Implement graceful shutdown

This change implements graceful shutdown so that a proper driver
termination leads to an exit code of zero.

Specifically, we handle signals, and let the HTTP and gRPC servers
finish requests gracefully.

We also remove the Stop() function in favor a context-based approach.

Author: Timo Reimann

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## unreleased
 
+* Implement graceful shutdown
+  [[GH-238]](https://github.com/digitalocean/csi-digitalocean/pull/238)
 * Reject requests for block access type
   [[GH-225]](https://github.com/digitalocean/csi-digitalocean/pull/225)
 * Assume detached state on 404 during ControllerUnpublishVolume

cmd/do-csi-plugin/main.go

@@ -17,10 +17,13 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"log"
 	"os"
+	"os/signal"
+	"syscall"
 
 	"github.com/digitalocean/csi-digitalocean/driver"
 )
@@ -47,7 +50,17 @@ func main() {
 		log.Fatalln(err)
 	}
 
-	if err := drv.Run(); err != nil {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	c := make(chan os.Signal, 1)
+	signal.Notify(c, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		<-c
+		cancel()
+	}()
+
+	if err := drv.Run(ctx); err != nil {
 		log.Fatalln(err)
 	}
 }

driver/driver.go

@@ -164,7 +164,7 @@ func NewDriver(ep, token, url, doTag, driverName, address string) (*Driver, erro
 }
 
 // Run starts the CSI plugin by communication over the given endpoint
-func (d *Driver) Run() error {
+func (d *Driver) Run(ctx context.Context) error {
 	u, err := url.Parse(d.endpoint)
 	if err != nil {
 		return fmt.Errorf("unable to parse address: %q", err)
@@ -240,25 +240,31 @@ func (d *Driver) Run() error {
 
 	var eg errgroup.Group
 	eg.Go(func() error {
+		<-ctx.Done()
+		return d.httpSrv.Shutdown(context.Background())
+	})
+	eg.Go(func() error {
+		go func() {
+			<-ctx.Done()
+			d.log.Info("server stopped")
+			d.readyMu.Lock()
+			d.ready = false
+			d.readyMu.Unlock()
+			d.srv.GracefulStop()
+		}()
 		return d.srv.Serve(grpcListener)
 	})
 	eg.Go(func() error {
-		return d.httpSrv.Serve(httpListener)
+		err := d.httpSrv.Serve(httpListener)
+		if err == http.ErrServerClosed {
+			return nil
+		}
+		return err
 	})
 
 	return eg.Wait()
 }
 
-// Stop stops the plugin
-func (d *Driver) Stop() {
-	d.readyMu.Lock()
-	d.ready = false
-	d.readyMu.Unlock()
-
-	d.log.Info("server stopped")
-	d.srv.Stop()
-}
-
 // When building any packages that import version, pass the build/install cmd
 // ldflags like so:
 //   go build -ldflags "-X github.com/digitalocean/csi-digitalocean/driver.version=0.0.1"

driver/driver_test.go

@@ -81,9 +81,16 @@ func TestDriverSuite(t *testing.T) {
 		account: &fakeAccountDriver{},
 		tags:    &fakeTagsDriver{},
 	}
-	defer driver.Stop()
 
-	go driver.Run()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go func() {
+		err := driver.Run(ctx)
+		if err != nil {
+			t.Error(err)
+		}
+	}()
 
 	cfg := &sanity.Config{
 		TargetPath:  os.TempDir() + "/csi-target",