Introduce CSI Metrics Library

Author: saad-ali

connection/connection.go

@@ -24,6 +24,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/kubernetes-csi/csi-lib-utils/metrics"
 	"github.com/kubernetes-csi/csi-lib-utils/protosanitizer"
 	"google.golang.org/grpc"
 	"k8s.io/klog"
@@ -58,8 +59,8 @@ const terminationLogPath = "/dev/termination-log"
 //
 // For other connections, the default behavior from gRPC is used and
 // loss of connection is not detected reliably.
-func Connect(address string, options ...Option) (*grpc.ClientConn, error) {
-	return connect(address, []grpc.DialOption{}, options)
+func Connect(address string, metricsManager metrics.CSIMetricsManager, options ...Option) (*grpc.ClientConn, error) {
+	return connect(address, metricsManager, []grpc.DialOption{}, options)
 }
 
 // Option is the type of all optional parameters for Connect.
@@ -93,7 +94,10 @@ type options struct {
 }
 
 // connect is the internal implementation of Connect. It has more options to enable testing.
-func connect(address string, dialOptions []grpc.DialOption, connectOptions []Option) (*grpc.ClientConn, error) {
+func connect(
+	address string,
+	metricsManager metrics.CSIMetricsManager,
+	dialOptions []grpc.DialOption, connectOptions []Option) (*grpc.ClientConn, error) {
 	var o options
 	for _, option := range connectOptions {
 		option(&o)
@@ -103,7 +107,10 @@ func connect(address string, dialOptions []grpc.DialOption, connectOptions []Opt
 		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.
-		grpc.WithUnaryInterceptor(LogGRPC),    // Log all messages.
+		grpc.WithChainUnaryInterceptor(
+			LogGRPC, // Log all messages.
+			extendedCSIMetricsManager{metricsManager}.recordMetricsInterceptor, // Record metrics for each gRPC call.
+		),
 	)
 	unixPrefix := "unix://"
 	if strings.HasPrefix(address, "/") {
@@ -179,3 +186,26 @@ func LogGRPC(ctx context.Context, method string, req, reply interface{}, cc *grp
 	klog.V(5).Infof("GRPC error: %v", err)
 	return err
 }
+
+type extendedCSIMetricsManager struct {
+	metrics.CSIMetricsManager
+}
+
+// recordMetricsInterceptor is a gPRC unary interceptor for recording metrics for CSI operations.
+func (cmm extendedCSIMetricsManager) recordMetricsInterceptor(
+	ctx context.Context,
+	method string,
+	req, reply interface{},
+	cc *grpc.ClientConn,
+	invoker grpc.UnaryInvoker,
+	opts ...grpc.CallOption) error {
+	start := time.Now()
+	err := invoker(ctx, method, req, reply, cc, opts...)
+	duration := time.Since(start)
+	cmm.RecordMetrics(
+		method,   /* operationName */
+		err,      /* operationErr */
+		duration, /* operationDuration */
+	)
+	return err
+}

connection/connection_test.go

@@ -17,11 +17,13 @@ limitations under the License.
 package connection
 
 import (
+	"bufio"
 	"context"
 	"io/ioutil"
 	"net"
 	"os"
 	"path"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -35,6 +37,9 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+
+	"github.com/kubernetes-csi/csi-lib-utils/metrics"
+	"k8s.io/component-base/metrics/testutil"
 )
 
 func tmpDir(t *testing.T) string {
@@ -84,7 +89,7 @@ func TestConnect(t *testing.T) {
 	addr, stopServer := startServer(t, tmp, nil, nil)
 	defer stopServer()
 
-	conn, err := Connect(addr)
+	conn, err := Connect(addr, metrics.NewCSIMetricsManager("fake.csi.driver.io"))
 	if assert.NoError(t, err, "connect via absolute path") &&
 		assert.NotNil(t, conn, "got a connection") {
 		assert.Equal(t, connectivity.Ready, conn.GetState(), "connection ready")
@@ -99,7 +104,7 @@ func TestConnectUnix(t *testing.T) {
 	addr, stopServer := startServer(t, tmp, nil, nil)
 	defer stopServer()
 
-	conn, err := Connect("unix:///" + addr)
+	conn, err := Connect("unix:///"+addr, metrics.NewCSIMetricsManager("fake.csi.driver.io"))
 	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")
@@ -139,7 +144,7 @@ func TestWaitForServer(t *testing.T) {
 		startTimeServer = time.Now()
 		_, stopServer = startServer(t, tmp, nil, nil)
 	}()
-	conn, err := Connect(path.Join(tmp, serverSock))
+	conn, err := Connect(path.Join(tmp, serverSock), metrics.NewCSIMetricsManager("fake.csi.driver.io"))
 	if assert.NoError(t, err, "connect via absolute path") {
 		endTime := time.Now()
 		assert.NotNil(t, conn, "got a connection")
@@ -158,7 +163,7 @@ func TestTimout(t *testing.T) {
 
 	startTime := time.Now()
 	timeout := 5 * time.Second
-	conn, err := connect(path.Join(tmp, "no-such.sock"), []grpc.DialOption{grpc.WithTimeout(timeout)}, nil)
+	conn, err := connect(path.Join(tmp, "no-such.sock"), metrics.NewCSIMetricsManager("fake.csi.driver.io"), []grpc.DialOption{grpc.WithTimeout(timeout)}, nil)
 	endTime := time.Now()
 	if assert.Error(t, err, "connection should fail") {
 		assert.InEpsilon(t, timeout, endTime.Sub(startTime), 1, "connection timeout")
@@ -177,7 +182,7 @@ func TestReconnect(t *testing.T) {
 	}()
 
 	// Allow reconnection (the default).
-	conn, err := Connect(addr)
+	conn, err := Connect(addr, metrics.NewCSIMetricsManager("fake.csi.driver.io"))
 	if assert.NoError(t, err, "connect via absolute path") &&
 		assert.NotNil(t, conn, "got a connection") {
 		defer conn.Close()
@@ -222,7 +227,7 @@ func TestDisconnect(t *testing.T) {
 	}()
 
 	reconnectCount := 0
-	conn, err := Connect(addr, OnConnectionLoss(func() bool {
+	conn, err := Connect(addr, metrics.NewCSIMetricsManager("fake.csi.driver.io"), OnConnectionLoss(func() bool {
 		reconnectCount++
 		// Don't reconnect.
 		return false
@@ -273,7 +278,7 @@ func TestExplicitReconnect(t *testing.T) {
 	}()
 
 	reconnectCount := 0
-	conn, err := Connect(addr, OnConnectionLoss(func() bool {
+	conn, err := Connect(addr, metrics.NewCSIMetricsManager("fake.csi.driver.io"), OnConnectionLoss(func() bool {
 		reconnectCount++
 		// Reconnect.
 		return true
@@ -314,3 +319,89 @@ func TestExplicitReconnect(t *testing.T) {
 		assert.Equal(t, 1, reconnectCount, "connection loss callback should be called once")
 	}
 }
+
+func TestConnectMetrics(t *testing.T) {
+	tmp := tmpDir(t)
+	defer os.RemoveAll(tmp)
+	addr, stopServer := startServer(t, tmp, nil, nil)
+	defer stopServer()
+
+	cmm := metrics.NewCSIMetricsManager("fake.csi.driver.io")
+	conn, err := Connect(addr, cmm)
+	if assert.NoError(t, err, "connect via absolute path") &&
+		assert.NotNil(t, conn, "got a connection") {
+		defer conn.Close()
+		assert.Equal(t, connectivity.Ready, conn.GetState(), "connection ready")
+
+		if err := conn.Invoke(context.Background(), "/csi.v1.Controller/ControllerGetCapabilities", nil, nil); assert.Error(t, err) {
+			errStatus, _ := status.FromError(err)
+			assert.Equal(t, codes.Unimplemented, errStatus.Code(), "not implemented")
+		}
+	}
+
+	expectedMetrics := `# HELP csi_sidecar_operations_seconds [ALPHA] Container Storage Interface operation duration with gRPC error code status total
+		# TYPE csi_sidecar_operations_seconds histogram
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="0.1"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="0.25"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="0.5"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="1"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="2.5"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="5"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="10"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="15"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="25"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="50"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="120"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="300"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="600"} 1
+		csi_sidecar_operations_seconds_bucket{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities",le="+Inf"} 1
+		csi_sidecar_operations_seconds_sum{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities"} 0
+		csi_sidecar_operations_seconds_count{driver_name="fake.csi.driver.io",grpc_status_code="Unimplemented",method_name="/csi.v1.Controller/ControllerGetCapabilities"} 1
+	`
+
+	if err := testutil.GatherAndCompare(
+		cmm.GetRegistry(), strings.NewReader(expectedMetrics)); err != nil {
+		// Ignore mismatches on csi_sidecar_operations_seconds_sum metric because execution time will vary from test to test.
+		err = verifyMetricsError(t, err, "csi_sidecar_operations_seconds_sum")
+		if err != nil {
+			t.Errorf("Expected metrics not found -- %v", err)
+		}
+	}
+}
+
+func verifyMetricsError(t *testing.T, err error, metricToIgnore string) error {
+	errStringArr := strings.Split(err.Error(), "got:")
+
+	if len(errStringArr) != 2 {
+		return err
+	}
+
+	want := errStringArr[0]
+	got := strings.TrimSpace(errStringArr[1])
+
+	if want == "" || got == "" {
+		return err
+	}
+
+	wantArr := strings.Split(err.Error(), "want:")
+	if len(wantArr) != 2 {
+		return err
+	}
+
+	want = strings.TrimSpace(wantArr[1])
+
+	//fmt.Printf("_want_ %q\r\n_got_ %q", want, got)
+	gotScanner := bufio.NewScanner(strings.NewReader(got))
+	wantScanner := bufio.NewScanner(strings.NewReader(want))
+	for gotScanner.Scan() {
+		wantScanner.Scan()
+		wantLine := wantScanner.Text()
+		gotLine := gotScanner.Text()
+		if wantLine != gotLine && !strings.HasPrefix(gotLine, metricToIgnore) {
+			t.Errorf("\r\nMetric Want: %q\r\nMetric Got:  %q\r\n", wantLine, gotLine)
+			return err
+		}
+	}
+
+	return nil
+}

go.mod

@@ -4,36 +4,17 @@ go 1.12
 
 require (
 	github.com/container-storage-interface/spec v1.1.0
-	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/evanphx/json-patch v4.5.0+incompatible // indirect
-	github.com/gogo/protobuf v1.2.1 // indirect
 	github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef // indirect
-	github.com/golang/protobuf v1.3.1
-	github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf // indirect
+	github.com/golang/protobuf v1.3.2
 	github.com/googleapis/gnostic v0.2.0 // indirect
-	github.com/json-iterator/go v1.1.6 // indirect
-	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 // indirect
 	github.com/onsi/ginkgo v1.10.2 // indirect
-	github.com/onsi/gomega v1.7.0 // indirect
 	github.com/pkg/errors v0.8.1 // indirect
-	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/stretchr/testify v1.3.0
-	golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c // indirect
-	golang.org/x/net v0.0.0-20190328230028-74de082e2cca
-	golang.org/x/oauth2 v0.0.0-20190319182350-c85d3e98c914 // indirect
-	golang.org/x/sys v0.0.0-20190329044733-9eb1bfa1ce65 // indirect
-	golang.org/x/time v0.0.0-20190308202827-9d24e82272b4 // indirect
-	google.golang.org/appengine v1.5.0 // indirect
-	google.golang.org/genproto v0.0.0-20190327125643-d831d65fe17d // indirect
+	github.com/stretchr/testify v1.4.0
+	golang.org/x/net v0.0.0-20191004110552-13f9640d40b9
 	google.golang.org/grpc v1.19.1
-	gopkg.in/inf.v0 v0.9.1 // indirect
-	gopkg.in/yaml.v2 v2.2.2 // indirect
-	k8s.io/api v0.0.0-20190313235455-40a48860b5ab
-	k8s.io/apimachinery v0.0.0-20190313205120-d7deff9243b1 // indirect
+	k8s.io/api v0.17.0
 	k8s.io/client-go v11.0.0+incompatible
-	k8s.io/klog v0.2.0
-	k8s.io/kube-openapi v0.0.0-20190320154901-5e45bb682580 // indirect
-	k8s.io/utils v0.0.0-20190308190857-21c4ce38f2a7 // indirect
-	sigs.k8s.io/yaml v1.1.0 // indirect
+	k8s.io/component-base v0.17.0
+	k8s.io/klog v1.0.0
 )