feat(metric): adds accepted/closed connections counters to worker and controller servers (#2668)

* adds counters and changelog entry

CHANGELOG.md

@@ -9,6 +9,9 @@ Canonical reference for changes, improvements, and bugfixes for Boundary.
 * Custom Response Headers: Adds ability to set api and ui response headers based
   on status code. Includes default secure CSP and other headers.
   ([PR](https://github.com/hashicorp/boundary/pull/2587))
+* metrics: Adds accepted connections and closed connections counters to keep track
+  downstream connections for worker and controller servers.
+  ([PR](https://github.com/hashicorp/boundary/pull/2668))
 
 ## 0.11.1 (2022/11/30)
 

internal/daemon/controller/internal/metric/cluster.go

@@ -2,6 +2,7 @@ package metric
 
 import (
 	"context"
+	"net"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/daemon/metric"
@@ -12,7 +13,7 @@ import (
 )
 
 const (
-	clusterSubSystem = "controller_cluster"
+	clusterSubsystem = "controller_cluster"
 )
 
 // Because we use grpc's stats.Handler to track close-to-the-wire server-side communication over grpc,
@@ -25,7 +26,7 @@ const (
 var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	prometheus.HistogramOpts{
 		Namespace: globals.MetricNamespace,
-		Subsystem: clusterSubSystem,
+		Subsystem: clusterSubsystem,
 		Name:      "grpc_request_duration_seconds",
 		Help:      "Histogram of latencies for gRPC requests.",
 		Buckets:   prometheus.DefBuckets,
@@ -33,6 +34,28 @@ var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	metric.ListGrpcLabels,
 )
 
+// acceptedConnsTotal keeps a count of the total accepted connections to a controller.
+var acceptedConnsTotal = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: clusterSubsystem,
+		Name:      "accepted_connections_total",
+		Help:      "Count of total accepted network connections to this controller.",
+	},
+	[]string{metric.LabelConnectionType},
+)
+
+// closedConnsTotal keeps a count of the total closed connections to a controller.
+var closedConnsTotal = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: clusterSubsystem,
+		Name:      "closed_connections_total",
+		Help:      "Count of total closed network connections to this controller.",
+	},
+	[]string{metric.LabelConnectionType},
+)
+
 // All the codes expected to be returned by boundary or the grpc framework to
 // requests to the cluster server.
 var expectedGrpcCodes = []codes.Code{
@@ -45,6 +68,11 @@ var expectedGrpcCodes = []codes.Code{
 	codes.Unavailable, codes.Unauthenticated,
 }
 
+func InstrumentClusterTrackingListener(l net.Listener, purpose string) net.Listener {
+	p := prometheus.Labels{metric.LabelConnectionType: purpose}
+	return metric.NewConnectionTrackingListener(l, acceptedConnsTotal.With(p), closedConnsTotal.With(p))
+}
+
 // InstrumentClusterStatsHandler returns a gRPC stats.Handler which observes
 // cluster specific metrics. Use with the cluster gRPC server.
 func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
@@ -57,3 +85,7 @@ func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
 func InitializeClusterCollectors(r prometheus.Registerer, server *grpc.Server) {
 	metric.InitializeGrpcCollectorsFromServer(r, grpcRequestLatency, server, expectedGrpcCodes)
 }
+
+func InitializeConnectionCounters(r prometheus.Registerer) {
+	metric.InitializeConnectionCounters(r, []prometheus.CounterVec{*acceptedConnsTotal, *closedConnsTotal})
+}

internal/daemon/controller/listeners.go

@@ -237,6 +237,7 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 		}
 	}
 
+	metric.InitializeConnectionCounters(c.conf.PrometheusRegisterer)
 	metric.InitializeClusterCollectors(c.conf.PrometheusRegisterer, workerServer)
 
 	ln.GrpcServer = workerServer
@@ -249,13 +250,13 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
 			}
 		}()
 		go func() {
-			err := handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener, c.downstreamRoutes, -1)
+			err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(reverseGrpcListener, "reverse-grpc"), c.downstreamRoutes, -1)
 			if err != nil {
 				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
 			}
 		}()
 		go func() {
-			err := ln.GrpcServer.Serve(multiplexingAuthedListener)
+			err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, "grpc"))
 			if err != nil {
 				event.WriteError(c.baseContext, op, err, event.WithInfoMsg("multiplexingAuthedListener error"))
 			}

