disconnect reason for subchannel metrics (A94)

Author: mbissa

balancer/pickfirst/metrics_test.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/balancer/pickfirst"
@@ -291,3 +292,125 @@ func metricsDataFromReader(ctx context.Context, reader *metric.ManualReader) map
 	}
 	return gotMetrics
 }
+
+func (s) TestDisconnectLabel(t *testing.T) {
+	// 1. Valid GOAWAY
+	// Server GracefulStop sends GOAWAY with active streams = 0.
+	// This usually sends NoError(0) code.
+	t.Run("GoAway", func(t *testing.T) {
+		runDisconnectLabelTest(t, "GOAWAY NO_ERROR", func(ss *stubserver.StubServer) {
+			ss.S.GracefulStop()
+			// GracefulStop waits for connections to close, which happens after
+			// GOAWAY is sent.
+		})
+	})
+
+	// 2. IO Error
+	// Server Stop closes the listener and active connections immediately.
+	// This often results in "connection reset" or "EOF" (unknown) depending on timing/OS.
+	// Let's check for "unknown" or "connection reset" or "subchannel shutdown" strictly.
+	// In this test env, it often results in io.EOF which we mapped to "unknown".
+	t.Run("IO_Error", func(t *testing.T) {
+		runDisconnectLabelTest(t, "unknown", func(ss *stubserver.StubServer) {
+			ss.Stop()
+		})
+	})
+
+	// Scenario 3: Unknown (Client closes - voluntary? actually client close might be UNKNOWN or not recorded as split)
+	// If client closes, we might not record "disconnections" metric from ClientConn perspective?
+	// disconnections metric is "Number of times the selected subchannel becomes disconnected".
+	// If we close 'cc', we tear down subchannels.
+	// But let's try to trigger a case where we just disconnect without server side action?
+	// Or maybe "unknown" is what we get for "Idle" timeout?
+	// Let's stick to IO and GoAway first which are explicit in A94.
+}
+
+func runDisconnectLabelTest(t *testing.T, wantLabel string, triggerFunc func(*stubserver.StubServer)) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	ss := &stubserver.StubServer{
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
+			return &testpb.Empty{}, nil
+		},
+	}
+	ss.StartServer()
+	defer ss.Stop() // Cleanup in case triggerFunc didn't fully stop or strict cleanup needed
+
+	sc := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(pfConfig)
+	r := manual.NewBuilderWithScheme("whatever")
+	r.InitialState(resolver.State{
+		ServiceConfig: sc,
+		Addresses:     []resolver.Address{{Addr: ss.Address}},
+	})
+
+	grpcTarget := r.Scheme() + ":///"
+	reader := metric.NewManualReader()
+	provider := metric.NewMeterProvider(metric.WithReader(reader))
+	mo := opentelemetry.MetricsOptions{
+		MeterProvider:  provider,
+		Metrics:        opentelemetry.DefaultMetrics().Add("grpc.subchannel.disconnections"),
+		OptionalLabels: []string{"grpc.disconnect_error"},
+	}
+
+	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: %v", err)
+	}
+	defer cc.Close()
+
+	tsc := testgrpc.NewTestServiceClient(cc)
+	// Ensure connected
+	if _, err := tsc.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
+		t.Fatalf("EmptyCall() failed: %v", err)
+	}
+
+	// Trigger disconnection
+	triggerFunc(ss)
+
+	// Wait for Idle state (disconnection happened)
+	testutils.AwaitState(ctx, t, cc, connectivity.Idle)
+
+	// Verify metrics
+
+	gotMetrics := metricsDataFromReader(ctx, reader)
+	val, ok := gotMetrics["grpc.subchannel.disconnections"]
+	if !ok {
+		t.Fatalf("Metric grpc.subchannel.disconnections not found")
+	}
+
+	// We used AssertEqual in the other test, let's use it here too if available.
+	// But checking attributes manually might be safer if AssertEqual is strict on other optional fields.
+	// Let's iterate datapoints.
+	start := time.Now()
+	for {
+		points := val.Data.(metricdata.Sum[int64]).DataPoints
+		if len(points) == 0 {
+			t.Fatalf("No data points for disconnections")
+		}
+		dp := points[0]
+		// Check attributes
+		seenLabel := false
+		var foundAttrs []string
+		for _, kv := range dp.Attributes.ToSlice() {
+			foundAttrs = append(foundAttrs, fmt.Sprintf("%s=%s", kv.Key, kv.Value.AsString()))
+			if kv.Key == "grpc.disconnect_error" {
+				seenLabel = true
+				if kv.Value.AsString() != wantLabel {
+					t.Errorf("Want label %q, got %q", wantLabel, kv.Value.AsString())
+				}
+			}
+		}
+		if !seenLabel && wantLabel != "" {
+			if time.Since(start) > time.Second {
+				t.Errorf("Label grpc.disconnect_error missing. Found attributes: %v", foundAttrs)
+			}
+		}
+		if seenLabel || time.Since(start) > time.Second {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+		gotMetrics = metricsDataFromReader(ctx, reader)
+		val = gotMetrics["grpc.subchannel.disconnections"]
+	}
+}

