internal/stats: Add metrics recorder list and usage in ClientConn (grpc#7428)

Author: zasweq

balancer/balancer.go

@@ -30,6 +30,7 @@ import (
 	"google.golang.org/grpc/channelz"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
+	estats "google.golang.org/grpc/experimental/stats"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/metadata"
@@ -256,6 +257,10 @@ type BuildOptions struct {
 	// same resolver.Target as passed to the resolver. See the documentation for
 	// the resolver.Target type for details about what it contains.
 	Target resolver.Target
+	// MetricsRecorder is the metrics recorder that balancers can use to record
+	// metrics. Balancer implementations which do not register metrics on
+	// metrics registry and record on them can ignore this field.
+	MetricsRecorder estats.MetricsRecorder
 }
 
 // Builder creates a balancer.

balancer_wrapper.go

@@ -82,6 +82,7 @@ func newCCBalancerWrapper(cc *ClientConn) *ccBalancerWrapper {
 			CustomUserAgent: cc.dopts.copts.UserAgent,
 			ChannelzParent:  cc.channelz,
 			Target:          cc.parsedTarget,
+			MetricsRecorder: cc.metricsRecorderList,
 		},
 		serializer:       grpcsync.NewCallbackSerializer(ctx),
 		serializerCancel: cancel,

clientconn.go

@@ -40,6 +40,7 @@ import (
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/idle"
 	iresolver "google.golang.org/grpc/internal/resolver"
+	"google.golang.org/grpc/internal/stats"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/resolver"
@@ -195,8 +196,11 @@ func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error)
 	cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz)
 	cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers)
 
+	cc.metricsRecorderList = stats.NewMetricsRecorderList(cc.dopts.copts.StatsHandlers)
+
 	cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc.
 	cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout)
+
 	return cc, nil
 }
 
