connection: refactor Connect to use Option more

Extend the existing `Option` functionality for connect to be more
flexible. Configuring how `connect` behaves relies on new functions like
`withMetrics` or `withTimeout` that need to be passed as options at the
caller level. This will make it easier to add new configuration options
without breaking the api.

Author: Fricounet

connection/connection.go

@@ -56,6 +56,7 @@ func SetMaxGRPCLogLength(characterCount int) {
 //
 // The function tries to connect for 30 seconds, and returns an error if no connection has been established at that point.
 // The function automatically disables TLS and adds interceptor for logging of all gRPC messages at level 5.
+// If the metricsManager is 'nil', no metrics will be recorded on the gRPC calls.
 //
 // For a connection to a Unix Domain socket, the behavior after
 // loosing the connection is configurable. The default is to
@@ -70,12 +71,11 @@ func SetMaxGRPCLogLength(characterCount int) {
 // For other connections, the default behavior from gRPC is used and
 // loss of connection is not detected reliably.
 func Connect(address string, metricsManager metrics.CSIMetricsManager, options ...Option) (*grpc.ClientConn, error) {
-	return connect(address, metricsManager, []grpc.DialOption{grpc.WithTimeout(time.Second * 30)}, options)
-}
-
-// ConnectWithoutMetrics behaves exactly like Connect except no metrics are recorded.
-func ConnectWithoutMetrics(address string, options ...Option) (*grpc.ClientConn, error) {
-	return connect(address, nil, []grpc.DialOption{grpc.WithTimeout(time.Second * 30)}, options)
+	options = append(options, withTimeout(time.Second*30))
+	if metricsManager != nil {
+		options = append(options, withMetrics(metricsManager))
+	}
+	return connect(address, options)
 }
 
 // Option is the type of all optional parameters for Connect.
@@ -105,29 +105,48 @@ func ExitOnConnectionLoss() func() bool {
 	}
 }
 
+// withTimeout adds a configurable timeout on the gRPC calls.
+func withTimeout(timeout time.Duration) Option {
+	return func(o *options) {
+		o.timeout = timeout
+	}
+}
+
+// withMetrics enables the recording of metrics on the gRPC calls with the provided CSIMetricsManager.
+func withMetrics(metricsManager metrics.CSIMetricsManager) Option {
+	return func(o *options) {
+		o.metricsManager = metricsManager
+	}
+}
+
 type options struct {
-	reconnect func() bool
+	reconnect      func() bool
+	timeout        time.Duration
+	metricsManager metrics.CSIMetricsManager
 }
 
 // connect is the internal implementation of Connect. It has more options to enable testing.
 func connect(
 	address string,
-	metricsManager metrics.CSIMetricsManager,
-	dialOptions []grpc.DialOption, connectOptions []Option) (*grpc.ClientConn, error) {
+	connectOptions []Option) (*grpc.ClientConn, error) {
 	var o options
 	for _, option := range connectOptions {
 		option(&o)
 	}
 
-	dialOptions = append(dialOptions,
+	dialOptions := []grpc.DialOption{
 		grpc.WithInsecure(),                   // Don't use TLS, it's usually local Unix domain socket in a container.
 		grpc.WithBackoffMaxDelay(time.Second), // Retry every second after failure.
 		grpc.WithBlock(),                      // Block until connection succeeds.
-	)
+	}
+
+	if o.timeout > 0 {
+		dialOptions = append(dialOptions, grpc.WithTimeout(o.timeout))
+	}
 
 	interceptors := []grpc.UnaryClientInterceptor{LogGRPC}
-	if metricsManager != nil {
-		interceptors = append(interceptors, ExtendedCSIMetricsManager{metricsManager}.RecordMetricsClientInterceptor)
+	if o.metricsManager != nil {
+		interceptors = append(interceptors, ExtendedCSIMetricsManager{o.metricsManager}.RecordMetricsClientInterceptor)
 	}
 	dialOptions = append(dialOptions, grpc.WithChainUnaryInterceptor(interceptors...))
 

connection/connection_test.go

@@ -138,7 +138,7 @@ func TestConnectWithoutMetrics(t *testing.T) {
 	addr, stopServer := startServer(t, tmp, nil, nil, nil)
 	defer stopServer()
 
-	conn, err := ConnectWithoutMetrics("unix:///" + addr)
+	conn, err := Connect("unix:///"+addr, nil)
 	if assert.NoError(t, err, "connect with unix:/// prefix") &&
 		assert.NotNil(t, conn, "got a connection") {
 		assert.Equal(t, connectivity.Ready, conn.GetState(), "connection ready")
@@ -191,13 +191,13 @@ func TestWaitForServer(t *testing.T) {
 	}
 }
 
-func TestTimout(t *testing.T) {
+func TestTimeout(t *testing.T) {
 	tmp := tmpDir(t)
 	defer os.RemoveAll(tmp)
 
 	startTime := time.Now()
 	timeout := 5 * time.Second
-	conn, err := connect(path.Join(tmp, "no-such.sock"), metrics.NewCSIMetricsManager("fake.csi.driver.io"), []grpc.DialOption{grpc.WithTimeout(timeout)}, nil)
+	conn, err := connect(path.Join(tmp, "no-such.sock"), []Option{withTimeout(timeout)})
 	endTime := time.Now()
 	if assert.Error(t, err, "connection should fail") {
 		assert.InEpsilon(t, timeout, endTime.Sub(startTime), 1, "connection timeout")