refactor: identity server does not need version passed, get rid of serve

move ParseEndpoint into util pkg

Author: hrak

pkg/driver/driver.go

@@ -6,12 +6,15 @@ package driver
 import (
 	"context"
 	"fmt"
+	"net"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"google.golang.org/grpc"
 	"k8s.io/klog/v2"
 
 	"github.com/leaseweb/cloudstack-csi-driver/pkg/cloud"
 	"github.com/leaseweb/cloudstack-csi-driver/pkg/mount"
+	"github.com/leaseweb/cloudstack-csi-driver/pkg/util"
 )
 
 // Interface is the CloudStack CSI driver interface.
@@ -22,7 +25,6 @@ type Interface interface {
 
 type cloudstackDriver struct {
 	controller csi.ControllerServer
-	identity   csi.IdentityServer
 	node       csi.NodeServer
 	options    *Options
 }
@@ -40,7 +42,6 @@ func New(ctx context.Context, csConnector cloud.Interface, options *Options, mou
 		options: options,
 	}
 
-	driver.identity = NewIdentityServer(driverVersion)
 	switch options.Mode {
 	case ControllerMode:
 		driver.controller = NewControllerServer(csConnector)
@@ -57,7 +58,46 @@ func New(ctx context.Context, csConnector cloud.Interface, options *Options, mou
 }
 
 func (cs *cloudstackDriver) Run(ctx context.Context) error {
-	return cs.serve(ctx)
+	logger := klog.FromContext(ctx)
+	scheme, addr, err := util.ParseEndpoint(cs.options.Endpoint)
+	if err != nil {
+		return err
+	}
+
+	listener, err := net.Listen(scheme, addr)
+	if err != nil {
+		return fmt.Errorf("failed to listen: %w", err)
+	}
+
+	// Log every request and payloads (request + response)
+	opts := []grpc.ServerOption{
+		grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+			resp, err := handler(klog.NewContext(ctx, logger), req)
+			if err != nil {
+				logger.Error(err, "GRPC method failed", "method", info.FullMethod)
+			}
+
+			return resp, err
+		}),
+	}
+	grpcServer := grpc.NewServer(opts...)
+
+	csi.RegisterIdentityServer(grpcServer, cs)
+	switch cs.options.Mode {
+	case ControllerMode:
+		csi.RegisterControllerServer(grpcServer, cs.controller)
+	case NodeMode:
+		csi.RegisterNodeServer(grpcServer, cs.node)
+	case AllMode:
+		csi.RegisterControllerServer(grpcServer, cs.controller)
+		csi.RegisterNodeServer(grpcServer, cs.node)
+	default:
+		return fmt.Errorf("unknown mode: %s", cs.options.Mode)
+	}
+
+	logger.Info("Listening for connections", "address", listener.Addr())
+
+	return grpcServer.Serve(listener)
 }
 
 func validateMode(mode Mode) error {

pkg/driver/identity.go

@@ -4,46 +4,28 @@ import (
 	"context"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 	"k8s.io/klog/v2"
 )
 
-type identityServer struct {
-	csi.UnimplementedIdentityServer
-	version string
-}
-
-// NewIdentityServer creates a new Identity gRPC server.
-func NewIdentityServer(version string) csi.IdentityServer {
-	return &identityServer{
-		version: version,
-	}
-}
-
-func (ids *identityServer) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
+func (cs *cloudstackDriver) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) {
 	logger := klog.FromContext(ctx)
 	logger.V(6).Info("GetPluginInfo: called", "args", *req)
-	if ids.version == "" {
-		return nil, status.Error(codes.Unavailable, "Driver is missing version")
-	}
-
 	resp := &csi.GetPluginInfoResponse{
 		Name:          DriverName,
-		VendorVersion: ids.version,
+		VendorVersion: driverVersion,
 	}
 
 	return resp, nil
 }
 
-func (ids *identityServer) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
+func (cs *cloudstackDriver) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) {
 	logger := klog.FromContext(ctx)
 	logger.V(6).Info("Probe: called", "args", *req)
 
 	return &csi.ProbeResponse{}, nil
 }
 
-func (ids *identityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
+func (cs *cloudstackDriver) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
 	logger := klog.FromContext(ctx)
 	logger.V(6).Info("Probe: called", "args", *req)
 

pkg/driver/server.go

@@ -0,0 +1,34 @@
+package util
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// ParseEndpoint parses the CSI socket endpoint and returns the components scheme and addr.
+func ParseEndpoint(endpoint string) (string, string, error) {
+	u, err := url.Parse(endpoint)
+	if err != nil {
+		return "", "", fmt.Errorf("could not parse endpoint: %w", err)
+	}
+
+	addr := filepath.Join(u.Host, filepath.FromSlash(u.Path))
+
+	scheme := strings.ToLower(u.Scheme)
+	switch scheme {
+	case "tcp":
+	case "unix":
+		addr = filepath.Join("/", addr)
+		// Remove the socket file if it already exists.
+		if err := os.Remove(addr); err != nil && !os.IsNotExist(err) {
+			return "", "", fmt.Errorf("could not remove unix domain socket %q: %w", addr, err)
+		}
+	default:
+		return "", "", fmt.Errorf("unsupported protocol: %s", scheme)
+	}
+
+	return scheme, addr, nil
+}

pkg/util/endpoint.go

@@ -0,0 +1,75 @@
+package util
+
+import (
+	"errors"
+	"testing"
+)
+
+func TestParseEndpoint(t *testing.T) {
+	testCases := []struct {
+		name      string
+		endpoint  string
+		expScheme string
+		expAddr   string
+		expErr    error
+	}{
+		{
+			name:      "valid unix endpoint 1",
+			endpoint:  "unix:///csi/csi.sock",
+			expScheme: "unix",
+			expAddr:   "/csi/csi.sock",
+		},
+		{
+			name:      "valid unix endpoint 2",
+			endpoint:  "unix://csi/csi.sock",
+			expScheme: "unix",
+			expAddr:   "/csi/csi.sock",
+		},
+		{
+			name:      "valid unix endpoint 3",
+			endpoint:  "unix:/csi/csi.sock",
+			expScheme: "unix",
+			expAddr:   "/csi/csi.sock",
+		},
+		{
+			name:      "valid tcp endpoint",
+			endpoint:  "tcp:///127.0.0.1/",
+			expScheme: "tcp",
+			expAddr:   "/127.0.0.1",
+		},
+		{
+			name:      "valid tcp endpoint",
+			endpoint:  "tcp:///127.0.0.1",
+			expScheme: "tcp",
+			expAddr:   "/127.0.0.1",
+		},
+		{
+			name:     "invalid endpoint",
+			endpoint: "http://127.0.0.1",
+			expErr:   errors.New("unsupported protocol: http"),
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			scheme, addr, err := ParseEndpoint(tc.endpoint)
+
+			if tc.expErr != nil {
+				if err.Error() != tc.expErr.Error() {
+					t.Fatalf("Expecting err: expected %v, got %v", tc.expErr, err)
+				}
+			} else {
+				if err != nil {
+					t.Fatalf("err is not nil. got: %v", err)
+				}
+				if scheme != tc.expScheme {
+					t.Fatalf("scheme mismatches: expected %v, got %v", tc.expScheme, scheme)
+				}
+
+				if addr != tc.expAddr {
+					t.Fatalf("addr mismatches: expected %v, got %v", tc.expAddr, addr)
+				}
+			}
+		})
+	}
+}