feat(exporter): dothill_csi_rpc_call metric

go.mod

@@ -9,7 +9,8 @@ require (
 	github.com/kubernetes-csi/csi-lib-iscsi v0.0.0-20200118015005-959f12c91ca8
 	github.com/kubernetes-csi/csi-test v0.0.0-20191016154743-6931aedb3df0
 	github.com/pkg/errors v0.9.1
-	golang.org/x/sync v0.0.0-20190423024810-112230192c58
+	github.com/prometheus/client_golang v1.9.0
+	golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e
 	google.golang.org/grpc v1.29.1
 	k8s.io/klog v1.0.0
 )

helm/dothill-csi/templates/daemonset.yaml

@@ -72,6 +72,9 @@ spec:
           - containerPort: 9808
             name: healthz
             protocol: TCP
+          - containerPort: 9842
+            name: prom-exporter
+            protocol: TCP
           livenessProbe:
             httpGet:
               path: /healthz

pkg/common/driver.go

@@ -9,6 +9,7 @@ import (
 	"syscall"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/enix/dothill-csi/pkg/exporter"
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"google.golang.org/grpc"
 	"k8s.io/klog"
@@ -35,9 +36,10 @@ const (
 // Driver contains main resources needed by the driver
 // and references the underlying specific driver
 type Driver struct {
-	impl   DriverImpl
-	socket net.Listener
-	server *grpc.Server
+	impl     DriverImpl
+	socket   net.Listener
+	server   *grpc.Server
+	exporter *exporter.Exporter
 }
 
 // DriverImpl is the implementation of the specific driver
@@ -67,13 +69,22 @@ type WithVolumeCaps interface {
 
 // NewDriver is a convenience function for creating an abstract driver
 func NewDriver(impl DriverImpl) *Driver {
+	exporter := exporter.New(9842)
+	interceptors := append(
+		*impl.NewServerInterceptors(newLogRoutineServerInterceptor(impl)),
+		func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+			resp, err := handler(ctx, req)
+			exporter.Collector.IncCSICall(info.FullMethod, err == nil)
+			return resp, err
+		},
+	)
+
 	return &Driver{
 		impl: impl,
 		server: grpc.NewServer(
-			grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
-				*impl.NewServerInterceptors(newLogRoutineServerInterceptor(impl))...,
-			)),
+			grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(interceptors...)),
 		),
+		exporter: exporter,
 	}
 }
 
@@ -131,6 +142,10 @@ func (driver *Driver) Start(bind string) {
 		driver.Stop()
 	}()
 
+	go func() {
+		driver.exporter.ListenAndServe()
+	}()
+
 	klog.Infof("driver listening on %s\n\n", bind)
 	driver.server.Serve(socket)
 }

pkg/exporter/collector.go

@@ -0,0 +1,58 @@
+package exporter
+
+import (
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type Collector struct {
+	csiRPCCallCounters map[string]prometheus.Counter
+}
+
+var (
+	csiRPCCallMetric = "dothill_csi_rpc_call"
+	csiRPCCallHelp   = "How many CSI RPC calls have been executed"
+)
+
+func NewCollector() *Collector {
+	return &Collector{
+		csiRPCCallCounters: map[string]prometheus.Counter{},
+	}
+}
+
+func (collector *Collector) Describe(ch chan<- *prometheus.Desc) {
+	for _, counter := range collector.csiRPCCallCounters {
+		ch <- counter.Desc()
+	}
+}
+
+func (collector *Collector) Collect(ch chan<- prometheus.Metric) {
+	for _, counter := range collector.csiRPCCallCounters {
+		ch <- counter
+	}
+}
+
+func (collector *Collector) getCSIRPCCallCounter(method string, success bool) prometheus.Counter {
+	ID := method + ":" + fmt.Sprintf("%t", success)
+
+	if counter, ok := collector.csiRPCCallCounters[ID]; ok {
+		return counter
+	}
+
+	counter := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: csiRPCCallMetric,
+		Help: csiRPCCallHelp,
+		ConstLabels: prometheus.Labels{
+			"method":  method,
+			"success": fmt.Sprintf("%t", success),
+		},
+	})
+	collector.csiRPCCallCounters[ID] = counter
+
+	return counter
+}
+
+func (collector *Collector) IncCSIRPCCall(method string, success bool) {
+	collector.getCSIRPCCallCounter(method, success).Inc()
+}

pkg/exporter/exporter.go

@@ -0,0 +1,76 @@
+package exporter
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"k8s.io/klog"
+)
+
+// Exporter : Configuration (from command-line)
+type Exporter struct {
+	Port      int
+	Collector *Collector
+
+	listener net.Listener
+	server   *http.Server
+}
+
+func New(port int) *Exporter {
+	return &Exporter{
+		Port:      port,
+		Collector: NewCollector(),
+	}
+}
+
+// ListenAndServe : Convenience function to start exporter
+func (exporter *Exporter) ListenAndServe() error {
+	if err := exporter.Listen(); err != nil {
+		return err
+	}
+
+	return exporter.Serve()
+}
+
+// Listen : Listen for requests
+func (exporter *Exporter) Listen() error {
+	err := prometheus.Register(exporter.Collector)
+	if err != nil {
+		if registered, ok := err.(prometheus.AlreadyRegisteredError); ok {
+			prometheus.Unregister(registered.ExistingCollector)
+			prometheus.MustRegister(exporter.Collector)
+		}
+	}
+
+	listen := fmt.Sprintf(":%d", exporter.Port)
+	klog.Infof("listening on %s", listen)
+
+	listener, err := net.Listen("tcp", listen)
+	if err != nil {
+		return err
+	}
+
+	exporter.listener = listener
+	return nil
+}
+
+// Serve : Actually reply to requests
+func (exporter *Exporter) Serve() error {
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+
+	exporter.server = &http.Server{
+		Handler: mux,
+	}
+
+	return exporter.server.Serve(exporter.listener)
+}
+
+// Shutdown : Properly tear down server
+func (exporter *Exporter) Shutdown() error {
+	return exporter.server.Shutdown(context.Background())
+}