clientconn.go

@@ -24,12 +24,16 @@ import (
 	"fmt"
 	"math"
 	"net/url"
+	"os"
 	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
+	"golang.org/x/net/http2"
+
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/balancer/base"
 	"google.golang.org/grpc/balancer/pickfirst"
@@ -1270,6 +1274,7 @@ type addrConn struct {
 
 	localityLabel       string
 	backendServiceLabel string
+	disconnectError     string
 }
 
 // Note: this requires a lock on ac.mu.
@@ -1286,9 +1291,14 @@ func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error)
 	// TODO: https://github.com/grpc/grpc-go/issues/7862 - Remove the second
 	// part of the if condition below once the issue is fixed.
 	if ac.state == connectivity.Ready || (ac.state == connectivity.Connecting && s == connectivity.Idle) {
-		disconnectionsMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, ac.backendServiceLabel, ac.localityLabel, "unknown")
+		disconnectError := ac.disconnectError
+		if disconnectError == "" {
+			disconnectError = "unknown"
+		}
+		disconnectionsMetric.Record(ac.cc.metricsRecorderList, 1, ac.cc.target, ac.backendServiceLabel, ac.localityLabel, disconnectError)
 		openConnectionsMetric.Record(ac.cc.metricsRecorderList, -1, ac.cc.target, ac.backendServiceLabel, ac.securityLevelLocked(), ac.localityLabel)
 	}
