balancer/pickfirst: Add pick first metrics #7839
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,281 @@ | ||
| /* | ||
| * | ||
| * 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 pickfirstleaf_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "google.golang.org/grpc" | ||
| "google.golang.org/grpc/balancer/pickfirst/pickfirstleaf" | ||
| "google.golang.org/grpc/credentials/insecure" | ||
| "google.golang.org/grpc/internal" | ||
| "google.golang.org/grpc/internal/stubserver" | ||
| "google.golang.org/grpc/internal/testutils/stats" | ||
| testgrpc "google.golang.org/grpc/interop/grpc_testing" | ||
| testpb "google.golang.org/grpc/interop/grpc_testing" | ||
| "google.golang.org/grpc/resolver" | ||
| "google.golang.org/grpc/resolver/manual" | ||
| "google.golang.org/grpc/serviceconfig" | ||
| "google.golang.org/grpc/stats/opentelemetry" | ||
|
|
||
| "go.opentelemetry.io/otel/attribute" | ||
| "go.opentelemetry.io/otel/sdk/metric" | ||
| "go.opentelemetry.io/otel/sdk/metric/metricdata" | ||
| "go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest" | ||
| ) | ||
|
|
||
| var pfConfig string | ||
|
|
||
| func init() { | ||
| pfConfig = fmt.Sprintf(`{ | ||
| "loadBalancingConfig": [ | ||
| { | ||
| %q: { | ||
| } | ||
| } | ||
| ] | ||
| }`, pickfirstleaf.Name) | ||
| } | ||
|
|
||
| // TestPickFirstMetrics tests pick first metrics. It configures a pick first | ||
| // balancer, causes it to connect and then disconnect, and expects the | ||
| // subsequent metrics to emit from that. | ||
| func (s) TestPickFirstMetrics(t *testing.T) { | ||
| ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) | ||
| defer cancel() | ||
|
|
||
| ss := &stubserver.StubServer{ | ||
| EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) { | ||
| return &testpb.Empty{}, nil | ||
| }, | ||
| } | ||
| ss.StartServer() | ||
arjan-bal marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| defer ss.Stop() | ||
|
|
||
| sc := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(pfConfig) | ||
arjan-bal marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| r := manual.NewBuilderWithScheme("whatever") | ||
| r.InitialState(resolver.State{ | ||
| ServiceConfig: sc, | ||
| Addresses: []resolver.Address{{Addr: ss.Address}}}, | ||
| ) | ||
|
|
||
| tmr := stats.NewTestMetricsRecorder() | ||
| cc, err := grpc.NewClient(r.Scheme()+":///", grpc.WithStatsHandler(tmr), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r)) | ||
| if err != nil { | ||
| t.Fatalf("NewClient() failed with error: %v", err) | ||
| } | ||
| defer cc.Close() | ||
|
|
||
| tsc := testgrpc.NewTestServiceClient(cc) | ||
| if _, err := tsc.EmptyCall(ctx, &testpb.Empty{}); err != nil { | ||
| t.Fatalf("EmptyCall() failed: %v", err) | ||
| } | ||
|
|
||
| if got, _ := tmr.Metric("grpc.lb.pick_first.connection_attempts_succeeded"); got != 1 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.connection_attempts_succeeded", got, 1) | ||
| } | ||
| if got, _ := tmr.Metric("grpc.lb.pick_first.connection_attempts_failed"); got != 0 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.connection_attempts_failed", got, 0) | ||
| } | ||
| if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 0 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0) | ||
| } | ||
|
|
||
| ss.Stop() | ||
| var foundDisconnections bool | ||
| for ; ctx.Err() == nil; <-time.After(time.Millisecond) { | ||
| if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got == 1 { | ||
| foundDisconnections = true | ||
| break | ||
| } | ||
| } | ||
| if !foundDisconnections { | ||
| t.Fatalf("error waiting for grpc.lb.pick_first.disconnections metric: %v", ctx.Err()) | ||
| } | ||
arjan-bal marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| // TestPickFirstMetricsFailure tests the connection attempts failed metric. It | ||
| // configures a channel and scenario that causes a pick first connection attempt | ||
| // to fail, and then expects that metric to emit. | ||
| func (s) TestPickFirstMetricsFailure(t *testing.T) { | ||
| ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) | ||
| defer cancel() | ||
|
|
||
| sc := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(pfConfig) | ||
|
|
||
| r := manual.NewBuilderWithScheme("whatever") | ||
| r.InitialState(resolver.State{ | ||
| ServiceConfig: sc, | ||
| Addresses: []resolver.Address{{Addr: "bad address"}}}, | ||
| ) | ||
| grpcTarget := r.Scheme() + ":///" | ||
| tmr := stats.NewTestMetricsRecorder() | ||
| cc, err := grpc.NewClient(grpcTarget, grpc.WithStatsHandler(tmr), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r)) | ||
| if err != nil { | ||
| t.Fatalf("NewClient() failed with error: %v", err) | ||
| } | ||
| defer cc.Close() | ||
|
|
||
| tsc := testgrpc.NewTestServiceClient(cc) | ||
| if _, err := tsc.EmptyCall(ctx, &testpb.Empty{}); err == nil { | ||
| t.Fatalf("EmptyCall() passed when expected to fail") | ||
| } | ||
|
|
||
| if got, _ := tmr.Metric("grpc.lb.pick_first.connection_attempts_succeeded"); got != 0 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.connection_attempts_succeeded", got, 0) | ||
| } | ||
| if got, _ := tmr.Metric("grpc.lb.pick_first.connection_attempts_failed"); got != 1 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.connection_attempts_failed", got, 1) | ||
| } | ||
| if got, _ := tmr.Metric("grpc.lb.pick_first.disconnections"); got != 0 { | ||
| t.Errorf("Unexpected data for metric %v, got: %v, want: %v", "grpc.lb.pick_first.disconnections", got, 0) | ||
| } | ||
| } | ||
|
|
||
| // TestPickFirstMetricsE2E tests the pick first metrics end to end. It | ||
| // configures a channel with an OpenTelemetry plugin, induces all 3 pick first | ||
| // metrics to emit, and makes sure the correct OpenTelemetry metrics atoms emit. | ||
| func (s) TestPickFirstMetricsE2E(t *testing.T) { | ||
| ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) | ||
| defer cancel() | ||
|
|
||
| ss := &stubserver.StubServer{ | ||
| EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) { | ||
| return &testpb.Empty{}, nil | ||
| }, | ||
| } | ||
| ss.StartServer() | ||
| defer ss.Stop() | ||
|
|
||
| sc := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(pfConfig) | ||
| r := manual.NewBuilderWithScheme("whatever") | ||
| r.InitialState(resolver.State{ | ||
| ServiceConfig: sc, | ||
| Addresses: []resolver.Address{{Addr: "bad address"}}}, | ||
| ) // Will trigger connection failed. | ||
|
|
||
| grpcTarget := r.Scheme() + ":///" | ||
| reader := metric.NewManualReader() | ||
| provider := metric.NewMeterProvider(metric.WithReader(reader)) | ||
| mo := opentelemetry.MetricsOptions{ | ||
| MeterProvider: provider, | ||
| Metrics: opentelemetry.DefaultMetrics().Add("grpc.lb.pick_first.disconnections", "grpc.lb.pick_first.connection_attempts_succeeded", "grpc.lb.pick_first.connection_attempts_failed"), | ||
| } | ||
|
|
||
| cc, err := grpc.NewClient(grpcTarget, opentelemetry.DialOption(opentelemetry.Options{MetricsOptions: mo}), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(r)) | ||
| if err != nil { | ||
| t.Fatalf("NewClient() failed with error: %v", err) | ||
| } | ||
| defer cc.Close() | ||
|
|
||
| tsc := testgrpc.NewTestServiceClient(cc) | ||
| if _, err := tsc.EmptyCall(ctx, &testpb.Empty{}); err == nil { | ||
| t.Fatalf("EmptyCall() passed when expected to fail") | ||
| } | ||
|
|
||
| r.UpdateState(resolver.State{ | ||
| ServiceConfig: sc, | ||
| Addresses: []resolver.Address{{Addr: ss.Address}}, | ||
| }) // Will trigger successful connection metric. | ||
| if _, err := tsc.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil { | ||
| t.Fatalf("EmptyCall() failed: %v", err) | ||
| } | ||
|
|
||
| ccState := cc.GetState() | ||
| // Stop the server, that should send signal to disconnect, which will | ||
| // eventually emit disconnection metric. | ||
| ss.Stop() | ||
| if ok := cc.WaitForStateChange(ctx, ccState); !ok { | ||
| t.Fatal("Timeout waiting for channel to go IDLE") | ||
| } | ||
| wantMetrics := []metricdata.Metrics{ | ||
| { | ||
| Name: "grpc.lb.pick_first.connection_attempts_succeeded", | ||
| Description: "EXPERIMENTAL. Number of successful connection attempts.", | ||
| Unit: "attempt", | ||
| Data: metricdata.Sum[int64]{ | ||
| DataPoints: []metricdata.DataPoint[int64]{ | ||
| { | ||
| Attributes: attribute.NewSet(attribute.String("grpc.target", grpcTarget)), | ||
| Value: 1, | ||
| }, | ||
| }, | ||
| Temporality: metricdata.CumulativeTemporality, | ||
| IsMonotonic: true, | ||
| }, | ||
| }, | ||
| { | ||
| Name: "grpc.lb.pick_first.connection_attempts_failed", | ||
| Description: "EXPERIMENTAL. Number of failed connection attempts.", | ||
| Unit: "attempt", | ||
| Data: metricdata.Sum[int64]{ | ||
| DataPoints: []metricdata.DataPoint[int64]{ | ||
| { | ||
| Attributes: attribute.NewSet(attribute.String("grpc.target", grpcTarget)), | ||
| Value: 1, | ||
| }, | ||
| }, | ||
| Temporality: metricdata.CumulativeTemporality, | ||
| IsMonotonic: true, | ||
| }, | ||
| }, | ||
| { | ||
| Name: "grpc.lb.pick_first.disconnections", | ||
| Description: "EXPERIMENTAL. Number of times the selected subchannel becomes disconnected.", | ||
| Unit: "disconnection", | ||
| Data: metricdata.Sum[int64]{ | ||
| DataPoints: []metricdata.DataPoint[int64]{ | ||
| { | ||
| Attributes: attribute.NewSet(attribute.String("grpc.target", grpcTarget)), | ||
| Value: 1, | ||
| }, | ||
| }, | ||
| Temporality: metricdata.CumulativeTemporality, | ||
| IsMonotonic: true, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| gotMetrics := metricsDataFromReader(ctx, reader) | ||
| for _, metric := range wantMetrics { | ||
| val, ok := gotMetrics[metric.Name] | ||
| if !ok { | ||
| t.Fatalf("Metric %v not present in recorded metrics", metric.Name) | ||
| } | ||
| if !metricdatatest.AssertEqual(t, metric, val, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars()) { | ||
| t.Fatalf("Metrics data type not equal for metric: %v", metric.Name) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func metricsDataFromReader(ctx context.Context, reader *metric.ManualReader) map[string]metricdata.Metrics { | ||
| rm := &metricdata.ResourceMetrics{} | ||
| reader.Collect(ctx, rm) | ||
| gotMetrics := map[string]metricdata.Metrics{} | ||
| for _, sm := range rm.ScopeMetrics { | ||
| for _, m := range sm.Metrics { | ||
| gotMetrics[m.Name] = m | ||
| } | ||
| } | ||
| return gotMetrics | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -36,6 +36,7 @@ | |
| "google.golang.org/grpc/balancer" | ||
| "google.golang.org/grpc/balancer/pickfirst/internal" | ||
| "google.golang.org/grpc/connectivity" | ||
| estats "google.golang.org/grpc/experimental/stats" | ||
easwars marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| "google.golang.org/grpc/grpclog" | ||
| "google.golang.org/grpc/internal/envconfig" | ||
| internalgrpclog "google.golang.org/grpc/internal/grpclog" | ||
|
|
@@ -57,7 +58,28 @@ | |
| // Name is the name of the pick_first_leaf balancer. | ||
| // It is changed to "pick_first" in init() if this balancer is to be | ||
| // registered as the default pickfirst. | ||
| Name = "pick_first_leaf" | ||
| disconnectionsMetric = estats.RegisterInt64Count(estats.MetricDescriptor{ | ||
|
There was a problem hiding this comment. This one is suffixed with There was a problem hiding this comment. Good point. Done. The RLS/WRR metrics are also suffixed with Metric, just checked. |
||
| Name: "grpc.lb.pick_first.disconnections", | ||
| Description: "EXPERIMENTAL. Number of times the selected subchannel becomes disconnected.", | ||
| Unit: "disconnection", | ||
| Labels: []string{"grpc.target"}, | ||
| Default: false, | ||
| }) | ||
| connectionAttemptsSucceeded = estats.RegisterInt64Count(estats.MetricDescriptor{ | ||
| Name: "grpc.lb.pick_first.connection_attempts_succeeded", | ||
| Description: "EXPERIMENTAL. Number of successful connection attempts.", | ||
| Unit: "attempt", | ||
| Labels: []string{"grpc.target"}, | ||
| Default: false, | ||
| }) | ||
| connectionAttemptsFailed = estats.RegisterInt64Count(estats.MetricDescriptor{ | ||
| Name: "grpc.lb.pick_first.connection_attempts_failed", | ||
| Description: "EXPERIMENTAL. Number of failed connection attempts.", | ||
| Unit: "attempt", | ||
| Labels: []string{"grpc.target"}, | ||
| Default: false, | ||
| }) | ||
| ) | ||
|
|
||
| const ( | ||
|
|
@@ -80,9 +102,12 @@ | |
|
|
||
| type pickfirstBuilder struct{} | ||
|
|
||
| func (pickfirstBuilder) Build(cc balancer.ClientConn, bo balancer.BuildOptions) balancer.Balancer { | ||
| b := &pickfirstBalancer{ | ||
| cc: cc, | ||
| target: bo.Target.String(), | ||
| metricsRecorder: bo.MetricsRecorder, // ClientConn will always create a Metrics Recorder so guaranteed to be non nil. | ||
|
There was a problem hiding this comment. This comment seems like it's unnecessary or in the wrong place? It's not being dereferenced here. There was a problem hiding this comment. This was Easwar's suggestion after our back and forth: #7839 (comment). What do you think about this. I'll drop "so guaranteed to be non nil" from this one because it's documented on the field. |
||
|
|
||
| addressList: addressList{}, | ||
| subConns: resolver.NewAddressMap(), | ||
| state: connectivity.Connecting, | ||
|
|
@@ -147,8 +172,10 @@ | |
| type pickfirstBalancer struct { | ||
| // The following fields are initialized at build time and read-only after | ||
| // that and therefore do not need to be guarded by a mutex. | ||
| logger *internalgrpclog.PrefixLogger | ||
| cc balancer.ClientConn | ||
| target string | ||
| metricsRecorder estats.MetricsRecorder // guaranteed to be non nil | ||
|
|
||
| // The mutex is used to ensure synchronization of updates triggered | ||
| // from the idle picker and the already serialized resolver, | ||
|
|
@@ -548,7 +575,12 @@ | |
| return | ||
| } | ||
|
|
||
| if newState.ConnectivityState == connectivity.TransientFailure { | ||
| connectionAttemptsFailed.Record(b.metricsRecorder, 1, b.target) | ||
| } | ||
|
|
||
| if newState.ConnectivityState == connectivity.Ready { | ||
| connectionAttemptsSucceeded.Record(b.metricsRecorder, 1, b.target) | ||
| b.shutdownRemainingLocked(sd) | ||
| if !b.addressList.seekTo(sd.addr) { | ||
| // This should not fail as we should have only one SubConn after | ||
|
|
@@ -575,6 +607,12 @@ | |
| // the first address when the picker is used. | ||
| b.shutdownRemainingLocked(sd) | ||
| b.state = connectivity.Idle | ||
| // READY SubConn interspliced in between CONNECTING and IDLE, need to | ||
| // account for that. | ||
| if oldState == connectivity.Connecting && newState.ConnectivityState == connectivity.Idle { | ||
|
There was a problem hiding this comment. +1 to something like: if oldState == Connecting {
// A known issue causes a race that prevents the READY state change notification. This works around it.Also it would be great if we could create an issue for that problem and link it here so we know to come back and delete this one it's resolved. There was a problem hiding this comment. Removed second conditional. Created a Github issue and linked to it. |
||
| connectionAttemptsSucceeded.Record(b.metricsRecorder, 1, b.target) | ||
| } | ||
| disconnectionsMetric.Record(b.metricsRecorder, 1, b.target) | ||
| b.addressList.reset() | ||
| b.cc.UpdateState(balancer.State{ | ||
| ConnectivityState: connectivity.Idle, | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think we should have tests for happy eyeballs cases. Did you consider enhancing existing tests to check for metric values at the end of the test? Going down that path would cover a lot of scenarios instead of very simple ones being tested here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I never considered that. Would you rather me write my own happy eyeballs or scale up the existing? I know there was lots of previous contention about adding too many assertions to tests.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That's a good idea. I would suggest adding assertions to the same test because asserting both behaviours in the same test would give a better indication that the features work together. If this severely hampers readability, we can have separate tests.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Agree with @arjan-bal.
I'm OK with it being a separate PR as well.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Scaled up two basic happy eyeballs tests (in a separate commit).
Note that there is a limitation in our test metrics recorder that it only persist the most recently emitted metric value for a given metric name. The reason for this is vast non determinism in other unit tests that use this component, where the most recent was deterministic but not the total metric. It's too far down the rabbit hole now to change I think, but let me know if y'all feel strongly about trying to add a summation assertion/if you want me to add metrics assertions to more happy eyeballs tests/what you think about the additional assertions in general.
I'll be happy to add more, remove the commit, leave as is, etc.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I added a test to the other test that test with a ClientConn, so I think that should be good enough outside the summation thing, which I don't think should be addressed in this PR if at all unless found assertions inadequate.
I saw other tests test interleaving addresses by sending TF, but figured this coverage should be good enough. Thanks for suggestion.