CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert " feat(metric): adds accepted/closed connections counters for controller and worker cluster connections (#2656)" for failing a build test (#2660)

Browse Source 
This reverts commit 80f72b8511.
This commit is contained in:
Haotian authored and GitHub committed 2022-11-29 10:22:53 -08:00
1 parent c506107127
commit 2b84013379
7 files changed
+13 -349
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/daemon/controller/internal/metric/cluster.go
-32
View File
@@ -2,7 +2,6 @@ package metric
import (
"context"
"net"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/daemon/metric"
@@ -34,28 +33,6 @@ 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.LabelConnectionPurpose},
)
// 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.LabelConnectionPurpose},
)
// All the codes expected to be returned by boundary or the grpc framework to
// requests to the cluster server.
var expectedGrpcCodes = []codes.Code{
@@ -68,11 +45,6 @@ var expectedGrpcCodes = []codes.Code{
codes.Unavailable, codes.Unauthenticated,
}
func InstrumentClusterTrackingListener(l net.Listener, purpose string) net.Listener {
p := prometheus.Labels{metric.LabelConnectionPurpose: 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) {
@@ -85,7 +57,3 @@ 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
+2 -3
View File
@@ -237,7 +237,6 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
}
}
metric.InitializeConnectionCounters(c.conf.PrometheusRegisterer)
metric.InitializeClusterCollectors(c.conf.PrometheusRegisterer, workerServer)
ln.GrpcServer = workerServer
@@ -250,13 +249,13 @@ func (c *Controller) configureForCluster(ln *base.ServerListener) (func(), error
}
}()
go func() {
err := handleSecondaryConnection(c.baseContext, metric.InstrumentClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), c.downstreamRoutes, -1)
err := handleSecondaryConnection(c.baseContext, multiplexingReverseGrpcListener, c.downstreamRoutes, -1)
if err != nil {
event.WriteError(c.baseContext, op, err, event.WithInfoMsg("handleSecondaryConnection error"))
}
}()
go func() {
err := ln.GrpcServer.Serve(metric.InstrumentClusterTrackingListener(multiplexingAuthedListener, "grpc"))
err := ln.GrpcServer.Serve(multiplexingAuthedListener)
if err != nil {
event.WriteError(c.baseContext, op, err, event.WithInfoMsg("multiplexingAuthedListener error"))
}
internal/daemon/metric/initialize_metrics.go
+6 -16
View File
@@ -15,13 +15,12 @@ import (
)
const (
LabelConnectionPurpose = "purpose"
LabelGrpcService = "grpc_service"
LabelGrpcMethod = "grpc_method"
LabelGrpcCode = "grpc_code"
LabelHttpPath = "path"
LabelHttpMethod = "method"
LabelHttpCode = "code"
LabelGrpcService = "grpc_service"
LabelGrpcMethod = "grpc_method"
LabelGrpcCode = "grpc_code"
LabelHttpPath = "path"
LabelHttpMethod = "method"
LabelHttpCode = "code"
invalidPathValue = "invalid"
)
@@ -87,15 +86,6 @@ 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
-37
View File
@@ -4,9 +4,7 @@ package metric
import (
"context"
"net"
"strings"
"sync"
"time"
"github.com/hashicorp/boundary/internal/errors"
@@ -97,41 +95,6 @@ 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
+1 -224
View File
@@ -2,9 +2,7 @@ package metric
import (
"context"
"errors"
"fmt"
"net"
"testing"
"time"
@@ -17,12 +15,10 @@ import (
"google.golang.org/grpc/status"
)
const testSubsystem = "test_metric"
var grpcRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Namespace: globals.MetricNamespace,
Subsystem: testSubsystem,
Subsystem: "test_metric",
Name: "grpc_request_duration_seconds",
Help: "Test histogram.",
Buckets: prometheus.DefBuckets,
@@ -30,225 +26,6 @@ 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{LabelConnectionPurpose},
)
var testClosedConns prometheus.CounterVec = *prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: globals.MetricNamespace,
Subsystem: testSubsystem,
Name: "test_closed_connections_total",
Help: "Test CounterVec.",
},
[]string{LabelConnectionPurpose},
)
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{LabelConnectionPurpose: "test_label"})
closedConns := testClosedConns.With(prometheus.Labels{LabelConnectionPurpose: "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_server.go
+1 -33
View File
@@ -2,7 +2,6 @@ package metric
import (
"context"
"net"
"github.com/hashicorp/boundary/globals"
"github.com/hashicorp/boundary/internal/daemon/metric"
@@ -25,34 +24,12 @@ var grpcServerRequestLatency prometheus.ObserverVec = prometheus.NewHistogramVec
Namespace: globals.MetricNamespace,
Subsystem: workerClusterSubsystem,
Name: "grpc_request_duration_seconds",
Help: "Histogram of latencies for gRPC requests between a worker server and a worker client.",
Help: "Histogram of latencies for gRPC requests between the 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.LabelConnectionPurpose},
)
// 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.LabelConnectionPurpose},
)
// All the codes expected to be returned by boundary or the grpc framework to
// requests to the cluster server.
var expectedGrpcCodes = []codes.Code{
@@ -65,11 +42,6 @@ var expectedGrpcCodes = []codes.Code{
codes.Unavailable, codes.Unauthenticated,
}
func InstrumentWorkerClusterTrackingListener(l net.Listener, purpose string) net.Listener {
p := prometheus.Labels{metric.LabelConnectionPurpose: 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) {
@@ -82,7 +54,3 @@ 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/listeners.go
+3 -4
View File
@@ -196,7 +196,6 @@ 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,9 +214,9 @@ func (w *Worker) configureForWorker(ln *base.ServerListener, logger *log.Logger,
return func() {
go w.workerAuthSplitListener.Start()
go httpServer.Serve(metric.InstrumentWorkerClusterTrackingListener(proxyListener, "proxy"))
go ln.GrpcServer.Serve(metric.InstrumentWorkerClusterTrackingListener(pkiWorkerTrackingListener, "grpc"))
go handleSecondaryConnection(cancelCtx, metric.InstrumentWorkerClusterTrackingListener(revPkiWorkerTrackingListener, "reverse-grpc"), w.downstreamRoutes, -1)
go httpServer.Serve(proxyListener)
go ln.GrpcServer.Serve(pkiWorkerTrackingListener)
go handleSecondaryConnection(cancelCtx, revPkiWorkerTrackingListener, w.downstreamRoutes, -1)
}, nil
}