chore: best-effort cleanup socket

Build a cancellable context and propagate it or the context's Done()
channel to the various concurrent processes.

Author: tam7t

cmd/secrets-store-csi-driver/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"time"
 
@@ -105,31 +106,31 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
-	stopCh := ctrl.SetupSignalHandler()
+	ctx := withShutdownSignal(context.Background())
+
 	go func() {
 		klog.Infof("starting manager")
-		if err := mgr.Start(stopCh); err != nil {
+		if err := mgr.Start(ctx.Done()); err != nil {
 			klog.Fatalf("failed to run manager, error: %+v", err)
 		}
 	}()
 
 	go func() {
-		reconciler.RunPatcher(stopCh)
+		reconciler.RunPatcher(ctx)
 	}()
 
 	if *enableSecretRotation {
 		rec, err := rotation.NewReconciler(scheme, *providerVolumePath, *nodeID, *rotationPollInterval)
 		if err != nil {
 			klog.Fatalf("failed to initialize rotation reconciler, error: %+v", err)
 		}
-		stopCh := make(<-chan struct{})
-		go rec.Run(stopCh)
+		go rec.Run(ctx.Done())
 	}
 
-	handle()
+	handle(ctx)
 }
 
-func handle() {
+func handle(ctx context.Context) {
 	driver := secretsstore.GetDriver()
 	cfg, err := config.GetConfig()
 	if err != nil {
@@ -139,5 +140,19 @@ func handle() {
 	if err != nil {
 		klog.Fatalf("failed to initialize driver, error creating client: %+v", err)
 	}
-	driver.Run(*driverName, *nodeID, *endpoint, *providerVolumePath, *minProviderVersion, *grpcSupportedProviders, c)
+	driver.Run(ctx, *driverName, *nodeID, *endpoint, *providerVolumePath, *minProviderVersion, *grpcSupportedProviders, c)
+}
+
+// withShutdownSignal returns a copy of the parent context that will close if
+// the process receives termination signals.
+func withShutdownSignal(ctx context.Context) context.Context {
+	nctx, cancel := context.WithCancel(ctx)
+	stopCh := ctrl.SetupSignalHandler()
+
+	go func() {
+		<-stopCh
+		klog.Info("received shutdown signal")
+		cancel()
+	}()
+	return nctx
 }

controllers/secretproviderclasspodstatus_controller.go

@@ -85,28 +85,27 @@ func New(mgr manager.Manager, nodeID string) (*SecretProviderClassPodStatusRecon
 	}, nil
 }
 
-func (r *SecretProviderClassPodStatusReconciler) RunPatcher(stopCh <-chan struct{}) {
+func (r *SecretProviderClassPodStatusReconciler) RunPatcher(ctx context.Context) {
 	ticker := time.NewTicker(5 * time.Second)
 	defer ticker.Stop()
 
 	for {
 		select {
-		case <-stopCh:
+		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			if err := r.Patcher(); err != nil {
+			if err := r.Patcher(ctx); err != nil {
 				klog.ErrorS(err, "failed to patch secret owner ref")
 			}
 		}
 	}
 }
 
-func (r *SecretProviderClassPodStatusReconciler) Patcher() error {
+func (r *SecretProviderClassPodStatusReconciler) Patcher(ctx context.Context) error {
 	klog.V(5).Infof("patcher started")
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
-	ctx := context.Background()
 	spcPodStatusList := &v1alpha1.SecretProviderClassPodStatusList{}
 	spcMap := make(map[string]v1alpha1.SecretProviderClass)
 	secretOwnerMap := make(map[types.NamespacedName][]*v1alpha1.SecretProviderClassPodStatus)

pkg/csi-common/server.go

@@ -17,6 +17,7 @@ limitations under the License.
 package csicommon
 
 import (
+	"context"
 	"net"
 	"os"
 	"runtime"
@@ -32,7 +33,7 @@ import (
 // Defines Non blocking GRPC server interfaces
 type NonBlockingGRPCServer interface {
 	// Start services at the endpoint
-	Start(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer)
+	Start(ctx context.Context, endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer)
 	// Waits for the service to stop
 	Wait()
 	// Stops the service gracefully
@@ -51,9 +52,9 @@ type nonBlockingGRPCServer struct {
 	server *grpc.Server
 }
 
-func (s *nonBlockingGRPCServer) Start(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
+func (s *nonBlockingGRPCServer) Start(ctx context.Context, endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
 	s.wg.Add(1)
-	go s.serve(endpoint, ids, cs, ns)
+	go s.serve(ctx, endpoint, ids, cs, ns)
 }
 
 func (s *nonBlockingGRPCServer) Wait() {
@@ -68,7 +69,7 @@ func (s *nonBlockingGRPCServer) ForceStop() {
 	s.server.Stop()
 }
 
-func (s *nonBlockingGRPCServer) serve(endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
+func (s *nonBlockingGRPCServer) serve(ctx context.Context, endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
 	proto, addr, err := ParseEndpoint(endpoint)
 	if err != nil {
 		klog.Fatal(err.Error())
@@ -87,6 +88,7 @@ func (s *nonBlockingGRPCServer) serve(endpoint string, ids csi.IdentityServer, c
 	if err != nil {
 		klog.Fatalf("Failed to listen: %v", err)
 	}
+	defer listener.Close()
 
 	opts := []grpc.ServerOption{
 		grpc.UnaryInterceptor(logGRPC),
@@ -106,8 +108,14 @@ func (s *nonBlockingGRPCServer) serve(endpoint string, ids csi.IdentityServer, c
 
 	klog.Infof("Listening for connections on address: %v", listener.Addr())
 
-	err = server.Serve(listener)
-	if err != nil {
-		klog.Fatalf("Failed to serve: %v", err)
-	}
+	go func() {
+		err = server.Serve(listener)
+		if err != nil {
+			klog.Errorf("Failed to serve: %v", err)
+		}
+	}()
+
+	<-ctx.Done()
+	server.GracefulStop()
+	s.wg.Done()
 }

pkg/csi-common/utils.go

@@ -75,23 +75,23 @@ func RunNodePublishServer(endpoint string, d *CSIDriver, ns csi.NodeServer) {
 	ids := NewDefaultIdentityServer(d)
 
 	s := NewNonBlockingGRPCServer()
-	s.Start(endpoint, ids, nil, ns)
+	s.Start(context.Background(), endpoint, ids, nil, ns)
 	s.Wait()
 }
 
 func RunControllerPublishServer(endpoint string, d *CSIDriver, cs csi.ControllerServer) {
 	ids := NewDefaultIdentityServer(d)
 
 	s := NewNonBlockingGRPCServer()
-	s.Start(endpoint, ids, cs, nil)
+	s.Start(context.Background(), endpoint, ids, cs, nil)
 	s.Wait()
 }
 
 func RunControllerandNodePublishServer(endpoint string, d *CSIDriver, cs csi.ControllerServer, ns csi.NodeServer) {
 	ids := NewDefaultIdentityServer(d)
 
 	s := NewNonBlockingGRPCServer()
-	s.Start(endpoint, ids, cs, ns)
+	s.Start(context.Background(), endpoint, ids, cs, ns)
 	s.Wait()
 }
 

pkg/secrets-store/secrets-store.go

@@ -17,6 +17,7 @@ limitations under the License.
 package secretsstore
 
 import (
+	"context"
 	"strings"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -93,7 +94,7 @@ func newIdentityServer(d *csicommon.CSIDriver) *identityServer {
 }
 
 // Run starts the CSI plugin
-func (s *SecretsStore) Run(driverName, nodeID, endpoint, providerVolumePath, minProviderVersions, grpcSupportedProviders string, client client.Client) {
+func (s *SecretsStore) Run(ctx context.Context, driverName, nodeID, endpoint, providerVolumePath, minProviderVersions, grpcSupportedProviders string, client client.Client) {
 	klog.Infof("Driver: %v ", driverName)
 	klog.Infof("Version: %s", vendorVersion)
 	klog.Infof("Provider Volume Path: %s", providerVolumePath)
@@ -123,6 +124,6 @@ func (s *SecretsStore) Run(driverName, nodeID, endpoint, providerVolumePath, min
 	s.ids = newIdentityServer(s.driver)
 
 	server := csicommon.NewNonBlockingGRPCServer()
-	server.Start(endpoint, s.ids, s.cs, s.ns)
+	server.Start(ctx, endpoint, s.ids, s.cs, s.ns)
 	server.Wait()
 }

test/sanity/sanity_test.go

@@ -14,6 +14,7 @@ limitations under the License.
 package sanity
 
 import (
+	"context"
 	"testing"
 
 	"github.com/kubernetes-csi/csi-test/pkg/sanity"
@@ -32,7 +33,7 @@ const (
 func TestSanity(t *testing.T) {
 	driver := secretsstore.GetDriver()
 	go func() {
-		driver.Run("secrets-store.csi.k8s.io", "somenodeid", endpoint, providerVolumePath, "provider1=0.0.2,provider2=0.0.4", "", nil)
+		driver.Run(context.Background(), "secrets-store.csi.k8s.io", "somenodeid", endpoint, providerVolumePath, "provider1=0.0.2,provider2=0.0.4", "", nil)
 	}()
 
 	config := &sanity.Config{