Instrument gRPC dial calls with metrics (temporalio#8162)

## What changed?
Instrument gRPC dial calls with metrics monitoring connection
establishment. This PR adds the following metrics:

1. ServiceDialLatency
2. ServiceDialSuccessCount
3. ServiceDialErrorCount

## Why?
Currently the visibility into gRPC dial calls is limited, there are no
metrics for how long establishing the connection took, or how many
failures we encountered. Since `grpc.NewClient` is "lazy", in the sense
that it doesn't establish connection right away, but rather on the first
RPC, we need to instrument it via some hooks.

This PR uses `grpc.WithContextDialer` to achieve this. gRPC library
itself uses platform-specific keep alive hack to use OS defaults for TCP
keepalive, but since we are on Go 1.23+ we can simply use newly
introduced KeepAliveConfig.

## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks

Author: simvlad

common/metrics/metric_defs.go

@@ -607,6 +607,9 @@ var (
 		"service_grpc_conn_active",
 		WithDescription("Current number of gRPC's active TCP connections."),
 	)
+	ServiceDialLatency                       = NewTimerDef("service_dial_latency", WithDescription("The latency of establishing a new TCP connection."))
+	ServiceDialSuccessCount                  = NewCounterDef("service_dial_success", WithDescription("Number of TCP dial attempts that successfully established a connection."))
+	ServiceDialErrorCount                    = NewCounterDef("service_dial_error", WithDescription("Number of TCP dial attempts that failed to establish a connection."))
 	ServiceLatency                           = NewTimerDef("service_latency")
 	ServiceLatencyNoUserLatency              = NewTimerDef("service_latency_nouserlatency")
 	ServiceLatencyUserLatency                = NewTimerDef("service_latency_userlatency")

common/resource/fx.go

@@ -337,6 +337,7 @@ func RPCFactoryProvider(
 	cfg *config.Config,
 	svcName primitives.ServiceName,
 	logger log.Logger,
+	metricsHandler metrics.Handler,
 	tlsConfigProvider encryption.TLSConfigProvider,
 	resolver *membership.GRPCResolver,
 	tracingStatsHandler telemetry.ClientStatsHandler,
@@ -358,6 +359,7 @@ func RPCFactoryProvider(
 		cfg,
 		svcName,
 		logger,
+		metricsHandler,
 		tlsConfigProvider,
 		frontendURL,
 		frontendHTTPURL,

common/rpc/dial_tracer.go

@@ -0,0 +1,91 @@
+package rpc
+
+import (
+	"context"
+	"net/http/httptrace"
+	"time"
+
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/metrics"
+)
+
+type dialTracer struct {
+	address        string
+	metricsHandler metrics.Handler
+	logger         log.Logger
+}
+
+// newDialTracer creates a dial tracer that logs errors during dial and produces metrics for different stages of the dial process (Connect)
+func newDialTracer(
+	address string,
+	mh metrics.Handler,
+	logger log.Logger,
+) *dialTracer {
+	l := log.With(
+		logger,
+		tag.NewStringTag("service", "client"),
+		tag.NewStringTag("address", address),
+	)
+
+	return &dialTracer{
+		address:        address,
+		metricsHandler: mh,
+		logger:         l,
+	}
+}
+
+func (d *dialTracer) beginNetworkDial(ctx context.Context) (context.Context, *networkDialTrace) {
+	ndt := &networkDialTrace{startedAt: time.Now()}
+
+	// Build a ClientTrace capturing TCP connections.
+	trace := &httptrace.ClientTrace{
+		ConnectStart: func(_, _ string) {
+			ndt.connectStart = time.Now()
+		},
+		ConnectDone: func(_ string, addr string, err error) {
+			// ConnectStart may not have been called if the address is reused
+			if !ndt.connectStart.IsZero() {
+				ndt.connectDuration = time.Since(ndt.connectStart)
+			}
+			ndt.connectAddr = addr
+			ndt.connectErr = err
+		},
+	}
+
+	return httptrace.WithClientTrace(ctx, trace), ndt
+}
+
+func (d *dialTracer) endNetworkDial(ndt *networkDialTrace, dialErr error) {
+	total := time.Since(ndt.startedAt)
+
+	if dialErr != nil {
+		fields := []tag.Tag{
+			tag.NewDurationTag("totalDuration", total),
+			tag.Error(dialErr),
+			tag.ErrorType(dialErr),
+			tag.NewDurationTag("connectDuration", ndt.connectDuration),
+			tag.NewStringTag("connectAddr", ndt.connectAddr),
+		}
+		if ndt.connectErr != nil {
+			fields = append(fields, tag.NewStringTag("connectErr", ndt.connectErr.Error()))
+		}
+		d.logger.Warn("network dial error", fields...)
+		metrics.ServiceDialErrorCount.With(d.metricsHandler).Record(1)
+	} else {
+		metrics.ServiceDialSuccessCount.With(d.metricsHandler).Record(1)
+	}
+
+	if ndt.connectDuration > 0 {
+		metrics.ServiceDialLatency.With(d.metricsHandler).Record(ndt.connectDuration)
+	}
+}
+
+type networkDialTrace struct {
+	startedAt time.Time
+
+	connectStart    time.Time
+	connectDuration time.Duration
+	connectAddr     string
+	connectErr      error
+}

common/rpc/dial_tracer_test.go

@@ -0,0 +1,86 @@
+package rpc
+
+import (
+	"context"
+	"net/http/httptrace"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/metrics"
+)
+
+func TestDialTracer_DialSuccess(t *testing.T) {
+	t.Parallel()
+
+	addr := "127.0.0.1:1234"
+	dt := newDialTracer(addr, metrics.NoopMetricsHandler, log.NewNoopLogger())
+
+	ctx := context.Background()
+	ctx, ndt := dt.beginNetworkDial(ctx)
+
+	trace := httptrace.ContextClientTrace(ctx)
+	require.NotNil(t, trace)
+
+	trace.ConnectStart("tcp", addr)
+	rewind(&ndt.connectStart)
+	trace.ConnectDone("tcp", addr, nil)
+
+	dt.endNetworkDial(ndt, nil)
+
+	assert.Positive(t, ndt.connectDuration)
+	assert.Equal(t, addr, ndt.connectAddr)
+	assert.Nil(t, ndt.connectErr)
+}
+
+func TestDialTracer_DialError(t *testing.T) {
+	t.Parallel()
+
+	addr := "127.0.0.1:1234"
+	dt := newDialTracer(addr, metrics.NoopMetricsHandler, log.NewNoopLogger())
+
+	ctx := context.Background()
+	ctx, ndt := dt.beginNetworkDial(ctx)
+
+	trace := httptrace.ContextClientTrace(ctx)
+	require.NotNil(t, trace)
+
+	connectErr := assert.AnError
+	trace.ConnectStart("tcp", addr)
+	rewind(&ndt.connectStart)
+	trace.ConnectDone("tcp", addr, connectErr)
+
+	dt.endNetworkDial(ndt, nil)
+
+	assert.Positive(t, ndt.connectDuration)
+	assert.Equal(t, addr, ndt.connectAddr)
+	assert.Equal(t, connectErr, ndt.connectErr)
+}
+
+func TestDialTracer_NoConnectStart(t *testing.T) {
+	t.Parallel()
+
+	addr := "127.0.0.1:1234"
+	dt := newDialTracer(addr, metrics.NoopMetricsHandler, log.NewNoopLogger())
+
+	ctx := context.Background()
+	ctx, ndt := dt.beginNetworkDial(ctx)
+
+	trace := httptrace.ContextClientTrace(ctx)
+	require.NotNil(t, trace)
+
+	trace.ConnectDone("tcp", addr, nil)
+
+	dt.endNetworkDial(ndt, nil)
+
+	assert.Zero(t, ndt.connectDuration)
+	assert.Equal(t, addr, ndt.connectAddr)
+	assert.Nil(t, ndt.connectErr)
+}
+
+// rewind moves a timestamp back by 1 millisecond to ensure positive durations in tests
+func rewind(ts *time.Time) {
+	*ts = ts.Add(-time.Millisecond)
+}

common/rpc/grpc.go

@@ -3,6 +3,7 @@ package rpc
 import (
 	"context"
 	"crypto/tls"
+	"net"
 	"time"
 
 	"go.temporal.io/server/common/headers"
@@ -46,7 +47,13 @@ const (
 // The hostName syntax is defined in
 // https://github.com/grpc/grpc/blob/master/doc/naming.md.
 // dns resolver is used by default
-func Dial(hostName string, tlsConfig *tls.Config, logger log.Logger, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
+func Dial(
+	hostName string,
+	tlsConfig *tls.Config,
+	logger log.Logger,
+	metricsHandler metrics.Handler,
+	opts ...grpc.DialOption,
+) (*grpc.ClientConn, error) {
 	var grpcSecureOpt grpc.DialOption
 	if tlsConfig == nil {
 		grpcSecureOpt = grpc.WithTransportCredentials(insecure.NewCredentials())
@@ -65,8 +72,27 @@ func Dial(hostName string, tlsConfig *tls.Config, logger log.Logger, opts ...grp
 	}
 	cp.Backoff.MaxDelay = MaxBackoffDelay
 
+	dtrace := newDialTracer(hostName, metricsHandler, logger)
+
+	contextDialer := func(ctx context.Context, s string) (net.Conn, error) {
+		// Keep the existing gRPC behavior by using OS defaults for TCP keepalive settings.
+		// We are on Go 1.23+ and can use KeepAliveConfig directly instead of the old KeepAlive/Control hacks.
+		dialer := &net.Dialer{
+			KeepAliveConfig: net.KeepAliveConfig{
+				Enable: true,
+			},
+		}
+
+		var ndt *networkDialTrace
+		ctx, ndt = dtrace.beginNetworkDial(ctx)
+		conn, dialErr := dialer.DialContext(ctx, "tcp", s)
+		dtrace.endNetworkDial(ndt, dialErr)
+		return conn, dialErr
+	}
+
 	dialOptions := []grpc.DialOption{
 		grpcSecureOpt,
+		grpc.WithContextDialer(contextDialer),
 		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxInternodeRecvPayloadSize)),
 		grpc.WithChainUnaryInterceptor(
 			headersInterceptor,

common/rpc/rpc.go

@@ -19,6 +19,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/membership"
+	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/primitives"
 	"go.temporal.io/server/common/rpc/encryption"
 	"go.temporal.io/server/temporal/environment"
@@ -31,9 +32,10 @@ var _ common.RPCFactory = (*RPCFactory)(nil)
 
 // RPCFactory is an implementation of common.RPCFactory interface
 type RPCFactory struct {
-	config      *config.Config
-	serviceName primitives.ServiceName
-	logger      log.Logger
+	config         *config.Config
+	serviceName    primitives.ServiceName
+	logger         log.Logger
+	metricsHandler metrics.Handler
 
 	frontendURL       string
 	frontendHTTPURL   string
@@ -59,6 +61,7 @@ func NewFactory(
 	cfg *config.Config,
 	sName primitives.ServiceName,
 	logger log.Logger,
+	metricsHandler metrics.Handler,
 	tlsProvider encryption.TLSConfigProvider,
 	frontendURL string,
 	frontendHTTPURL string,
@@ -71,6 +74,7 @@ func NewFactory(
 		config:            cfg,
 		serviceName:       sName,
 		logger:            logger,
+		metricsHandler:    metricsHandler,
 		frontendURL:       frontendURL,
 		frontendHTTPURL:   frontendHTTPURL,
 		frontendHTTPPort:  frontendHTTPPort,
@@ -248,7 +252,7 @@ func (d *RPCFactory) CreateMatchingGRPCConnection(rpcAddress string) *grpc.Clien
 
 func (d *RPCFactory) dial(hostName string, tlsClientConfig *tls.Config, dialOptions ...grpc.DialOption) *grpc.ClientConn {
 	dialOptions = append(d.dialOptions, dialOptions...)
-	connection, err := Dial(hostName, tlsClientConfig, d.logger, dialOptions...)
+	connection, err := Dial(hostName, tlsClientConfig, d.logger, d.metricsHandler, dialOptions...)
 	if err != nil {
 		d.logger.Fatal("Failed to create gRPC connection", tag.Error(err))
 		return nil

common/rpc/test/http_test.go

@@ -34,6 +34,7 @@ func TestCreateLocalFrontendHTTPClient_UsingMembership(t *testing.T) {
 		nil,
 		primitives.HistoryService,
 		nil, // No logger
+		nil, // No metrics handler
 		nil,
 		membership.GRPCResolverURLForTesting(monitor, primitives.FrontendService),
 		membership.GRPCResolverURLForTesting(monitor, primitives.FrontendService),
@@ -64,6 +65,7 @@ func TestCreateLocalFrontendHTTPClient_UsingFixedHostPort(t *testing.T) {
 		nil, // unused
 		primitives.HistoryService,
 		nil, // No logger
+		nil, // No metrics handler
 		nil,
 		membership.GRPCResolverURLForTesting(nil, primitives.FrontendService),
 		addr.String(),
@@ -95,6 +97,7 @@ func TestCreateLocalFrontendHTTPClient_UsingFixedHostPort_AndTLS(t *testing.T) {
 		nil, // unused
 		primitives.HistoryService,
 		nil, // No logger
+		nil, // No metrics handler
 		nil,
 		membership.GRPCResolverURLForTesting(nil, primitives.FrontendService),
 		addr.String(),

common/rpc/test/rpc_common_test.go

@@ -157,7 +157,7 @@ func dialTestServiceAndGetTLSInfo(
 		s.NoError(err)
 	}
 
-	clientConn, err := rpc.Dial(hostport, cfg, logger)
+	clientConn, err := rpc.Dial(hostport, cfg, logger, nil)
 	s.NoError(err)
 
 	client := testservice.NewTestServiceClient(clientConn)