Merge pull request #394 from tam7t/reusable-connection

feat: Reuse provider grpc clients

Author: k8s-ci-robot

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

@@ -19,6 +19,7 @@ package main
 import (
 	"context"
 	"flag"
+	"strings"
 	"time"
 
 	"sigs.k8s.io/secrets-store-csi-driver/pkg/metrics"
@@ -85,10 +86,10 @@ func main() {
 		klog.Fatalf("failed to initialize metrics exporter, error: %+v", err)
 	}
 
-	config := ctrl.GetConfigOrDie()
-	config.UserAgent = "csi-secrets-store/controller"
+	cfg := ctrl.GetConfigOrDie()
+	cfg.UserAgent = "csi-secrets-store/controller"
 
-	mgr, err := ctrl.NewManager(config, ctrl.Options{
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
 		Scheme:             scheme,
 		MetricsBindAddress: *metricsAddr,
 		LeaderElection:     false,
@@ -108,6 +109,26 @@ func main() {
 
 	ctx := withShutdownSignal(context.Background())
 
+	// create provider clients
+	providerClients := make(map[string]*secretsstore.CSIProviderClient)
+	for _, provider := range strings.Split(*grpcSupportedProviders, ";") {
+		p := strings.TrimSpace(provider)
+		if len(p) != 0 {
+			// dialing clients is non-blocking and will be retried on errors
+			providerClients[provider], err = secretsstore.NewProviderClient(secretsstore.CSIProviderName(p), *providerVolumePath)
+			if err != nil {
+				klog.Fatalf("failed to create provider client, err: %+v", err)
+			}
+		}
+	}
+	defer func() {
+		for k, v := range providerClients {
+			if err := v.Close(); err != nil {
+				klog.ErrorS(err, "closing grpc client failed", "provider", k)
+			}
+		}
+	}()
+
 	go func() {
 		klog.Infof("starting manager")
 		if err := mgr.Start(ctx.Done()); err != nil {
@@ -120,27 +141,23 @@ func main() {
 	}()
 
 	if *enableSecretRotation {
-		rec, err := rotation.NewReconciler(scheme, *providerVolumePath, *nodeID, *rotationPollInterval)
+		rec, err := rotation.NewReconciler(scheme, *providerVolumePath, *nodeID, *rotationPollInterval, providerClients)
 		if err != nil {
 			klog.Fatalf("failed to initialize rotation reconciler, error: %+v", err)
 		}
 		go rec.Run(ctx.Done())
 	}
 
-	handle(ctx)
-}
-
-func handle(ctx context.Context) {
-	driver := secretsstore.GetDriver()
-	cfg, err := config.GetConfig()
+	ccfg, err := config.GetConfig()
 	if err != nil {
 		klog.Fatalf("failed to initialize driver, error getting config: %+v", err)
 	}
-	c, err := client.New(cfg, client.Options{Scheme: scheme, Mapper: nil})
+	c, err := client.New(ccfg, client.Options{Scheme: scheme, Mapper: nil})
 	if err != nil {
 		klog.Fatalf("failed to initialize driver, error creating client: %+v", err)
 	}
-	driver.Run(ctx, *driverName, *nodeID, *endpoint, *providerVolumePath, *minProviderVersion, *grpcSupportedProviders, c)
+	driver := secretsstore.GetDriver()
+	driver.Run(ctx, *driverName, *nodeID, *endpoint, *providerVolumePath, *minProviderVersion, providerClients, c)
 }
 
 // withShutdownSignal returns a copy of the parent context that will close if

pkg/csi-common/server.go

@@ -70,6 +70,8 @@ func (s *nonBlockingGRPCServer) ForceStop() {
 }
 
 func (s *nonBlockingGRPCServer) serve(ctx context.Context, endpoint string, ids csi.IdentityServer, cs csi.ControllerServer, ns csi.NodeServer) {
+	defer s.wg.Done()
+
 	proto, addr, err := ParseEndpoint(endpoint)
 	if err != nil {
 		klog.Fatal(err.Error())
@@ -117,5 +119,4 @@ func (s *nonBlockingGRPCServer) serve(ctx context.Context, endpoint string, ids
 
 	<-ctx.Done()
 	server.GracefulStop()
-	s.wg.Done()
 }

pkg/errors/errors.go

@@ -29,8 +29,8 @@ const (
 	FailedToMount = "FailedToMount"
 	// SecretProviderClassNotFound error
 	SecretProviderClassNotFound = "SecretProviderClassNotFound"
-	// FailedToCreateProviderGRPCClient error
-	FailedToCreateProviderGRPCClient = "FailedToCreateProviderGRPCClient"
+	// FailedToLookupProviderGRPCClient error
+	FailedToLookupProviderGRPCClient = "FailedToLookupProviderGRPCClient"
 	// GRPCProviderError error
 	GRPCProviderError = "GRPCProviderError"
 	// FailedToRotate error

pkg/rotation/reconciler.go

@@ -83,7 +83,7 @@ type Reconciler struct {
 }
 
 // NewReconciler returns a new reconciler for rotation
-func NewReconciler(s *runtime.Scheme, providerVolumePath, nodeName string, rotationPollInterval time.Duration) (*Reconciler, error) {
+func NewReconciler(s *runtime.Scheme, providerVolumePath, nodeName string, rotationPollInterval time.Duration, providerClients map[string]*secretsstore.CSIProviderClient) (*Reconciler, error) {
 	config, err := buildConfig()
 	if err != nil {
 		return nil, err
@@ -109,7 +109,7 @@ func NewReconciler(s *runtime.Scheme, providerVolumePath, nodeName string, rotat
 		scheme:               s,
 		providerVolumePath:   providerVolumePath,
 		rotationPollInterval: rotationPollInterval,
-		providerClients:      make(map[string]*secretsstore.CSIProviderClient),
+		providerClients:      providerClients,
 		reporter:             newStatsReporter(),
 		queue:                workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		eventRecorder:        recorder,
@@ -274,11 +274,11 @@ func (r *Reconciler) reconcile(ctx context.Context, spcps *v1alpha1.SecretProvid
 	}
 
 	providerName = string(spc.Spec.Provider)
-	providerClient, err := r.getProviderClient(providerName)
-	if err != nil {
-		errorReason = internalerrors.FailedToCreateProviderGRPCClient
-		r.generateEvent(pod, v1.EventTypeWarning, mountRotationFailedReason, fmt.Sprintf("failed to create provider client, err: %+v", err))
-		return fmt.Errorf("failed to create provider client, err: %+v", err)
+	providerClient, exists := r.providerClients[providerName]
+	if !exists {
+		errorReason = internalerrors.FailedToLookupProviderGRPCClient
+		r.generateEvent(pod, v1.EventTypeWarning, mountRotationFailedReason, fmt.Sprintf("failed to lookup provider client: %q", providerName))
+		return fmt.Errorf("failed to lookup provider client: %q", providerName)
 	}
 	newObjectVersions, errorReason, err := providerClient.MountContent(ctx, string(paramsJSON), string(secretsJSON), spcps.Status.TargetPath, string(permissionJSON), oldObjectVersions)
 	if err != nil {
@@ -446,21 +446,6 @@ func (r *Reconciler) patchSecret(ctx context.Context, name, namespace string, da
 	return r.ctrlWriterClient.Patch(ctx, secret, patch)
 }
 
-// getProviderClient returns the GRPC provider client to use for mount request
-func (r *Reconciler) getProviderClient(providerName string) (*secretsstore.CSIProviderClient, error) {
-	// check if the provider client already exists
-	if providerClient, exists := r.providerClients[providerName]; exists {
-		return providerClient, nil
-	}
-	// create a new client as it doesn't exist in the reconciler cache
-	providerClient, err := secretsstore.NewProviderClient(secretsstore.CSIProviderName(providerName), r.providerVolumePath)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create %s provider client, err: %+v", providerName, err)
-	}
-	r.providerClients[providerName] = providerClient
-	return providerClient, nil
-}
-
 // runWorker runs a thread that process the queue
 func (r *Reconciler) runWorker() {
 	for r.processNextItem() {

pkg/rotation/reconciler_test.go

@@ -75,17 +75,24 @@ func newTestReconciler(s *runtime.Scheme, kubeClient kubernetes.Interface, crdCl
 		return nil, err
 	}
 
+	client, err := secretsstore.NewProviderClient("provider1", socketPath)
+	if err != nil {
+		return nil, err
+	}
+
 	return &Reconciler{
 		store:                store,
 		ctrlReaderClient:     ctrlClient,
 		ctrlWriterClient:     ctrlClient,
 		scheme:               s,
 		providerVolumePath:   socketPath,
 		rotationPollInterval: rotationPollInterval,
-		providerClients:      map[string]*secretsstore.CSIProviderClient{},
-		queue:                workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
-		reporter:             newStatsReporter(),
-		eventRecorder:        fakeRecorder,
+		providerClients: map[string]*secretsstore.CSIProviderClient{
+			"provider1": client,
+		},
+		queue:         workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		reporter:      newStatsReporter(),
+		eventRecorder: fakeRecorder,
 	}, nil
 }
 
@@ -370,7 +377,7 @@ func TestReconcileError(t *testing.T) {
 			expectedErrorEvents:    false,
 		},
 		{
-			name:                 "failed to create provider client",
+			name:                 "failed to lookup provider client",
 			rotationPollInterval: 60 * time.Second,
 			secretProviderClassPodStatusToProcess: &v1alpha1.SecretProviderClassPodStatus{
 				ObjectMeta: metav1.ObjectMeta{

pkg/secrets-store/nodeserver.go

@@ -41,13 +41,13 @@ import (
 
 type nodeServer struct {
 	*csicommon.DefaultNodeServer
-	providerVolumePath     string
-	minProviderVersions    map[string]string
-	mounter                mount.Interface
-	reporter               StatsReporter
-	nodeID                 string
-	client                 client.Client
-	grpcSupportedProviders map[string]bool
+	providerVolumePath  string
+	minProviderVersions map[string]string
+	mounter             mount.Interface
+	reporter            StatsReporter
+	nodeID              string
+	client              client.Client
+	providerClients     map[string]*CSIProviderClient
 }
 
 const (
@@ -296,14 +296,9 @@ func (ns *nodeServer) mountSecretsStoreObjectContent(ctx context.Context, provid
 	// if the provider supports and is running grpc server, then communicate with
 	// provider using the grpc client, otherwise fallback to invoking the provider
 	// binary which is how it was initially implemented
-	_, exists := ns.grpcSupportedProviders[providerName]
-	if exists {
+	if client, exists := ns.providerClients[providerName]; exists {
 		klog.InfoS("Using grpc client", "provider", providerName, "pod", podName)
-		providerClient, err := NewProviderClient(CSIProviderName(providerName), ns.providerVolumePath)
-		if err != nil {
-			return nil, internalerrors.FailedToCreateProviderGRPCClient, fmt.Errorf("failed to create provider client, err: %+v", err)
-		}
-		return providerClient.MountContent(ctx, attributes, secrets, targetPath, permission, nil)
+		return client.MountContent(ctx, attributes, secrets, targetPath, permission, nil)
 	}
 
 	providerBinary := ns.getProviderPath(runtime.GOOS, providerName)

pkg/secrets-store/nodeserver_test.go

@@ -40,16 +40,18 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
-func testNodeServer(mountPoints []mount.MountPoint, client client.Client, grpcSupportProviders string, reporter StatsReporter, providerBinaryName string) (*nodeServer, error) {
+func testNodeServer(mountPoints []mount.MountPoint, client client.Client, grpcSupportProvider string, reporter StatsReporter, providerBinaryName string) (*nodeServer, error) {
 	tmpDir, err := ioutil.TempDir("", "ut")
 	if err != nil {
 		return nil, err
 	}
-	if grpcSupportProviders != "" {
-		err = ioutil.WriteFile(fmt.Sprintf("%s/%s.sock", tmpDir, grpcSupportProviders), nil, permission)
+	providerClients := map[string]*CSIProviderClient{}
+	if grpcSupportProvider != "" {
+		client, err := NewProviderClient(CSIProviderName(grpcSupportProvider), tmpDir)
 		if err != nil {
 			return nil, err
 		}
+		providerClients[grpcSupportProvider] = client
 	}
 	if providerBinaryName != "" {
 		dirPath := fmt.Sprintf("%s/%s", tmpDir, providerBinaryName)
@@ -59,12 +61,15 @@ func testNodeServer(mountPoints []mount.MountPoint, client client.Client, grpcSu
 			return nil, err
 		}
 		f, err := os.Create(filePath)
+		if err != nil {
+			return nil, err
+		}
 		defer f.Close()
 		if err != nil {
 			return nil, err
 		}
 	}
-	return newNodeServer(NewFakeDriver(), tmpDir, "", grpcSupportProviders, "testnode", mount.NewFakeMounter(mountPoints), client, reporter)
+	return newNodeServer(NewFakeDriver(), tmpDir, "", "testnode", mount.NewFakeMounter(mountPoints), providerClients, client, reporter)
 }
 
 func getTestTargetPath(pattern string, t *testing.T) string {
@@ -402,15 +407,6 @@ func TestMountSecretsStoreObjectContent(t *testing.T) {
 			expectedErrorReason: internalerrors.ProviderBinaryNotFound,
 			expectedErr:         true,
 		},
-		{
-			name:                 "failed to create provider grpc client",
-			attributes:           "{}",
-			targetPath:           getTestTargetPath("", t),
-			permission:           fmt.Sprint(permission),
-			grpcSupportProviders: "provider1",
-			expectedErrorReason:  "GRPCProviderError",
-			expectedErr:          true,
-		},
 	}
 
 	for _, test := range tests {