internal/daemon/metric/initialize_metrics.go

@@ -15,12 +15,13 @@ import (
 )
 
 const (
-	LabelGrpcService = "grpc_service"
-	LabelGrpcMethod  = "grpc_method"
-	LabelGrpcCode    = "grpc_code"
-	LabelHttpPath    = "path"
-	LabelHttpMethod  = "method"
-	LabelHttpCode    = "code"
+	LabelConnectionType = "conn_type"
+	LabelGrpcService    = "grpc_service"
+	LabelGrpcMethod     = "grpc_method"
+	LabelGrpcCode       = "grpc_code"
+	LabelHttpPath       = "path"
+	LabelHttpMethod     = "method"
+	LabelHttpCode       = "code"
 
 	invalidPathValue = "invalid"
 )
@@ -86,6 +87,15 @@ func InitializeGrpcCollectorsFromPackage(r prometheus.Registerer, v prometheus.O
 	}
 }
 
+func InitializeConnectionCounters(r prometheus.Registerer, counters []prometheus.CounterVec) {
+	if r == nil {
+		return
+	}
+	for _, c := range counters {
+		r.MustRegister(c)
+	}
+}
+
 // InitializeGrpcCollectorsFromServer registers and zeroes a Prometheus
 // histogram, finding all service and method labels from the provided gRPC
 // server.

internal/daemon/metric/instrument_handlers.go