+	ac.disconnectError = "" // Reset for next time
 	ac.state = s
 	ac.channelz.ChannelMetrics.State.Store(&s)
 	if lastErr == nil {
@@ -1483,7 +1493,7 @@ func (ac *addrConn) createTransport(ctx context.Context, addr resolver.Address,
 	addr.ServerName = ac.cc.getServerName(addr)
 	hctx, hcancel := context.WithCancel(ctx)
 
-	onClose := func(r transport.GoAwayReason) {
+	onClose := func(r transport.GoAwayReason, goAwayCode http2.ErrCode, err error) {
 		ac.mu.Lock()
 		defer ac.mu.Unlock()
 		// adjust params based on GoAwayReason
@@ -1504,6 +1514,7 @@ func (ac *addrConn) createTransport(ctx context.Context, addr resolver.Address,
 			return
 		}
 		ac.transport = nil
+		ac.disconnectError = disconnectErrorString(r, goAwayCode, err)
 		// Refresh the name resolver on any connection loss.
 		ac.cc.resolveNow(resolver.ResolveNowOptions{})
 		// Always go idle and wait for the LB policy to initiate a new
@@ -1560,6 +1571,31 @@ func (ac *addrConn) createTransport(ctx context.Context, addr resolver.Address,
 	return nil
 }
 
+func disconnectErrorString(r transport.GoAwayReason, goAwayCode http2.ErrCode, err error) string {
+	if r != transport.GoAwayInvalid {
+		return fmt.Sprintf("GOAWAY %s", goAwayCode.String())
+	}
+	if err == nil {
+		return "unknown"
+	}
+	if errors.Is(err, context.Canceled) {
+		return "subchannel shutdown"
+	}
+	if errors.Is(err, syscall.ECONNRESET) {
+		return "connection reset"
+	}
+	if errors.Is(err, syscall.ETIMEDOUT) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, os.ErrDeadlineExceeded) {
+		return "connection timed out"
+	}
+	if errors.Is(err, syscall.ECONNABORTED) {
+		return "connection aborted"
+	}
+	if errors.Is(err, syscall.ECONNREFUSED) {
+		return "socket error"
+	}
+	return "unknown"
+}
+
 // startHealthCheck starts the health checking stream (RPC) to watch the health
 // stats of this connection if health checking is requested and configured.
 //
@@ -1665,6 +1701,9 @@ func (ac *addrConn) tearDown(err error) {
 	ac.transport = nil
 	// We have to set the state to Shutdown before anything else to prevent races
 	// between setting the state and logic that waits on context cancellation / etc.
+	if ac.disconnectError == "" {
+		ac.disconnectError = "subchannel shutdown"
+	}
 	ac.updateConnectivityState(connectivity.Shutdown, nil)
 	ac.cancel()
 	ac.curAddr = resolver.Address{}

internal/transport/http2_client.go

@@ -134,6 +134,8 @@ type http2Client struct {
 	// goAwayDebugMessage contains a detailed human readable string about a
 	// GoAway frame, useful for error messages.
 	goAwayDebugMessage string
+	// goAwayCode records the http2.ErrCode received with the GoAway frame.
+	goAwayCode http2.ErrCode
 	// A condition variable used to signal when the keepalive goroutine should
 	// go dormant. The condition for dormancy is based on the number of active
 	// streams and the `PermitWithoutStream` keepalive client parameter. And
@@ -147,7 +149,7 @@ type http2Client struct {
 
 	channelz *channelz.Socket
 
-	onClose func(GoAwayReason)
+	onClose func(GoAwayReason, http2.ErrCode, error)
 
 	bufferPool mem.BufferPool
 
@@ -204,7 +206,7 @@ func isTemporary(err error) bool {
 // NewHTTP2Client constructs a connected ClientTransport to addr based on HTTP2
 // and starts to receive messages on it. Non-nil error returns if construction
 // fails.
-func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason)) (_ ClientTransport, err error) {
+func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason, http2.ErrCode, error)) (_ ClientTransport, err error) {
 	scheme := "http"
 	ctx, cancel := context.WithCancel(ctx)
 	defer func() {
@@ -1015,7 +1017,7 @@ func (t *http2Client) Close(err error) {
 	// Call t.onClose ASAP to prevent the client from attempting to create new
 	// streams.
 	if t.state != draining {
-		t.onClose(GoAwayInvalid)
+		t.onClose(GoAwayInvalid, http2.ErrCodeNo, err)
 	}
 	t.state = closing
 	streams := t.activeStreams
@@ -1086,7 +1088,7 @@ func (t *http2Client) GracefulClose() {
 	if t.logger.V(logLevel) {
 		t.logger.Infof("GracefulClose called")
 	}
-	t.onClose(GoAwayInvalid)
+	t.onClose(GoAwayInvalid, http2.ErrCodeNo, nil)
 	t.state = draining
 	active := len(t.activeStreams)
 	t.mu.Unlock()
@@ -1372,7 +1374,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) error {
 		// draining, to allow the client to stop attempting to create streams
 		// before disallowing new streams on this connection.
 		if t.state != draining {
-			t.onClose(t.goAwayReason)
+			t.onClose(t.goAwayReason, t.goAwayCode, nil)
 			t.state = draining
 		}
 	}
@@ -1406,22 +1408,26 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) error {
 	return nil
 }
 
+// setGoAwayReason sets the value of t.goAwayReason based
+// on the GoAway frame received.
+// It expects a lock on transport's mutex to be held by
+// the caller.
 // setGoAwayReason sets the value of t.goAwayReason based
 // on the GoAway frame received.
 // It expects a lock on transport's mutex to be held by
 // the caller.
 func (t *http2Client) setGoAwayReason(f *http2.GoAwayFrame) {
 	t.goAwayReason = GoAwayNoReason
+	t.goAwayDebugMessage = fmt.Sprintf("code: %s", f.ErrCode)
+	if len(f.DebugData()) > 0 {
+		t.goAwayDebugMessage += fmt.Sprintf(", debug data: %q", string(f.DebugData()))
+	}
 	if f.ErrCode == http2.ErrCodeEnhanceYourCalm {
 		if string(f.DebugData()) == "too_many_pings" {
 			t.goAwayReason = GoAwayTooManyPings
 		}
 	}
-	if len(f.DebugData()) == 0 {
-		t.goAwayDebugMessage = fmt.Sprintf("code: %s", f.ErrCode)
-	} else {
-		t.goAwayDebugMessage = fmt.Sprintf("code: %s, debug data: %q", f.ErrCode, string(f.DebugData()))
-	}
+	t.goAwayCode = f.ErrCode
 }
 
 func (t *http2Client) GetGoAwayReason() (GoAwayReason, string) {