@@ -591,13 +595,14 @@ type ClientConn struct {
 	cancel context.CancelFunc // Cancelled on close.
 
 	// The following are initialized at dial time, and are read-only after that.
-	target          string            // User's dial target.
-	parsedTarget    resolver.Target   // See initParsedTargetAndResolverBuilder().
-	authority       string            // See initAuthority().
-	dopts           dialOptions       // Default and user specified dial options.
-	channelz        *channelz.Channel // Channelz object.
-	resolverBuilder resolver.Builder  // See initParsedTargetAndResolverBuilder().
-	idlenessMgr     *idle.Manager
+	target              string            // User's dial target.
+	parsedTarget        resolver.Target   // See initParsedTargetAndResolverBuilder().
+	authority           string            // See initAuthority().
+	dopts               dialOptions       // Default and user specified dial options.
+	channelz            *channelz.Channel // Channelz object.
+	resolverBuilder     resolver.Builder  // See initParsedTargetAndResolverBuilder().
+	idlenessMgr         *idle.Manager
+	metricsRecorderList *stats.MetricsRecorderList
 
 	// The following provide their own synchronization, and therefore don't
 	// require cc.mu to be held to access them.

experimental/stats/metricregistry.go

@@ -81,6 +81,12 @@ const (
 // on.
 type Int64CountHandle MetricDescriptor
 
+// Descriptor returns the int64 count handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Int64CountHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
 // Record records the int64 count value on the metrics recorder provided.
 func (h *Int64CountHandle) Record(recorder MetricsRecorder, incr int64, labels ...string) {
 	recorder.RecordInt64Count(h, incr, labels...)
@@ -90,6 +96,12 @@ func (h *Int64CountHandle) Record(recorder MetricsRecorder, incr int64, labels .
 // passed at the recording point in order to know which metric to record on.
 type Float64CountHandle MetricDescriptor
 
+// Descriptor returns the float64 count handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Float64CountHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
 // Record records the float64 count value on the metrics recorder provided.
 func (h *Float64CountHandle) Record(recorder MetricsRecorder, incr float64, labels ...string) {
 	recorder.RecordFloat64Count(h, incr, labels...)
@@ -99,6 +111,12 @@ func (h *Float64CountHandle) Record(recorder MetricsRecorder, incr float64, labe
 // is passed at the recording point in order to know which metric to record on.
 type Int64HistoHandle MetricDescriptor
 
+// Descriptor returns the int64 histo handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Int64HistoHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
 // Record records the int64 histo value on the metrics recorder provided.
 func (h *Int64HistoHandle) Record(recorder MetricsRecorder, incr int64, labels ...string) {
 	recorder.RecordInt64Histo(h, incr, labels...)
@@ -109,6 +127,12 @@ func (h *Int64HistoHandle) Record(recorder MetricsRecorder, incr int64, labels .
 // record on.
 type Float64HistoHandle MetricDescriptor
 
+// Descriptor returns the float64 histo handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Float64HistoHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
 // Record records the float64 histo value on the metrics recorder provided.
 func (h *Float64HistoHandle) Record(recorder MetricsRecorder, incr float64, labels ...string) {
 	recorder.RecordFloat64Histo(h, incr, labels...)
@@ -118,6 +142,12 @@ func (h *Float64HistoHandle) Record(recorder MetricsRecorder, incr float64, labe
 // passed at the recording point in order to know which metric to record on.
 type Int64GaugeHandle MetricDescriptor
 
+// Descriptor returns the int64 gauge handle typecast to a pointer to a
+// MetricDescriptor.
+func (h *Int64GaugeHandle) Descriptor() *MetricDescriptor {
+	return (*MetricDescriptor)(h)
+}
+
 // Record records the int64 histo value on the metrics recorder provided.
 func (h *Int64GaugeHandle) Record(recorder MetricsRecorder, incr int64, labels ...string) {
 	recorder.RecordInt64Gauge(h, incr, labels...)

experimental/stats/metricregistry_test.go

@@ -112,27 +112,27 @@ func (s) TestMetricRegistry(t *testing.T) {
 	// The Metric Descriptor in the handle should be able to identify the metric
 	// information. This is the key passed to metrics recorder to identify
 	// metric.
-	if got := fmr.intValues[(*MetricDescriptor)(intCountHandle1)]; got != 1 {
+	if got := fmr.intValues[intCountHandle1.Descriptor()]; got != 1 {
 		t.Fatalf("fmr.intValues[intCountHandle1.MetricDescriptor] got %v, want: %v", got, 1)
 	}
 
 	floatCountHandle1.Record(fmr, 1.2, []string{"some label value", "some optional label value"}...)
-	if got := fmr.floatValues[(*MetricDescriptor)(floatCountHandle1)]; got != 1.2 {
+	if got := fmr.floatValues[floatCountHandle1.Descriptor()]; got != 1.2 {
 		t.Fatalf("fmr.floatValues[floatCountHandle1.MetricDescriptor] got %v, want: %v", got, 1.2)
 	}
 
 	intHistoHandle1.Record(fmr, 3, []string{"some label value", "some optional label value"}...)
-	if got := fmr.intValues[(*MetricDescriptor)(intHistoHandle1)]; got != 3 {
+	if got := fmr.intValues[intHistoHandle1.Descriptor()]; got != 3 {
 		t.Fatalf("fmr.intValues[intHistoHandle1.MetricDescriptor] got %v, want: %v", got, 3)
 	}
 
 	floatHistoHandle1.Record(fmr, 4.3, []string{"some label value", "some optional label value"}...)
-	if got := fmr.floatValues[(*MetricDescriptor)(floatHistoHandle1)]; got != 4.3 {
+	if got := fmr.floatValues[floatHistoHandle1.Descriptor()]; got != 4.3 {
 		t.Fatalf("fmr.floatValues[floatHistoHandle1.MetricDescriptor] got %v, want: %v", got, 4.3)
 	}
 
 	intGaugeHandle1.Record(fmr, 7, []string{"some label value", "some optional label value"}...)
-	if got := fmr.intValues[(*MetricDescriptor)(intGaugeHandle1)]; got != 7 {
+	if got := fmr.intValues[intGaugeHandle1.Descriptor()]; got != 7 {
 		t.Fatalf("fmr.intValues[intGaugeHandle1.MetricDescriptor] got %v, want: %v", got, 7)
 	}
 }
@@ -170,28 +170,28 @@ func TestNumerousIntCounts(t *testing.T) {
 	fmr := newFakeMetricsRecorder(t)
 
 	intCountHandle1.Record(fmr, 1, []string{"some label value", "some optional label value"}...)
-	got := []int64{fmr.intValues[(*MetricDescriptor)(intCountHandle1)], fmr.intValues[(*MetricDescriptor)(intCountHandle2)], fmr.intValues[(*MetricDescriptor)(intCountHandle3)]}
+	got := []int64{fmr.intValues[intCountHandle1.Descriptor()], fmr.intValues[intCountHandle2.Descriptor()], fmr.intValues[intCountHandle3.Descriptor()]}
 	want := []int64{1, 0, 0}
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Fatalf("fmr.intValues (-got, +want): %v", diff)
 	}
 
 	intCountHandle2.Record(fmr, 1, []string{"some label value", "some optional label value"}...)
-	got = []int64{fmr.intValues[(*MetricDescriptor)(intCountHandle1)], fmr.intValues[(*MetricDescriptor)(intCountHandle2)], fmr.intValues[(*MetricDescriptor)(intCountHandle3)]}
+	got = []int64{fmr.intValues[intCountHandle1.Descriptor()], fmr.intValues[intCountHandle2.Descriptor()], fmr.intValues[intCountHandle3.Descriptor()]}
 	want = []int64{1, 1, 0}
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Fatalf("fmr.intValues (-got, +want): %v", diff)
 	}
 
 	intCountHandle3.Record(fmr, 1, []string{"some label value", "some optional label value"}...)
-	got = []int64{fmr.intValues[(*MetricDescriptor)(intCountHandle1)], fmr.intValues[(*MetricDescriptor)(intCountHandle2)], fmr.intValues[(*MetricDescriptor)(intCountHandle3)]}
+	got = []int64{fmr.intValues[intCountHandle1.Descriptor()], fmr.intValues[intCountHandle2.Descriptor()], fmr.intValues[intCountHandle3.Descriptor()]}
 	want = []int64{1, 1, 1}
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Fatalf("fmr.intValues (-got, +want): %v", diff)
 	}
 
 	intCountHandle3.Record(fmr, 1, []string{"some label value", "some optional label value"}...)
-	got = []int64{fmr.intValues[(*MetricDescriptor)(intCountHandle1)], fmr.intValues[(*MetricDescriptor)(intCountHandle2)], fmr.intValues[(*MetricDescriptor)(intCountHandle3)]}
+	got = []int64{fmr.intValues[intCountHandle1.Descriptor()], fmr.intValues[intCountHandle2.Descriptor()], fmr.intValues[intCountHandle3.Descriptor()]}
 	want = []int64{1, 1, 2}
 	if diff := cmp.Diff(got, want); diff != "" {
 		t.Fatalf("fmr.intValues (-got, +want): %v", diff)
@@ -236,26 +236,26 @@ func verifyLabels(t *testing.T, labelsWant []string, optionalLabelsWant []string
 }
 
 func (r *fakeMetricsRecorder) RecordInt64Count(handle *Int64CountHandle, incr int64, labels ...string) {
-	verifyLabels(r.t, (*MetricDescriptor)(handle).Labels, (*MetricDescriptor)(handle).OptionalLabels, labels)
-	r.intValues[(*MetricDescriptor)(handle)] += incr
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.intValues[handle.Descriptor()] += incr
 }
 
 func (r *fakeMetricsRecorder) RecordFloat64Count(handle *Float64CountHandle, incr float64, labels ...string) {
-	verifyLabels(r.t, (*MetricDescriptor)(handle).Labels, (*MetricDescriptor)(handle).OptionalLabels, labels)
-	r.floatValues[(*MetricDescriptor)(handle)] += incr
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.floatValues[handle.Descriptor()] += incr
 }
 
 func (r *fakeMetricsRecorder) RecordInt64Histo(handle *Int64HistoHandle, incr int64, labels ...string) {
-	verifyLabels(r.t, (*MetricDescriptor)(handle).Labels, (*MetricDescriptor)(handle).OptionalLabels, labels)
-	r.intValues[(*MetricDescriptor)(handle)] += incr
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.intValues[handle.Descriptor()] += incr
 }
 
 func (r *fakeMetricsRecorder) RecordFloat64Histo(handle *Float64HistoHandle, incr float64, labels ...string) {
-	verifyLabels(r.t, (*MetricDescriptor)(handle).Labels, (*MetricDescriptor)(handle).OptionalLabels, labels)
-	r.floatValues[(*MetricDescriptor)(handle)] += incr
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.floatValues[handle.Descriptor()] += incr
 }
 
 func (r *fakeMetricsRecorder) RecordInt64Gauge(handle *Int64GaugeHandle, incr int64, labels ...string) {
-	verifyLabels(r.t, (*MetricDescriptor)(handle).Labels, (*MetricDescriptor)(handle).OptionalLabels, labels)
-	r.intValues[(*MetricDescriptor)(handle)] += incr
+	verifyLabels(r.t, handle.Descriptor().Labels, handle.Descriptor().OptionalLabels, labels)
+	r.intValues[handle.Descriptor()] += incr
 }

internal/stats/metrics_recorder_list.go

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package stats
+
+import (
+	"fmt"
+
+	estats "google.golang.org/grpc/experimental/stats"
+	"google.golang.org/grpc/stats"
+)
+
+// MetricsRecorderList forwards Record calls to all of its metricsRecorders.
+//
+// It eats any record calls where the label values provided do not match the
+// number of label keys.
+type MetricsRecorderList struct {
+	// metricsRecorders are the metrics recorders this list will forward to.
+	metricsRecorders []estats.MetricsRecorder
+}
+
+// NewMetricsRecorderList creates a new metric recorder list with all the stats
+// handlers provided which implement the MetricsRecorder interface.
+// If no stats handlers provided implement the MetricsRecorder interface,
+// the MetricsRecorder list returned is a no-op.
+func NewMetricsRecorderList(shs []stats.Handler) *MetricsRecorderList {
+	var mrs []estats.MetricsRecorder
+	for _, sh := range shs {
+		if mr, ok := sh.(estats.MetricsRecorder); ok {
+			mrs = append(mrs, mr)
+		}
+	}
+	return &MetricsRecorderList{
+		metricsRecorders: mrs,
+	}
+}
+
+func verifyLabels(desc *estats.MetricDescriptor, labelsRecv ...string) {
+	if got, want := len(labelsRecv), len(desc.Labels)+len(desc.OptionalLabels); got != want {
+		panic(fmt.Sprintf("Received %d labels in call to record metric %q, but expected %d.", got, desc.Name, want))
+	}
+}
+
+func (l *MetricsRecorderList) RecordInt64Count(handle *estats.Int64CountHandle, incr int64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordInt64Count(handle, incr, labels...)
+	}
+}
+
+func (l *MetricsRecorderList) RecordFloat64Count(handle *estats.Float64CountHandle, incr float64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordFloat64Count(handle, incr, labels...)
+	}
+}
+
+func (l *MetricsRecorderList) RecordInt64Histo(handle *estats.Int64HistoHandle, incr int64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordInt64Histo(handle, incr, labels...)
+	}
+}
+
+func (l *MetricsRecorderList) RecordFloat64Histo(handle *estats.Float64HistoHandle, incr float64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordFloat64Histo(handle, incr, labels...)
+	}
+}
+
+func (l *MetricsRecorderList) RecordInt64Gauge(handle *estats.Int64GaugeHandle, incr int64, labels ...string) {
+	verifyLabels(handle.Descriptor(), labels...)
+
+	for _, metricRecorder := range l.metricsRecorders {
+		metricRecorder.RecordInt64Gauge(handle, incr, labels...)
+	}
+}