@@ -4,7 +4,9 @@ package metric
 
 import (
 	"context"
+	"net"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/errors"
@@ -95,6 +97,41 @@ func (r requestRecorder) Record(err error) {
 	r.reqLatency.With(r.labels).Observe(time.Since(r.start).Seconds())
 }
 
+type connectionTrackingListener struct {
+	net.Listener
+	acceptedConns prometheus.Counter
+	closedConns   prometheus.Counter
+}
+
+func (l *connectionTrackingListener) Accept() (net.Conn, error) {
+	conn, err := l.Listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+	l.acceptedConns.Inc()
+	return &connectionTrackingListenerConn{Conn: conn, closedConns: l.closedConns}, nil
+}
+
+// NewConnectionTrackingListener registers a new Prometheus gauge with an unique
+// connection type label and wraps an existing listener to track when connections
+// are accepted and closed.
+// Multiple calls to Close() a listener connection will only decrement the gauge
+// once. A call to Close() will decrement the gauge even if Close() errors.
+func NewConnectionTrackingListener(l net.Listener, ac prometheus.Counter, cc prometheus.Counter) *connectionTrackingListener {
+	return &connectionTrackingListener{Listener: l, acceptedConns: ac, closedConns: cc}
+}
+
+type connectionTrackingListenerConn struct {
+	net.Conn
+	dec         sync.Once
+	closedConns prometheus.Counter
+}
+
+func (c *connectionTrackingListenerConn) Close() error {
+	c.dec.Do(func() { c.closedConns.Inc() })
+	return c.Conn.Close()
+}
+
 // StatusFromError retrieves the *status.Status from the provided error.  It'll
 // attempt to unwrap the *status.Error, which is something status.FromError
 // does not do.

internal/daemon/metric/instrument_handlers_test.go

@@ -2,7 +2,9 @@ package metric
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"net"
 	"testing"
 	"time"
 
@@ -15,10 +17,12 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+const testSubsystem = "test_metric"
+
 var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	prometheus.HistogramOpts{
 		Namespace: globals.MetricNamespace,
-		Subsystem: "test_metric",
+		Subsystem: testSubsystem,
 		Name:      "grpc_request_duration_seconds",
 		Help:      "Test histogram.",
 		Buckets:   prometheus.DefBuckets,
@@ -26,6 +30,225 @@ var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	ListGrpcLabels,
 )
 
+var testAcceptedConns prometheus.CounterVec = *prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: testSubsystem,
+		Name:      "test_accepted_connections_total",
+		Help:      "Test CounterVec.",
+	},
+	[]string{LabelConnectionType},
+)
+
+var testClosedConns prometheus.CounterVec = *prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: testSubsystem,
+		Name:      "test_closed_connections_total",
+		Help:      "Test CounterVec.",
+	},
+	[]string{LabelConnectionType},
+)
+
+type testPrometheusCounter struct {
+	prometheus.Metric
+	prometheus.Collector
+
+	incCalledN int
+	t          *testing.T
+}
+
+func (tpg *testPrometheusCounter) Inc()        { tpg.incCalledN++ }
+func (tpg *testPrometheusCounter) Add(float64) { tpg.t.Fatal("testPrometheusCounter Add() called") }
+
+type testListener struct {
+	net.Listener
+	lastClientConn net.Conn
+}
+
+func (l *testListener) Accept() (net.Conn, error) {
+	s, c := net.Pipe()
+	l.lastClientConn = c
+	return s, nil
+}
+
+func (l *testListener) Close() error {
+	return l.lastClientConn.Close()
+}
+
+type erroringAcceptListener struct {
+	net.Listener
+}
+
+func (l *erroringAcceptListener) Accept() (net.Conn, error) {
+	return nil, errors.New("error for testcase")
+}
+
+type erroringCloseListener struct {
+	net.Listener
+	lastClientConn net.Conn
+}
+
+func (l *erroringCloseListener) Accept() (net.Conn, error) {
+	s, c := net.Pipe()
+	l.lastClientConn = c
+	return &erroringConn{Conn: s}, nil
+}
+
+type erroringConn struct {
+	net.Conn
+}
+
+func (c *erroringConn) Close() error {
+	c.Conn.Close()
+	return errors.New("error for testcase")
+}
+
+func TestNewConnectionTrackingListener(t *testing.T) {
+	t.Run("set-label",
+		func(t *testing.T) {
+			l := &testListener{}
+			acceptedConns := testAcceptedConns.With(prometheus.Labels{LabelConnectionType: "test_label"})
+			closedConns := testClosedConns.With(prometheus.Labels{LabelConnectionType: "test_label"})
+			ctl := NewConnectionTrackingListener(l, acceptedConns, closedConns)
+			require.NotNil(t, ctl)
+
+			assert.Equal(t, ctl.Listener, l)
+			cc, err := ctl.Accept()
+			require.NoError(t, err)
+			require.NotNil(t, cc)
+
+			// check purpose label was populated correctly by attempting to delete it
+			assert.Equal(t, testAcceptedConns.DeleteLabelValues("test_label"), true)
+			assert.Equal(t, testClosedConns.DeleteLabelValues("test_label"), true)
+			require.NoError(t, cc.Close())
+		})
+	t.Run("accept-err",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			el := &erroringAcceptListener{}
+			ctl := NewConnectionTrackingListener(el, tpcAcc, tpcClo)
+			require.NotNil(t, ctl)
+
+			cc, err := ctl.Accept()
+			assert.Nil(t, cc)
+			assert.Contains(t, "error for testcase", err.Error())
+			assert.Equal(t, 0, tpcAcc.incCalledN)
+			assert.Equal(t, 0, tpcClo.incCalledN)
+		})
+	t.Run("accept-multiple",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			n := 10
+			for i := 0; i < n; i++ {
+				l := &testListener{}
+				ctl := NewConnectionTrackingListener(l, tpcAcc, tpcClo)
+				require.NotNil(t, ctl)
+				cc, err := ctl.Accept()
+				require.NotNil(t, cc)
+				require.NoError(t, err)
+				assert.Equal(t, i+1, tpcAcc.incCalledN)
+
+				require.NoError(t, cc.Close())
+				assert.Equal(t, i+1, tpcClo.incCalledN)
+			}
+		})
+	t.Run("close-err",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			el := &erroringCloseListener{}
+			ctl := NewConnectionTrackingListener(el, tpcAcc, tpcClo)
+			require.NotNil(t, ctl)
+
+			cc, err := ctl.Accept()
+			require.NotNil(t, cc)
+			require.NoError(t, err)
+			assert.Equal(t, 1, tpcAcc.incCalledN)
+
+			assert.Error(t, cc.Close())
+			assert.Equal(t, 1, tpcClo.incCalledN)
+		})
+	t.Run("close-repeat-calls",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			l := &testListener{}
+			ctl := NewConnectionTrackingListener(l, tpcAcc, tpcClo)
+			require.NotNil(t, ctl)
+
+			cc, err := ctl.Accept()
+			require.Nil(t, err)
+			require.NotNil(t, cc)
+			assert.Equal(t, 0, tpcClo.incCalledN)
+
+			for i := 0; i <= 5; i++ {
+				assert.NoError(t, cc.Close())
+			}
+			assert.Equal(t, 1, tpcClo.incCalledN)
+		})
+	t.Run("inc-dec",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			l := &testListener{}
+
+			ctl := NewConnectionTrackingListener(l, tpcAcc, tpcClo)
+			require.NotNil(t, ctl)
+			assert.Equal(t, 0, tpcAcc.incCalledN)
+
+			cc, err := ctl.Accept()
+			require.Nil(t, err)
+			require.NotNil(t, cc)
+			assert.Equal(t, 1, tpcAcc.incCalledN)
+
+			require.NoError(t, cc.Close())
+			assert.Equal(t, 1, tpcClo.incCalledN)
+		},
+	)
+	t.Run("more-inc-dec",
+		func(t *testing.T) {
+			tpcAcc := &testPrometheusCounter{t: t}
+			tpcClo := &testPrometheusCounter{t: t}
+			l1 := &testListener{}
+			l2 := &testListener{}
+			l3 := &testListener{}
+
+			ctl1 := NewConnectionTrackingListener(l1, tpcAcc, tpcClo)
+			require.NotNil(t, ctl1)
+			assert.Equal(t, 0, tpcAcc.incCalledN)
+
+			cc1, err := ctl1.Accept()
+			require.NoError(t, err)
+			require.NotNil(t, cc1)
+			assert.Equal(t, 1, tpcAcc.incCalledN)
+
+			ctl2 := NewConnectionTrackingListener(l2, tpcAcc, tpcClo)
+			require.NotNil(t, ctl2)
+			cc2, err := ctl2.Accept()
+			require.NoError(t, err)
+			require.NotNil(t, cc2)
+			assert.Equal(t, 2, tpcAcc.incCalledN)
+
+			require.NoError(t, cc1.Close())
+			require.NoError(t, cc2.Close())
+			assert.Equal(t, 2, tpcClo.incCalledN)
+
+			ctl3 := NewConnectionTrackingListener(l3, tpcAcc, tpcClo)
+			require.NotNil(t, ctl3)
+			cc3, err := ctl3.Accept()
+			require.NoError(t, err)
+			require.NotNil(t, cc3)
+			assert.Equal(t, 3, tpcAcc.incCalledN)
+
+			require.NoError(t, cc3.Close())
+			assert.Equal(t, 3, tpcClo.incCalledN)
+		},
+	)
+}
+
 func TestNewStatsHandler(t *testing.T) {
 	cases := []struct {
 		name           string

internal/daemon/worker/internal/metric/cluster_client.go

@@ -2,6 +2,7 @@ package metric
 
 import (
 	"context"
+	"time"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/daemon/metric"
@@ -28,6 +29,35 @@ var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
 	metric.ListGrpcLabels,
 )
 
+type requestRecorder struct {
+	reqLatency prometheus.ObserverVec
+	labels     prometheus.Labels
+
+	// measurements
+	start time.Time
+}
+
+// NewRequestRecorder creates a requestRecorder struct which is used to measure gRPC client request latencies.
+// For testing purposes, this method is exported.
+func newRequestRecorder(fullMethodName string, reqLatency prometheus.ObserverVec) requestRecorder {
+	service, method := metric.SplitMethodName(fullMethodName)
+	r := requestRecorder{
+		reqLatency: reqLatency,
+		labels: prometheus.Labels{
+			metric.LabelGrpcMethod:  method,
+			metric.LabelGrpcService: service,
+		},
+		start: time.Now(),
+	}
+
+	return r
+}
+
+func (r requestRecorder) Record(err error) {
+	r.labels[metric.LabelGrpcCode] = metric.StatusFromError(err).Code().String()
+	r.reqLatency.With(r.labels).Observe(time.Since(r.start).Seconds())
+}
+
 // The expected codes returned by the grpc client calls to cluster services.
 var expectedGrpcClientCodes = []codes.Code{
 	codes.OK, codes.Canceled, codes.Unknown, codes.InvalidArgument, codes.DeadlineExceeded, codes.NotFound,

internal/daemon/worker/internal/metric/cluster_server.go

@@ -2,6 +2,7 @@ package metric
 
 import (
 	"context"
+	"net"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/daemon/metric"
@@ -24,12 +25,34 @@ var grpcServerRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec
 		Namespace: globals.MetricNamespace,
 		Subsystem: workerClusterSubsystem,
 		Name:      "grpc_request_duration_seconds",
-		Help:      "Histogram of latencies for gRPC requests between the a worker server and a worker client.",
+		Help:      "Histogram of latencies for gRPC requests between a worker server and a worker client.",
 		Buckets:   prometheus.DefBuckets,
 	},
 	metric.ListGrpcLabels,
 )
 
+// acceptedConnsTotal keeps a count of the total accepted connections to a worker.
+var acceptedConnsTotal = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: workerClusterSubsystem,
+		Name:      "accepted_connections_total",
+		Help:      "Count of total accepted network connections to this worker.",
+	},
+	[]string{metric.LabelConnectionType},
+)
+
+// closedConnsTotal keeps a count of the total closed connections to a worker.
+var closedConnsTotal = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Namespace: globals.MetricNamespace,
+		Subsystem: workerClusterSubsystem,
+		Name:      "closed_connections_total",
+		Help:      "Count of total closed network connections to this worker.",
+	},
+	[]string{metric.LabelConnectionType},
+)
+
 // All the codes expected to be returned by boundary or the grpc framework to
 // requests to the cluster server.
 var expectedGrpcCodes = []codes.Code{
@@ -42,6 +65,11 @@ var expectedGrpcCodes = []codes.Code{
 	codes.Unavailable, codes.Unauthenticated,
 }
 
+func InstrumentWorkerClusterTrackingListener(l net.Listener, purpose string) net.Listener {
+	p := prometheus.Labels{metric.LabelConnectionType: purpose}
+	return metric.NewConnectionTrackingListener(l, acceptedConnsTotal.With(p), closedConnsTotal.With(p))
+}
+
 // InstrumentClusterStatsHandler returns a gRPC stats.Handler which observes
 // cluster-specific metrics for a gRPC server.
 func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
@@ -54,3 +82,7 @@ func InstrumentClusterStatsHandler(ctx context.Context) (stats.Handler, error) {
 func InitializeClusterServerCollectors(r prometheus.Registerer, server *grpc.Server) {
 	metric.InitializeGrpcCollectorsFromServer(r, grpcServerRequestLatency, server, expectedGrpcCodes)
 }
+
+func InitializeConnectionCounters(r prometheus.Registerer) {
+	metric.InitializeConnectionCounters(r, []prometheus.CounterVec{*acceptedConnsTotal, *closedConnsTotal})
+}

internal/daemon/worker/internal/metric/proxy_http.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	proxySubSystem   = "worker_proxy"
+	proxySubsystem   = "worker_proxy"
 	proxyPathValue   = "/v1/proxy"
 	invalidPathValue = "invalid"
 )
@@ -44,7 +44,7 @@ var (
 var httpTimeUntilHeader prometheus.ObserverVec = prometheus.NewHistogramVec(
 	prometheus.HistogramOpts{
 		Namespace: globals.MetricNamespace,
-		Subsystem: proxySubSystem,
+		Subsystem: proxySubsystem,
 		Name:      "http_write_header_duration_seconds",
 		Help:      "Histogram of time elapsed after the TLS connection is established to when the first http header is written back from the server.",
 		Buckets:   prometheus.DefBuckets,

internal/daemon/worker/listeners.go

@@ -196,6 +196,7 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 		}
 	}
 
+	metric.InitializeConnectionCounters(w.conf.PrometheusRegisterer)
 	metric.InitializeClusterServerCollectors(w.conf.PrometheusRegisterer, downstreamServer)
 
 	ln.GrpcServer = downstreamServer
@@ -215,8 +216,8 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
 	return func() {
 		go w.workerAuthSplitListener.Start()
 		go httpServer.Serve(proxyListener)
-		go ln.GrpcServer.Serve(pkiWorkerTrackingListener)
-		go handleSecondaryConnection(cancelCtx, revPkiWorkerTrackingListener, w.downstreamRoutes, -1)
+		go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, "grpc"))
+		go handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), w.downstreamRoutes, -1)
 	}, nil
 }