Source file src/google.golang.org/grpc/clientconn.go

Documentation: google.golang.org/grpc

     1  /*
     2   *
     3   * Copyright 2014 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package grpc
    20  
    21  import (
    22  	"context"
    23  	"errors"
    24  	"fmt"
    25  	"math"
    26  	"net/url"
    27  	"strings"
    28  	"sync"
    29  	"sync/atomic"
    30  	"time"
    31  
    32  	"google.golang.org/grpc/balancer"
    33  	"google.golang.org/grpc/balancer/base"
    34  	"google.golang.org/grpc/codes"
    35  	"google.golang.org/grpc/connectivity"
    36  	"google.golang.org/grpc/internal"
    37  	"google.golang.org/grpc/internal/channelz"
    38  	"google.golang.org/grpc/internal/grpcsync"
    39  	"google.golang.org/grpc/internal/idle"
    40  	iresolver "google.golang.org/grpc/internal/resolver"
    41  	"google.golang.org/grpc/internal/transport"
    42  	"google.golang.org/grpc/keepalive"
    43  	"google.golang.org/grpc/resolver"
    44  	"google.golang.org/grpc/serviceconfig"
    45  	"google.golang.org/grpc/status"
    46  
    47  	_ "google.golang.org/grpc/balancer/roundrobin"           // To register roundrobin.
    48  	_ "google.golang.org/grpc/internal/resolver/passthrough" // To register passthrough resolver.
    49  	_ "google.golang.org/grpc/internal/resolver/unix"        // To register unix resolver.
    50  	_ "google.golang.org/grpc/resolver/dns"                  // To register dns resolver.
    51  )
    52  
    53  const (
    54  	// minimum time to give a connection to complete
    55  	minConnectTimeout = 20 * time.Second
    56  )
    57  
    58  var (
    59  	// ErrClientConnClosing indicates that the operation is illegal because
    60  	// the ClientConn is closing.
    61  	//
    62  	// Deprecated: this error should not be relied upon by users; use the status
    63  	// code of Canceled instead.
    64  	ErrClientConnClosing = status.Error(codes.Canceled, "grpc: the client connection is closing")
    65  	// errConnDrain indicates that the connection starts to be drained and does not accept any new RPCs.
    66  	errConnDrain = errors.New("grpc: the connection is drained")
    67  	// errConnClosing indicates that the connection is closing.
    68  	errConnClosing = errors.New("grpc: the connection is closing")
    69  	// errConnIdling indicates the connection is being closed as the channel
    70  	// is moving to an idle mode due to inactivity.
    71  	errConnIdling = errors.New("grpc: the connection is closing due to channel idleness")
    72  	// invalidDefaultServiceConfigErrPrefix is used to prefix the json parsing error for the default
    73  	// service config.
    74  	invalidDefaultServiceConfigErrPrefix = "grpc: the provided default service config is invalid"
    75  )
    76  
    77  // The following errors are returned from Dial and DialContext
    78  var (
    79  	// errNoTransportSecurity indicates that there is no transport security
    80  	// being set for ClientConn. Users should either set one or explicitly
    81  	// call WithInsecure DialOption to disable security.
    82  	errNoTransportSecurity = errors.New("grpc: no transport security set (use grpc.WithTransportCredentials(insecure.NewCredentials()) explicitly or set credentials)")
    83  	// errTransportCredsAndBundle indicates that creds bundle is used together
    84  	// with other individual Transport Credentials.
    85  	errTransportCredsAndBundle = errors.New("grpc: credentials.Bundle may not be used with individual TransportCredentials")
    86  	// errNoTransportCredsInBundle indicated that the configured creds bundle
    87  	// returned a transport credentials which was nil.
    88  	errNoTransportCredsInBundle = errors.New("grpc: credentials.Bundle must return non-nil transport credentials")
    89  	// errTransportCredentialsMissing indicates that users want to transmit
    90  	// security information (e.g., OAuth2 token) which requires secure
    91  	// connection on an insecure connection.
    92  	errTransportCredentialsMissing = errors.New("grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)")
    93  )
    94  
    95  const (
    96  	defaultClientMaxReceiveMessageSize = 1024 * 1024 * 4
    97  	defaultClientMaxSendMessageSize    = math.MaxInt32
    98  	// http2IOBufSize specifies the buffer size for sending frames.
    99  	defaultWriteBufSize = 32 * 1024
   100  	defaultReadBufSize  = 32 * 1024
   101  )
   102  
   103  type defaultConfigSelector struct {
   104  	sc *ServiceConfig
   105  }
   106  
   107  func (dcs *defaultConfigSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RPCConfig, error) {
   108  	return &iresolver.RPCConfig{
   109  		Context:      rpcInfo.Context,
   110  		MethodConfig: getMethodConfig(dcs.sc, rpcInfo.Method),
   111  	}, nil
   112  }
   113  
   114  // NewClient creates a new gRPC "channel" for the target URI provided.  No I/O
   115  // is performed.  Use of the ClientConn for RPCs will automatically cause it to
   116  // connect.  Connect may be used to manually create a connection, but for most
   117  // users this is unnecessary.
   118  //
   119  // The target name syntax is defined in
   120  // https://github.com/grpc/grpc/blob/master/doc/naming.md.  e.g. to use dns
   121  // resolver, a "dns:///" prefix should be applied to the target.
   122  //
   123  // The DialOptions returned by WithBlock, WithTimeout,
   124  // WithReturnConnectionError, and FailOnNonTempDialError are ignored by this
   125  // function.
   126  func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error) {
   127  	cc := &ClientConn{
   128  		target: target,
   129  		conns:  make(map[*addrConn]struct{}),
   130  		dopts:  defaultDialOptions(),
   131  	}
   132  
   133  	cc.retryThrottler.Store((*retryThrottler)(nil))
   134  	cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil})
   135  	cc.ctx, cc.cancel = context.WithCancel(context.Background())
   136  
   137  	// Apply dial options.
   138  	disableGlobalOpts := false
   139  	for _, opt := range opts {
   140  		if _, ok := opt.(*disableGlobalDialOptions); ok {
   141  			disableGlobalOpts = true
   142  			break
   143  		}
   144  	}
   145  
   146  	if !disableGlobalOpts {
   147  		for _, opt := range globalDialOptions {
   148  			opt.apply(&cc.dopts)
   149  		}
   150  	}
   151  
   152  	for _, opt := range opts {
   153  		opt.apply(&cc.dopts)
   154  	}
   155  	chainUnaryClientInterceptors(cc)
   156  	chainStreamClientInterceptors(cc)
   157  
   158  	if err := cc.validateTransportCredentials(); err != nil {
   159  		return nil, err
   160  	}
   161  
   162  	if cc.dopts.defaultServiceConfigRawJSON != nil {
   163  		scpr := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON)
   164  		if scpr.Err != nil {
   165  			return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, scpr.Err)
   166  		}
   167  		cc.dopts.defaultServiceConfig, _ = scpr.Config.(*ServiceConfig)
   168  	}
   169  	cc.mkp = cc.dopts.copts.KeepaliveParams
   170  
   171  	// Register ClientConn with channelz.
   172  	cc.channelzRegistration(target)
   173  
   174  	// TODO: Ideally it should be impossible to error from this function after
   175  	// channelz registration.  This will require removing some channelz logs
   176  	// from the following functions that can error.  Errors can be returned to
   177  	// the user, and successful logs can be emitted here, after the checks have
   178  	// passed and channelz is subsequently registered.
   179  
   180  	// Determine the resolver to use.
   181  	if err := cc.parseTargetAndFindResolver(); err != nil {
   182  		channelz.RemoveEntry(cc.channelz.ID)
   183  		return nil, err
   184  	}
   185  	if err = cc.determineAuthority(); err != nil {
   186  		channelz.RemoveEntry(cc.channelz.ID)
   187  		return nil, err
   188  	}
   189  
   190  	cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz)
   191  	cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers)
   192  
   193  	cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc.
   194  	cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout)
   195  	return cc, nil
   196  }
   197  
   198  // Dial calls DialContext(context.Background(), target, opts...).
   199  //
   200  // Deprecated: use NewClient instead.  Will be supported throughout 1.x.
   201  func Dial(target string, opts ...DialOption) (*ClientConn, error) {
   202  	return DialContext(context.Background(), target, opts...)
   203  }
   204  
   205  // DialContext calls NewClient and then exits idle mode.  If WithBlock(true) is
   206  // used, it calls Connect and WaitForStateChange until either the context
   207  // expires or the state of the ClientConn is Ready.
   208  //
   209  // One subtle difference between NewClient and Dial and DialContext is that the
   210  // former uses "dns" as the default name resolver, while the latter use
   211  // "passthrough" for backward compatibility.  This distinction should not matter
   212  // to most users, but could matter to legacy users that specify a custom dialer
   213  // and expect it to receive the target string directly.
   214  //
   215  // Deprecated: use NewClient instead.  Will be supported throughout 1.x.
   216  func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
   217  	// At the end of this method, we kick the channel out of idle, rather than
   218  	// waiting for the first rpc.
   219  	opts = append([]DialOption{withDefaultScheme("passthrough")}, opts...)
   220  	cc, err := NewClient(target, opts...)
   221  	if err != nil {
   222  		return nil, err
   223  	}
   224  
   225  	// We start the channel off in idle mode, but kick it out of idle now,
   226  	// instead of waiting for the first RPC.  This is the legacy behavior of
   227  	// Dial.
   228  	defer func() {
   229  		if err != nil {
   230  			cc.Close()
   231  		}
   232  	}()
   233  
   234  	// This creates the name resolver, load balancer, etc.
   235  	if err := cc.idlenessMgr.ExitIdleMode(); err != nil {
   236  		return nil, err
   237  	}
   238  
   239  	// Return now for non-blocking dials.
   240  	if !cc.dopts.block {
   241  		return cc, nil
   242  	}
   243  
   244  	if cc.dopts.timeout > 0 {
   245  		var cancel context.CancelFunc
   246  		ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout)
   247  		defer cancel()
   248  	}
   249  	defer func() {
   250  		select {
   251  		case <-ctx.Done():
   252  			switch {
   253  			case ctx.Err() == err:
   254  				conn = nil
   255  			case err == nil || !cc.dopts.returnLastError:
   256  				conn, err = nil, ctx.Err()
   257  			default:
   258  				conn, err = nil, fmt.Errorf("%v: %v", ctx.Err(), err)
   259  			}
   260  		default:
   261  		}
   262  	}()
   263  
   264  	// A blocking dial blocks until the clientConn is ready.
   265  	for {
   266  		s := cc.GetState()
   267  		if s == connectivity.Idle {
   268  			cc.Connect()
   269  		}
   270  		if s == connectivity.Ready {
   271  			return cc, nil
   272  		} else if cc.dopts.copts.FailOnNonTempDialError && s == connectivity.TransientFailure {
   273  			if err = cc.connectionError(); err != nil {
   274  				terr, ok := err.(interface {
   275  					Temporary() bool
   276  				})
   277  				if ok && !terr.Temporary() {
   278  					return nil, err
   279  				}
   280  			}
   281  		}
   282  		if !cc.WaitForStateChange(ctx, s) {
   283  			// ctx got timeout or canceled.
   284  			if err = cc.connectionError(); err != nil && cc.dopts.returnLastError {
   285  				return nil, err
   286  			}
   287  			return nil, ctx.Err()
   288  		}
   289  	}
   290  }
   291  
   292  // addTraceEvent is a helper method to add a trace event on the channel. If the
   293  // channel is a nested one, the same event is also added on the parent channel.
   294  func (cc *ClientConn) addTraceEvent(msg string) {
   295  	ted := &channelz.TraceEvent{
   296  		Desc:     fmt.Sprintf("Channel %s", msg),
   297  		Severity: channelz.CtInfo,
   298  	}
   299  	if cc.dopts.channelzParent != nil {
   300  		ted.Parent = &channelz.TraceEvent{
   301  			Desc:     fmt.Sprintf("Nested channel(id:%d) %s", cc.channelz.ID, msg),
   302  			Severity: channelz.CtInfo,
   303  		}
   304  	}
   305  	channelz.AddTraceEvent(logger, cc.channelz, 0, ted)
   306  }
   307  
   308  type idler ClientConn
   309  
   310  func (i *idler) EnterIdleMode() {
   311  	(*ClientConn)(i).enterIdleMode()
   312  }
   313  
   314  func (i *idler) ExitIdleMode() error {
   315  	return (*ClientConn)(i).exitIdleMode()
   316  }
   317  
   318  // exitIdleMode moves the channel out of idle mode by recreating the name
   319  // resolver and load balancer.  This should never be called directly; use
   320  // cc.idlenessMgr.ExitIdleMode instead.
   321  func (cc *ClientConn) exitIdleMode() (err error) {
   322  	cc.mu.Lock()
   323  	if cc.conns == nil {
   324  		cc.mu.Unlock()
   325  		return errConnClosing
   326  	}
   327  	cc.mu.Unlock()
   328  
   329  	// This needs to be called without cc.mu because this builds a new resolver
   330  	// which might update state or report error inline, which would then need to
   331  	// acquire cc.mu.
   332  	if err := cc.resolverWrapper.start(); err != nil {
   333  		return err
   334  	}
   335  
   336  	cc.addTraceEvent("exiting idle mode")
   337  	return nil
   338  }
   339  
   340  // initIdleStateLocked initializes common state to how it should be while idle.
   341  func (cc *ClientConn) initIdleStateLocked() {
   342  	cc.resolverWrapper = newCCResolverWrapper(cc)
   343  	cc.balancerWrapper = newCCBalancerWrapper(cc)
   344  	cc.firstResolveEvent = grpcsync.NewEvent()
   345  	// cc.conns == nil is a proxy for the ClientConn being closed. So, instead
   346  	// of setting it to nil here, we recreate the map. This also means that we
   347  	// don't have to do this when exiting idle mode.
   348  	cc.conns = make(map[*addrConn]struct{})
   349  }
   350  
   351  // enterIdleMode puts the channel in idle mode, and as part of it shuts down the
   352  // name resolver, load balancer, and any subchannels.  This should never be
   353  // called directly; use cc.idlenessMgr.EnterIdleMode instead.
   354  func (cc *ClientConn) enterIdleMode() {
   355  	cc.mu.Lock()
   356  
   357  	if cc.conns == nil {
   358  		cc.mu.Unlock()
   359  		return
   360  	}
   361  
   362  	conns := cc.conns
   363  
   364  	rWrapper := cc.resolverWrapper
   365  	rWrapper.close()
   366  	cc.pickerWrapper.reset()
   367  	bWrapper := cc.balancerWrapper
   368  	bWrapper.close()
   369  	cc.csMgr.updateState(connectivity.Idle)
   370  	cc.addTraceEvent("entering idle mode")
   371  
   372  	cc.initIdleStateLocked()
   373  
   374  	cc.mu.Unlock()
   375  
   376  	// Block until the name resolver and LB policy are closed.
   377  	<-rWrapper.serializer.Done()
   378  	<-bWrapper.serializer.Done()
   379  
   380  	// Close all subchannels after the LB policy is closed.
   381  	for ac := range conns {
   382  		ac.tearDown(errConnIdling)
   383  	}
   384  }
   385  
   386  // validateTransportCredentials performs a series of checks on the configured
   387  // transport credentials. It returns a non-nil error if any of these conditions
   388  // are met:
   389  //   - no transport creds and no creds bundle is configured
   390  //   - both transport creds and creds bundle are configured
   391  //   - creds bundle is configured, but it lacks a transport credentials
   392  //   - insecure transport creds configured alongside call creds that require
   393  //     transport level security
   394  //
   395  // If none of the above conditions are met, the configured credentials are
   396  // deemed valid and a nil error is returned.
   397  func (cc *ClientConn) validateTransportCredentials() error {
   398  	if cc.dopts.copts.TransportCredentials == nil && cc.dopts.copts.CredsBundle == nil {
   399  		return errNoTransportSecurity
   400  	}
   401  	if cc.dopts.copts.TransportCredentials != nil && cc.dopts.copts.CredsBundle != nil {
   402  		return errTransportCredsAndBundle
   403  	}
   404  	if cc.dopts.copts.CredsBundle != nil && cc.dopts.copts.CredsBundle.TransportCredentials() == nil {
   405  		return errNoTransportCredsInBundle
   406  	}
   407  	transportCreds := cc.dopts.copts.TransportCredentials
   408  	if transportCreds == nil {
   409  		transportCreds = cc.dopts.copts.CredsBundle.TransportCredentials()
   410  	}
   411  	if transportCreds.Info().SecurityProtocol == "insecure" {
   412  		for _, cd := range cc.dopts.copts.PerRPCCredentials {
   413  			if cd.RequireTransportSecurity() {
   414  				return errTransportCredentialsMissing
   415  			}
   416  		}
   417  	}
   418  	return nil
   419  }
   420  
   421  // channelzRegistration registers the newly created ClientConn with channelz and
   422  // stores the returned identifier in `cc.channelz`.  A channelz trace event is
   423  // emitted for ClientConn creation. If the newly created ClientConn is a nested
   424  // one, i.e a valid parent ClientConn ID is specified via a dial option, the
   425  // trace event is also added to the parent.
   426  //
   427  // Doesn't grab cc.mu as this method is expected to be called only at Dial time.
   428  func (cc *ClientConn) channelzRegistration(target string) {
   429  	parentChannel, _ := cc.dopts.channelzParent.(*channelz.Channel)
   430  	cc.channelz = channelz.RegisterChannel(parentChannel, target)
   431  	cc.addTraceEvent("created")
   432  }
   433  
   434  // chainUnaryClientInterceptors chains all unary client interceptors into one.
   435  func chainUnaryClientInterceptors(cc *ClientConn) {
   436  	interceptors := cc.dopts.chainUnaryInts
   437  	// Prepend dopts.unaryInt to the chaining interceptors if it exists, since unaryInt will
   438  	// be executed before any other chained interceptors.
   439  	if cc.dopts.unaryInt != nil {
   440  		interceptors = append([]UnaryClientInterceptor{cc.dopts.unaryInt}, interceptors...)
   441  	}
   442  	var chainedInt UnaryClientInterceptor
   443  	if len(interceptors) == 0 {
   444  		chainedInt = nil
   445  	} else if len(interceptors) == 1 {
   446  		chainedInt = interceptors[0]
   447  	} else {
   448  		chainedInt = func(ctx context.Context, method string, req, reply any, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error {
   449  			return interceptors[0](ctx, method, req, reply, cc, getChainUnaryInvoker(interceptors, 0, invoker), opts...)
   450  		}
   451  	}
   452  	cc.dopts.unaryInt = chainedInt
   453  }
   454  
   455  // getChainUnaryInvoker recursively generate the chained unary invoker.
   456  func getChainUnaryInvoker(interceptors []UnaryClientInterceptor, curr int, finalInvoker UnaryInvoker) UnaryInvoker {
   457  	if curr == len(interceptors)-1 {
   458  		return finalInvoker
   459  	}
   460  	return func(ctx context.Context, method string, req, reply any, cc *ClientConn, opts ...CallOption) error {
   461  		return interceptors[curr+1](ctx, method, req, reply, cc, getChainUnaryInvoker(interceptors, curr+1, finalInvoker), opts...)
   462  	}
   463  }
   464  
   465  // chainStreamClientInterceptors chains all stream client interceptors into one.
   466  func chainStreamClientInterceptors(cc *ClientConn) {
   467  	interceptors := cc.dopts.chainStreamInts
   468  	// Prepend dopts.streamInt to the chaining interceptors if it exists, since streamInt will
   469  	// be executed before any other chained interceptors.
   470  	if cc.dopts.streamInt != nil {
   471  		interceptors = append([]StreamClientInterceptor{cc.dopts.streamInt}, interceptors...)
   472  	}
   473  	var chainedInt StreamClientInterceptor
   474  	if len(interceptors) == 0 {
   475  		chainedInt = nil
   476  	} else if len(interceptors) == 1 {
   477  		chainedInt = interceptors[0]
   478  	} else {
   479  		chainedInt = func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error) {
   480  			return interceptors[0](ctx, desc, cc, method, getChainStreamer(interceptors, 0, streamer), opts...)
   481  		}
   482  	}
   483  	cc.dopts.streamInt = chainedInt
   484  }
   485  
   486  // getChainStreamer recursively generate the chained client stream constructor.
   487  func getChainStreamer(interceptors []StreamClientInterceptor, curr int, finalStreamer Streamer) Streamer {
   488  	if curr == len(interceptors)-1 {
   489  		return finalStreamer
   490  	}
   491  	return func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (ClientStream, error) {
   492  		return interceptors[curr+1](ctx, desc, cc, method, getChainStreamer(interceptors, curr+1, finalStreamer), opts...)
   493  	}
   494  }
   495  
   496  // newConnectivityStateManager creates an connectivityStateManager with
   497  // the specified channel.
   498  func newConnectivityStateManager(ctx context.Context, channel *channelz.Channel) *connectivityStateManager {
   499  	return &connectivityStateManager{
   500  		channelz: channel,
   501  		pubSub:   grpcsync.NewPubSub(ctx),
   502  	}
   503  }
   504  
   505  // connectivityStateManager keeps the connectivity.State of ClientConn.
   506  // This struct will eventually be exported so the balancers can access it.
   507  //
   508  // TODO: If possible, get rid of the `connectivityStateManager` type, and
   509  // provide this functionality using the `PubSub`, to avoid keeping track of
   510  // the connectivity state at two places.
   511  type connectivityStateManager struct {
   512  	mu         sync.Mutex
   513  	state      connectivity.State
   514  	notifyChan chan struct{}
   515  	channelz   *channelz.Channel
   516  	pubSub     *grpcsync.PubSub
   517  }
   518  
   519  // updateState updates the connectivity.State of ClientConn.
   520  // If there's a change it notifies goroutines waiting on state change to
   521  // happen.
   522  func (csm *connectivityStateManager) updateState(state connectivity.State) {
   523  	csm.mu.Lock()
   524  	defer csm.mu.Unlock()
   525  	if csm.state == connectivity.Shutdown {
   526  		return
   527  	}
   528  	if csm.state == state {
   529  		return
   530  	}
   531  	csm.state = state
   532  	csm.channelz.ChannelMetrics.State.Store(&state)
   533  	csm.pubSub.Publish(state)
   534  
   535  	channelz.Infof(logger, csm.channelz, "Channel Connectivity change to %v", state)
   536  	if csm.notifyChan != nil {
   537  		// There are other goroutines waiting on this channel.
   538  		close(csm.notifyChan)
   539  		csm.notifyChan = nil
   540  	}
   541  }
   542  
   543  func (csm *connectivityStateManager) getState() connectivity.State {
   544  	csm.mu.Lock()
   545  	defer csm.mu.Unlock()
   546  	return csm.state
   547  }
   548  
   549  func (csm *connectivityStateManager) getNotifyChan() <-chan struct{} {
   550  	csm.mu.Lock()
   551  	defer csm.mu.Unlock()
   552  	if csm.notifyChan == nil {
   553  		csm.notifyChan = make(chan struct{})
   554  	}
   555  	return csm.notifyChan
   556  }
   557  
   558  // ClientConnInterface defines the functions clients need to perform unary and
   559  // streaming RPCs.  It is implemented by *ClientConn, and is only intended to
   560  // be referenced by generated code.
   561  type ClientConnInterface interface {
   562  	// Invoke performs a unary RPC and returns after the response is received
   563  	// into reply.
   564  	Invoke(ctx context.Context, method string, args any, reply any, opts ...CallOption) error
   565  	// NewStream begins a streaming RPC.
   566  	NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error)
   567  }
   568  
   569  // Assert *ClientConn implements ClientConnInterface.
   570  var _ ClientConnInterface = (*ClientConn)(nil)
   571  
   572  // ClientConn represents a virtual connection to a conceptual endpoint, to
   573  // perform RPCs.
   574  //
   575  // A ClientConn is free to have zero or more actual connections to the endpoint
   576  // based on configuration, load, etc. It is also free to determine which actual
   577  // endpoints to use and may change it every RPC, permitting client-side load
   578  // balancing.
   579  //
   580  // A ClientConn encapsulates a range of functionality including name
   581  // resolution, TCP connection establishment (with retries and backoff) and TLS
   582  // handshakes. It also handles errors on established connections by
   583  // re-resolving the name and reconnecting.
   584  type ClientConn struct {
   585  	ctx    context.Context    // Initialized using the background context at dial time.
   586  	cancel context.CancelFunc // Cancelled on close.
   587  
   588  	// The following are initialized at dial time, and are read-only after that.
   589  	target          string            // User's dial target.
   590  	parsedTarget    resolver.Target   // See parseTargetAndFindResolver().
   591  	authority       string            // See determineAuthority().
   592  	dopts           dialOptions       // Default and user specified dial options.
   593  	channelz        *channelz.Channel // Channelz object.
   594  	resolverBuilder resolver.Builder  // See parseTargetAndFindResolver().
   595  	idlenessMgr     *idle.Manager
   596  
   597  	// The following provide their own synchronization, and therefore don't
   598  	// require cc.mu to be held to access them.
   599  	csMgr              *connectivityStateManager
   600  	pickerWrapper      *pickerWrapper
   601  	safeConfigSelector iresolver.SafeConfigSelector
   602  	retryThrottler     atomic.Value // Updated from service config.
   603  
   604  	// mu protects the following fields.
   605  	// TODO: split mu so the same mutex isn't used for everything.
   606  	mu              sync.RWMutex
   607  	resolverWrapper *ccResolverWrapper         // Always recreated whenever entering idle to simplify Close.
   608  	balancerWrapper *ccBalancerWrapper         // Always recreated whenever entering idle to simplify Close.
   609  	sc              *ServiceConfig             // Latest service config received from the resolver.
   610  	conns           map[*addrConn]struct{}     // Set to nil on close.
   611  	mkp             keepalive.ClientParameters // May be updated upon receipt of a GoAway.
   612  	// firstResolveEvent is used to track whether the name resolver sent us at
   613  	// least one update. RPCs block on this event.  May be accessed without mu
   614  	// if we know we cannot be asked to enter idle mode while accessing it (e.g.
   615  	// when the idle manager has already been closed, or if we are already
   616  	// entering idle mode).
   617  	firstResolveEvent *grpcsync.Event
   618  
   619  	lceMu               sync.Mutex // protects lastConnectionError
   620  	lastConnectionError error
   621  }
   622  
   623  // WaitForStateChange waits until the connectivity.State of ClientConn changes from sourceState or
   624  // ctx expires. A true value is returned in former case and false in latter.
   625  //
   626  // # Experimental
   627  //
   628  // Notice: This API is EXPERIMENTAL and may be changed or removed in a
   629  // later release.
   630  func (cc *ClientConn) WaitForStateChange(ctx context.Context, sourceState connectivity.State) bool {
   631  	ch := cc.csMgr.getNotifyChan()
   632  	if cc.csMgr.getState() != sourceState {
   633  		return true
   634  	}
   635  	select {
   636  	case <-ctx.Done():
   637  		return false
   638  	case <-ch:
   639  		return true
   640  	}
   641  }
   642  
   643  // GetState returns the connectivity.State of ClientConn.
   644  //
   645  // # Experimental
   646  //
   647  // Notice: This API is EXPERIMENTAL and may be changed or removed in a later
   648  // release.
   649  func (cc *ClientConn) GetState() connectivity.State {
   650  	return cc.csMgr.getState()
   651  }
   652  
   653  // Connect causes all subchannels in the ClientConn to attempt to connect if
   654  // the channel is idle.  Does not wait for the connection attempts to begin
   655  // before returning.
   656  //
   657  // # Experimental
   658  //
   659  // Notice: This API is EXPERIMENTAL and may be changed or removed in a later
   660  // release.
   661  func (cc *ClientConn) Connect() {
   662  	if err := cc.idlenessMgr.ExitIdleMode(); err != nil {
   663  		cc.addTraceEvent(err.Error())
   664  		return
   665  	}
   666  	// If the ClientConn was not in idle mode, we need to call ExitIdle on the
   667  	// LB policy so that connections can be created.
   668  	cc.mu.Lock()
   669  	cc.balancerWrapper.exitIdle()
   670  	cc.mu.Unlock()
   671  }
   672  
   673  // waitForResolvedAddrs blocks until the resolver has provided addresses or the
   674  // context expires.  Returns nil unless the context expires first; otherwise
   675  // returns a status error based on the context.
   676  func (cc *ClientConn) waitForResolvedAddrs(ctx context.Context) error {
   677  	// This is on the RPC path, so we use a fast path to avoid the
   678  	// more-expensive "select" below after the resolver has returned once.
   679  	if cc.firstResolveEvent.HasFired() {
   680  		return nil
   681  	}
   682  	select {
   683  	case <-cc.firstResolveEvent.Done():
   684  		return nil
   685  	case <-ctx.Done():
   686  		return status.FromContextError(ctx.Err()).Err()
   687  	case <-cc.ctx.Done():
   688  		return ErrClientConnClosing
   689  	}
   690  }
   691  
   692  var emptyServiceConfig *ServiceConfig
   693  
   694  func init() {
   695  	balancer.Register(pickfirstBuilder{})
   696  	cfg := parseServiceConfig("{}")
   697  	if cfg.Err != nil {
   698  		panic(fmt.Sprintf("impossible error parsing empty service config: %v", cfg.Err))
   699  	}
   700  	emptyServiceConfig = cfg.Config.(*ServiceConfig)
   701  
   702  	internal.SubscribeToConnectivityStateChanges = func(cc *ClientConn, s grpcsync.Subscriber) func() {
   703  		return cc.csMgr.pubSub.Subscribe(s)
   704  	}
   705  	internal.EnterIdleModeForTesting = func(cc *ClientConn) {
   706  		cc.idlenessMgr.EnterIdleModeForTesting()
   707  	}
   708  	internal.ExitIdleModeForTesting = func(cc *ClientConn) error {
   709  		return cc.idlenessMgr.ExitIdleMode()
   710  	}
   711  }
   712  
   713  func (cc *ClientConn) maybeApplyDefaultServiceConfig() {
   714  	if cc.sc != nil {
   715  		cc.applyServiceConfigAndBalancer(cc.sc, nil)
   716  		return
   717  	}
   718  	if cc.dopts.defaultServiceConfig != nil {
   719  		cc.applyServiceConfigAndBalancer(cc.dopts.defaultServiceConfig, &defaultConfigSelector{cc.dopts.defaultServiceConfig})
   720  	} else {
   721  		cc.applyServiceConfigAndBalancer(emptyServiceConfig, &defaultConfigSelector{emptyServiceConfig})
   722  	}
   723  }
   724  
   725  func (cc *ClientConn) updateResolverStateAndUnlock(s resolver.State, err error) error {
   726  	defer cc.firstResolveEvent.Fire()
   727  	// Check if the ClientConn is already closed. Some fields (e.g.
   728  	// balancerWrapper) are set to nil when closing the ClientConn, and could
   729  	// cause nil pointer panic if we don't have this check.
   730  	if cc.conns == nil {
   731  		cc.mu.Unlock()
   732  		return nil
   733  	}
   734  
   735  	if err != nil {
   736  		// May need to apply the initial service config in case the resolver
   737  		// doesn't support service configs, or doesn't provide a service config
   738  		// with the new addresses.
   739  		cc.maybeApplyDefaultServiceConfig()
   740  
   741  		cc.balancerWrapper.resolverError(err)
   742  
   743  		// No addresses are valid with err set; return early.
   744  		cc.mu.Unlock()
   745  		return balancer.ErrBadResolverState
   746  	}
   747  
   748  	var ret error
   749  	if cc.dopts.disableServiceConfig {
   750  		channelz.Infof(logger, cc.channelz, "ignoring service config from resolver (%v) and applying the default because service config is disabled", s.ServiceConfig)
   751  		cc.maybeApplyDefaultServiceConfig()
   752  	} else if s.ServiceConfig == nil {
   753  		cc.maybeApplyDefaultServiceConfig()
   754  		// TODO: do we need to apply a failing LB policy if there is no
   755  		// default, per the error handling design?
   756  	} else {
   757  		if sc, ok := s.ServiceConfig.Config.(*ServiceConfig); s.ServiceConfig.Err == nil && ok {
   758  			configSelector := iresolver.GetConfigSelector(s)
   759  			if configSelector != nil {
   760  				if len(s.ServiceConfig.Config.(*ServiceConfig).Methods) != 0 {
   761  					channelz.Infof(logger, cc.channelz, "method configs in service config will be ignored due to presence of config selector")
   762  				}
   763  			} else {
   764  				configSelector = &defaultConfigSelector{sc}
   765  			}
   766  			cc.applyServiceConfigAndBalancer(sc, configSelector)
   767  		} else {
   768  			ret = balancer.ErrBadResolverState
   769  			if cc.sc == nil {
   770  				// Apply the failing LB only if we haven't received valid service config
   771  				// from the name resolver in the past.
   772  				cc.applyFailingLBLocked(s.ServiceConfig)
   773  				cc.mu.Unlock()
   774  				return ret
   775  			}
   776  		}
   777  	}
   778  
   779  	var balCfg serviceconfig.LoadBalancingConfig
   780  	if cc.sc != nil && cc.sc.lbConfig != nil {
   781  		balCfg = cc.sc.lbConfig
   782  	}
   783  	bw := cc.balancerWrapper
   784  	cc.mu.Unlock()
   785  
   786  	uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})
   787  	if ret == nil {
   788  		ret = uccsErr // prefer ErrBadResolver state since any other error is
   789  		// currently meaningless to the caller.
   790  	}
   791  	return ret
   792  }
   793  
   794  // applyFailingLBLocked is akin to configuring an LB policy on the channel which
   795  // always fails RPCs. Here, an actual LB policy is not configured, but an always
   796  // erroring picker is configured, which returns errors with information about
   797  // what was invalid in the received service config. A config selector with no
   798  // service config is configured, and the connectivity state of the channel is
   799  // set to TransientFailure.
   800  func (cc *ClientConn) applyFailingLBLocked(sc *serviceconfig.ParseResult) {
   801  	var err error
   802  	if sc.Err != nil {
   803  		err = status.Errorf(codes.Unavailable, "error parsing service config: %v", sc.Err)
   804  	} else {
   805  		err = status.Errorf(codes.Unavailable, "illegal service config type: %T", sc.Config)
   806  	}
   807  	cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil})
   808  	cc.pickerWrapper.updatePicker(base.NewErrPicker(err))
   809  	cc.csMgr.updateState(connectivity.TransientFailure)
   810  }
   811  
   812  // Makes a copy of the input addresses slice and clears out the balancer
   813  // attributes field. Addresses are passed during subconn creation and address
   814  // update operations. In both cases, we will clear the balancer attributes by
   815  // calling this function, and therefore we will be able to use the Equal method
   816  // provided by the resolver.Address type for comparison.
   817  func copyAddressesWithoutBalancerAttributes(in []resolver.Address) []resolver.Address {
   818  	out := make([]resolver.Address, len(in))
   819  	for i := range in {
   820  		out[i] = in[i]
   821  		out[i].BalancerAttributes = nil
   822  	}
   823  	return out
   824  }
   825  
   826  // newAddrConnLocked creates an addrConn for addrs and adds it to cc.conns.
   827  //
   828  // Caller needs to make sure len(addrs) > 0.
   829  func (cc *ClientConn) newAddrConnLocked(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
   830  	if cc.conns == nil {
   831  		return nil, ErrClientConnClosing
   832  	}
   833  
   834  	ac := &addrConn{
   835  		state:        connectivity.Idle,
   836  		cc:           cc,
   837  		addrs:        copyAddressesWithoutBalancerAttributes(addrs),
   838  		scopts:       opts,
   839  		dopts:        cc.dopts,
   840  		channelz:     channelz.RegisterSubChannel(cc.channelz, ""),
   841  		resetBackoff: make(chan struct{}),
   842  		stateChan:    make(chan struct{}),
   843  	}
   844  	ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
   845  	// Start with our address set to the first address; this may be updated if
   846  	// we connect to different addresses.
   847  	ac.channelz.ChannelMetrics.Target.Store(&addrs[0].Addr)
   848  
   849  	channelz.AddTraceEvent(logger, ac.channelz, 0, &channelz.TraceEvent{
   850  		Desc:     "Subchannel created",
   851  		Severity: channelz.CtInfo,
   852  		Parent: &channelz.TraceEvent{
   853  			Desc:     fmt.Sprintf("Subchannel(id:%d) created", ac.channelz.ID),
   854  			Severity: channelz.CtInfo,
   855  		},
   856  	})
   857  
   858  	// Track ac in cc. This needs to be done before any getTransport(...) is called.
   859  	cc.conns[ac] = struct{}{}
   860  	return ac, nil
   861  }
   862  
   863  // removeAddrConn removes the addrConn in the subConn from clientConn.
   864  // It also tears down the ac with the given error.
   865  func (cc *ClientConn) removeAddrConn(ac *addrConn, err error) {
   866  	cc.mu.Lock()
   867  	if cc.conns == nil {
   868  		cc.mu.Unlock()
   869  		return
   870  	}
   871  	delete(cc.conns, ac)
   872  	cc.mu.Unlock()
   873  	ac.tearDown(err)
   874  }
   875  
   876  // Target returns the target string of the ClientConn.
   877  func (cc *ClientConn) Target() string {
   878  	return cc.target
   879  }
   880  
   881  // CanonicalTarget returns the canonical target string of the ClientConn.
   882  func (cc *ClientConn) CanonicalTarget() string {
   883  	return cc.parsedTarget.String()
   884  }
   885  
   886  func (cc *ClientConn) incrCallsStarted() {
   887  	cc.channelz.ChannelMetrics.CallsStarted.Add(1)
   888  	cc.channelz.ChannelMetrics.LastCallStartedTimestamp.Store(time.Now().UnixNano())
   889  }
   890  
   891  func (cc *ClientConn) incrCallsSucceeded() {
   892  	cc.channelz.ChannelMetrics.CallsSucceeded.Add(1)
   893  }
   894  
   895  func (cc *ClientConn) incrCallsFailed() {
   896  	cc.channelz.ChannelMetrics.CallsFailed.Add(1)
   897  }
   898  
   899  // connect starts creating a transport.
   900  // It does nothing if the ac is not IDLE.
   901  // TODO(bar) Move this to the addrConn section.
   902  func (ac *addrConn) connect() error {
   903  	ac.mu.Lock()
   904  	if ac.state == connectivity.Shutdown {
   905  		if logger.V(2) {
   906  			logger.Infof("connect called on shutdown addrConn; ignoring.")
   907  		}
   908  		ac.mu.Unlock()
   909  		return errConnClosing
   910  	}
   911  	if ac.state != connectivity.Idle {
   912  		if logger.V(2) {
   913  			logger.Infof("connect called on addrConn in non-idle state (%v); ignoring.", ac.state)
   914  		}
   915  		ac.mu.Unlock()
   916  		return nil
   917  	}
   918  	ac.mu.Unlock()
   919  
   920  	ac.resetTransport()
   921  	return nil
   922  }
   923  
   924  func equalAddresses(a, b []resolver.Address) bool {
   925  	if len(a) != len(b) {
   926  		return false
   927  	}
   928  	for i, v := range a {
   929  		if !v.Equal(b[i]) {
   930  			return false
   931  		}
   932  	}
   933  	return true
   934  }
   935  
   936  // updateAddrs updates ac.addrs with the new addresses list and handles active
   937  // connections or connection attempts.
   938  func (ac *addrConn) updateAddrs(addrs []resolver.Address) {
   939  	addrs = copyAddressesWithoutBalancerAttributes(addrs)
   940  	limit := len(addrs)
   941  	if limit > 5 {
   942  		limit = 5
   943  	}
   944  	channelz.Infof(logger, ac.channelz, "addrConn: updateAddrs addrs (%d of %d): %v", limit, len(addrs), addrs[:limit])
   945  
   946  	ac.mu.Lock()
   947  	if equalAddresses(ac.addrs, addrs) {
   948  		ac.mu.Unlock()
   949  		return
   950  	}
   951  
   952  	ac.addrs = addrs
   953  
   954  	if ac.state == connectivity.Shutdown ||
   955  		ac.state == connectivity.TransientFailure ||
   956  		ac.state == connectivity.Idle {
   957  		// We were not connecting, so do nothing but update the addresses.
   958  		ac.mu.Unlock()
   959  		return
   960  	}
   961  
   962  	if ac.state == connectivity.Ready {
   963  		// Try to find the connected address.
   964  		for _, a := range addrs {
   965  			a.ServerName = ac.cc.getServerName(a)
   966  			if a.Equal(ac.curAddr) {
   967  				// We are connected to a valid address, so do nothing but
   968  				// update the addresses.
   969  				ac.mu.Unlock()
   970  				return
   971  			}
   972  		}
   973  	}
   974  
   975  	// We are either connected to the wrong address or currently connecting.
   976  	// Stop the current iteration and restart.
   977  
   978  	ac.cancel()
   979  	ac.ctx, ac.cancel = context.WithCancel(ac.cc.ctx)
   980  
   981  	// We have to defer here because GracefulClose => onClose, which requires
   982  	// locking ac.mu.
   983  	if ac.transport != nil {
   984  		defer ac.transport.GracefulClose()
   985  		ac.transport = nil
   986  	}
   987  
   988  	if len(addrs) == 0 {
   989  		ac.updateConnectivityState(connectivity.Idle, nil)
   990  	}
   991  
   992  	ac.mu.Unlock()
   993  
   994  	// Since we were connecting/connected, we should start a new connection
   995  	// attempt.
   996  	go ac.resetTransport()
   997  }
   998  
   999  // getServerName determines the serverName to be used in the connection
  1000  // handshake. The default value for the serverName is the authority on the
  1001  // ClientConn, which either comes from the user's dial target or through an
  1002  // authority override specified using the WithAuthority dial option. Name
  1003  // resolvers can specify a per-address override for the serverName through the
  1004  // resolver.Address.ServerName field which is used only if the WithAuthority
  1005  // dial option was not used. The rationale is that per-address authority
  1006  // overrides specified by the name resolver can represent a security risk, while
  1007  // an override specified by the user is more dependable since they probably know
  1008  // what they are doing.
  1009  func (cc *ClientConn) getServerName(addr resolver.Address) string {
  1010  	if cc.dopts.authority != "" {
  1011  		return cc.dopts.authority
  1012  	}
  1013  	if addr.ServerName != "" {
  1014  		return addr.ServerName
  1015  	}
  1016  	return cc.authority
  1017  }
  1018  
  1019  func getMethodConfig(sc *ServiceConfig, method string) MethodConfig {
  1020  	if sc == nil {
  1021  		return MethodConfig{}
  1022  	}
  1023  	if m, ok := sc.Methods[method]; ok {
  1024  		return m
  1025  	}
  1026  	i := strings.LastIndex(method, "/")
  1027  	if m, ok := sc.Methods[method[:i+1]]; ok {
  1028  		return m
  1029  	}
  1030  	return sc.Methods[""]
  1031  }
  1032  
  1033  // GetMethodConfig gets the method config of the input method.
  1034  // If there's an exact match for input method (i.e. /service/method), we return
  1035  // the corresponding MethodConfig.
  1036  // If there isn't an exact match for the input method, we look for the service's default
  1037  // config under the service (i.e /service/) and then for the default for all services (empty string).
  1038  //
  1039  // If there is a default MethodConfig for the service, we return it.
  1040  // Otherwise, we return an empty MethodConfig.
  1041  func (cc *ClientConn) GetMethodConfig(method string) MethodConfig {
  1042  	// TODO: Avoid the locking here.
  1043  	cc.mu.RLock()
  1044  	defer cc.mu.RUnlock()
  1045  	return getMethodConfig(cc.sc, method)
  1046  }
  1047  
  1048  func (cc *ClientConn) healthCheckConfig() *healthCheckConfig {
  1049  	cc.mu.RLock()
  1050  	defer cc.mu.RUnlock()
  1051  	if cc.sc == nil {
  1052  		return nil
  1053  	}
  1054  	return cc.sc.healthCheckConfig
  1055  }
  1056  
  1057  func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, balancer.PickResult, error) {
  1058  	return cc.pickerWrapper.pick(ctx, failfast, balancer.PickInfo{
  1059  		Ctx:            ctx,
  1060  		FullMethodName: method,
  1061  	})
  1062  }
  1063  
  1064  func (cc *ClientConn) applyServiceConfigAndBalancer(sc *ServiceConfig, configSelector iresolver.ConfigSelector) {
  1065  	if sc == nil {
  1066  		// should never reach here.
  1067  		return
  1068  	}
  1069  	cc.sc = sc
  1070  	if configSelector != nil {
  1071  		cc.safeConfigSelector.UpdateConfigSelector(configSelector)
  1072  	}
  1073  
  1074  	if cc.sc.retryThrottling != nil {
  1075  		newThrottler := &retryThrottler{
  1076  			tokens: cc.sc.retryThrottling.MaxTokens,
  1077  			max:    cc.sc.retryThrottling.MaxTokens,
  1078  			thresh: cc.sc.retryThrottling.MaxTokens / 2,
  1079  			ratio:  cc.sc.retryThrottling.TokenRatio,
  1080  		}
  1081  		cc.retryThrottler.Store(newThrottler)
  1082  	} else {
  1083  		cc.retryThrottler.Store((*retryThrottler)(nil))
  1084  	}
  1085  }
  1086  
  1087  func (cc *ClientConn) resolveNow(o resolver.ResolveNowOptions) {
  1088  	cc.mu.RLock()
  1089  	cc.resolverWrapper.resolveNow(o)
  1090  	cc.mu.RUnlock()
  1091  }
  1092  
  1093  func (cc *ClientConn) resolveNowLocked(o resolver.ResolveNowOptions) {
  1094  	cc.resolverWrapper.resolveNow(o)
  1095  }
  1096  
  1097  // ResetConnectBackoff wakes up all subchannels in transient failure and causes
  1098  // them to attempt another connection immediately.  It also resets the backoff
  1099  // times used for subsequent attempts regardless of the current state.
  1100  //
  1101  // In general, this function should not be used.  Typical service or network
  1102  // outages result in a reasonable client reconnection strategy by default.
  1103  // However, if a previously unavailable network becomes available, this may be
  1104  // used to trigger an immediate reconnect.
  1105  //
  1106  // # Experimental
  1107  //
  1108  // Notice: This API is EXPERIMENTAL and may be changed or removed in a
  1109  // later release.
  1110  func (cc *ClientConn) ResetConnectBackoff() {
  1111  	cc.mu.Lock()
  1112  	conns := cc.conns
  1113  	cc.mu.Unlock()
  1114  	for ac := range conns {
  1115  		ac.resetConnectBackoff()
  1116  	}
  1117  }
  1118  
  1119  // Close tears down the ClientConn and all underlying connections.
  1120  func (cc *ClientConn) Close() error {
  1121  	defer func() {
  1122  		cc.cancel()
  1123  		<-cc.csMgr.pubSub.Done()
  1124  	}()
  1125  
  1126  	// Prevent calls to enter/exit idle immediately, and ensure we are not
  1127  	// currently entering/exiting idle mode.
  1128  	cc.idlenessMgr.Close()
  1129  
  1130  	cc.mu.Lock()
  1131  	if cc.conns == nil {
  1132  		cc.mu.Unlock()
  1133  		return ErrClientConnClosing
  1134  	}
  1135  
  1136  	conns := cc.conns
  1137  	cc.conns = nil
  1138  	cc.csMgr.updateState(connectivity.Shutdown)
  1139  
  1140  	// We can safely unlock and continue to access all fields now as
  1141  	// cc.conns==nil, preventing any further operations on cc.
  1142  	cc.mu.Unlock()
  1143  
  1144  	cc.resolverWrapper.close()
  1145  	// The order of closing matters here since the balancer wrapper assumes the
  1146  	// picker is closed before it is closed.
  1147  	cc.pickerWrapper.close()
  1148  	cc.balancerWrapper.close()
  1149  
  1150  	<-cc.resolverWrapper.serializer.Done()
  1151  	<-cc.balancerWrapper.serializer.Done()
  1152  
  1153  	for ac := range conns {
  1154  		ac.tearDown(ErrClientConnClosing)
  1155  	}
  1156  	cc.addTraceEvent("deleted")
  1157  	// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add
  1158  	// trace reference to the entity being deleted, and thus prevent it from being
  1159  	// deleted right away.
  1160  	channelz.RemoveEntry(cc.channelz.ID)
  1161  
  1162  	return nil
  1163  }
  1164  
  1165  // addrConn is a network connection to a given address.
  1166  type addrConn struct {
  1167  	ctx    context.Context
  1168  	cancel context.CancelFunc
  1169  
  1170  	cc     *ClientConn
  1171  	dopts  dialOptions
  1172  	acbw   *acBalancerWrapper
  1173  	scopts balancer.NewSubConnOptions
  1174  
  1175  	// transport is set when there's a viable transport (note: ac state may not be READY as LB channel
  1176  	// health checking may require server to report healthy to set ac to READY), and is reset
  1177  	// to nil when the current transport should no longer be used to create a stream (e.g. after GoAway
  1178  	// is received, transport is closed, ac has been torn down).
  1179  	transport transport.ClientTransport // The current transport.
  1180  
  1181  	// This mutex is used on the RPC path, so its usage should be minimized as
  1182  	// much as possible.
  1183  	// TODO: Find a lock-free way to retrieve the transport and state from the
  1184  	// addrConn.
  1185  	mu      sync.Mutex
  1186  	curAddr resolver.Address   // The current address.
  1187  	addrs   []resolver.Address // All addresses that the resolver resolved to.
  1188  
  1189  	// Use updateConnectivityState for updating addrConn's connectivity state.
  1190  	state     connectivity.State
  1191  	stateChan chan struct{} // closed and recreated on every state change.
  1192  
  1193  	backoffIdx   int // Needs to be stateful for resetConnectBackoff.
  1194  	resetBackoff chan struct{}
  1195  
  1196  	channelz *channelz.SubChannel
  1197  }
  1198  
  1199  // Note: this requires a lock on ac.mu.
  1200  func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error) {
  1201  	if ac.state == s {
  1202  		return
  1203  	}
  1204  	// When changing states, reset the state change channel.
  1205  	close(ac.stateChan)
  1206  	ac.stateChan = make(chan struct{})
  1207  	ac.state = s
  1208  	ac.channelz.ChannelMetrics.State.Store(&s)
  1209  	if lastErr == nil {
  1210  		channelz.Infof(logger, ac.channelz, "Subchannel Connectivity change to %v", s)
  1211  	} else {
  1212  		channelz.Infof(logger, ac.channelz, "Subchannel Connectivity change to %v, last error: %s", s, lastErr)
  1213  	}
  1214  	ac.acbw.updateState(s, lastErr)
  1215  }
  1216  
  1217  // adjustParams updates parameters used to create transports upon
  1218  // receiving a GoAway.
  1219  func (ac *addrConn) adjustParams(r transport.GoAwayReason) {
  1220  	switch r {
  1221  	case transport.GoAwayTooManyPings:
  1222  		v := 2 * ac.dopts.copts.KeepaliveParams.Time
  1223  		ac.cc.mu.Lock()
  1224  		if v > ac.cc.mkp.Time {
  1225  			ac.cc.mkp.Time = v
  1226  		}
  1227  		ac.cc.mu.Unlock()
  1228  	}
  1229  }
  1230  
  1231  func (ac *addrConn) resetTransport() {
  1232  	ac.mu.Lock()
  1233  	acCtx := ac.ctx
  1234  	if acCtx.Err() != nil {
  1235  		ac.mu.Unlock()
  1236  		return
  1237  	}
  1238  
  1239  	addrs := ac.addrs
  1240  	backoffFor := ac.dopts.bs.Backoff(ac.backoffIdx)
  1241  	// This will be the duration that dial gets to finish.
  1242  	dialDuration := minConnectTimeout
  1243  	if ac.dopts.minConnectTimeout != nil {
  1244  		dialDuration = ac.dopts.minConnectTimeout()
  1245  	}
  1246  
  1247  	if dialDuration < backoffFor {
  1248  		// Give dial more time as we keep failing to connect.
  1249  		dialDuration = backoffFor
  1250  	}
  1251  	// We can potentially spend all the time trying the first address, and
  1252  	// if the server accepts the connection and then hangs, the following
  1253  	// addresses will never be tried.
  1254  	//
  1255  	// The spec doesn't mention what should be done for multiple addresses.
  1256  	// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md#proposed-backoff-algorithm
  1257  	connectDeadline := time.Now().Add(dialDuration)
  1258  
  1259  	ac.updateConnectivityState(connectivity.Connecting, nil)
  1260  	ac.mu.Unlock()
  1261  
  1262  	if err := ac.tryAllAddrs(acCtx, addrs, connectDeadline); err != nil {
  1263  		ac.cc.resolveNow(resolver.ResolveNowOptions{})
  1264  		ac.mu.Lock()
  1265  		if acCtx.Err() != nil {
  1266  			// addrConn was torn down.
  1267  			ac.mu.Unlock()
  1268  			return
  1269  		}
  1270  		// After exhausting all addresses, the addrConn enters
  1271  		// TRANSIENT_FAILURE.
  1272  		ac.updateConnectivityState(connectivity.TransientFailure, err)
  1273  
  1274  		// Backoff.
  1275  		b := ac.resetBackoff
  1276  		ac.mu.Unlock()
  1277  
  1278  		timer := time.NewTimer(backoffFor)
  1279  		select {
  1280  		case <-timer.C:
  1281  			ac.mu.Lock()
  1282  			ac.backoffIdx++
  1283  			ac.mu.Unlock()
  1284  		case <-b:
  1285  			timer.Stop()
  1286  		case <-acCtx.Done():
  1287  			timer.Stop()
  1288  			return
  1289  		}
  1290  
  1291  		ac.mu.Lock()
  1292  		if acCtx.Err() == nil {
  1293  			ac.updateConnectivityState(connectivity.Idle, err)
  1294  		}
  1295  		ac.mu.Unlock()
  1296  		return
  1297  	}
  1298  	// Success; reset backoff.
  1299  	ac.mu.Lock()
  1300  	ac.backoffIdx = 0
  1301  	ac.mu.Unlock()
  1302  }
  1303  
  1304  // tryAllAddrs tries to creates a connection to the addresses, and stop when at
  1305  // the first successful one. It returns an error if no address was successfully
  1306  // connected, or updates ac appropriately with the new transport.
  1307  func (ac *addrConn) tryAllAddrs(ctx context.Context, addrs []resolver.Address, connectDeadline time.Time) error {
  1308  	var firstConnErr error
  1309  	for _, addr := range addrs {
  1310  		ac.channelz.ChannelMetrics.Target.Store(&addr.Addr)
  1311  		if ctx.Err() != nil {
  1312  			return errConnClosing
  1313  		}
  1314  		ac.mu.Lock()
  1315  
  1316  		ac.cc.mu.RLock()
  1317  		ac.dopts.copts.KeepaliveParams = ac.cc.mkp
  1318  		ac.cc.mu.RUnlock()
  1319  
  1320  		copts := ac.dopts.copts
  1321  		if ac.scopts.CredsBundle != nil {
  1322  			copts.CredsBundle = ac.scopts.CredsBundle
  1323  		}
  1324  		ac.mu.Unlock()
  1325  
  1326  		channelz.Infof(logger, ac.channelz, "Subchannel picks a new address %q to connect", addr.Addr)
  1327  
  1328  		err := ac.createTransport(ctx, addr, copts, connectDeadline)
  1329  		if err == nil {
  1330  			return nil
  1331  		}
  1332  		if firstConnErr == nil {
  1333  			firstConnErr = err
  1334  		}
  1335  		ac.cc.updateConnectionError(err)
  1336  	}
  1337  
  1338  	// Couldn't connect to any address.
  1339  	return firstConnErr
  1340  }
  1341  
  1342  // createTransport creates a connection to addr. It returns an error if the
  1343  // address was not successfully connected, or updates ac appropriately with the
  1344  // new transport.
  1345  func (ac *addrConn) createTransport(ctx context.Context, addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) error {
  1346  	addr.ServerName = ac.cc.getServerName(addr)
  1347  	hctx, hcancel := context.WithCancel(ctx)
  1348  
  1349  	onClose := func(r transport.GoAwayReason) {
  1350  		ac.mu.Lock()
  1351  		defer ac.mu.Unlock()
  1352  		// adjust params based on GoAwayReason
  1353  		ac.adjustParams(r)
  1354  		if ctx.Err() != nil {
  1355  			// Already shut down or connection attempt canceled.  tearDown() or
  1356  			// updateAddrs() already cleared the transport and canceled hctx
  1357  			// via ac.ctx, and we expected this connection to be closed, so do
  1358  			// nothing here.
  1359  			return
  1360  		}
  1361  		hcancel()
  1362  		if ac.transport == nil {
  1363  			// We're still connecting to this address, which could error.  Do
  1364  			// not update the connectivity state or resolve; these will happen
  1365  			// at the end of the tryAllAddrs connection loop in the event of an
  1366  			// error.
  1367  			return
  1368  		}
  1369  		ac.transport = nil
  1370  		// Refresh the name resolver on any connection loss.
  1371  		ac.cc.resolveNow(resolver.ResolveNowOptions{})
  1372  		// Always go idle and wait for the LB policy to initiate a new
  1373  		// connection attempt.
  1374  		ac.updateConnectivityState(connectivity.Idle, nil)
  1375  	}
  1376  
  1377  	connectCtx, cancel := context.WithDeadline(ctx, connectDeadline)
  1378  	defer cancel()
  1379  	copts.ChannelzParent = ac.channelz
  1380  
  1381  	newTr, err := transport.NewClientTransport(connectCtx, ac.cc.ctx, addr, copts, onClose)
  1382  	if err != nil {
  1383  		if logger.V(2) {
  1384  			logger.Infof("Creating new client transport to %q: %v", addr, err)
  1385  		}
  1386  		// newTr is either nil, or closed.
  1387  		hcancel()
  1388  		channelz.Warningf(logger, ac.channelz, "grpc: addrConn.createTransport failed to connect to %s. Err: %v", addr, err)
  1389  		return err
  1390  	}
  1391  
  1392  	ac.mu.Lock()
  1393  	defer ac.mu.Unlock()
  1394  	if ctx.Err() != nil {
  1395  		// This can happen if the subConn was removed while in `Connecting`
  1396  		// state. tearDown() would have set the state to `Shutdown`, but
  1397  		// would not have closed the transport since ac.transport would not
  1398  		// have been set at that point.
  1399  		//
  1400  		// We run this in a goroutine because newTr.Close() calls onClose()
  1401  		// inline, which requires locking ac.mu.
  1402  		//
  1403  		// The error we pass to Close() is immaterial since there are no open
  1404  		// streams at this point, so no trailers with error details will be sent
  1405  		// out. We just need to pass a non-nil error.
  1406  		//
  1407  		// This can also happen when updateAddrs is called during a connection
  1408  		// attempt.
  1409  		go newTr.Close(transport.ErrConnClosing)
  1410  		return nil
  1411  	}
  1412  	if hctx.Err() != nil {
  1413  		// onClose was already called for this connection, but the connection
  1414  		// was successfully established first.  Consider it a success and set
  1415  		// the new state to Idle.
  1416  		ac.updateConnectivityState(connectivity.Idle, nil)
  1417  		return nil
  1418  	}
  1419  	ac.curAddr = addr
  1420  	ac.transport = newTr
  1421  	ac.startHealthCheck(hctx) // Will set state to READY if appropriate.
  1422  	return nil
  1423  }
  1424  
  1425  // startHealthCheck starts the health checking stream (RPC) to watch the health
  1426  // stats of this connection if health checking is requested and configured.
  1427  //
  1428  // LB channel health checking is enabled when all requirements below are met:
  1429  // 1. it is not disabled by the user with the WithDisableHealthCheck DialOption
  1430  // 2. internal.HealthCheckFunc is set by importing the grpc/health package
  1431  // 3. a service config with non-empty healthCheckConfig field is provided
  1432  // 4. the load balancer requests it
  1433  //
  1434  // It sets addrConn to READY if the health checking stream is not started.
  1435  //
  1436  // Caller must hold ac.mu.
  1437  func (ac *addrConn) startHealthCheck(ctx context.Context) {
  1438  	var healthcheckManagingState bool
  1439  	defer func() {
  1440  		if !healthcheckManagingState {
  1441  			ac.updateConnectivityState(connectivity.Ready, nil)
  1442  		}
  1443  	}()
  1444  
  1445  	if ac.cc.dopts.disableHealthCheck {
  1446  		return
  1447  	}
  1448  	healthCheckConfig := ac.cc.healthCheckConfig()
  1449  	if healthCheckConfig == nil {
  1450  		return
  1451  	}
  1452  	if !ac.scopts.HealthCheckEnabled {
  1453  		return
  1454  	}
  1455  	healthCheckFunc := ac.cc.dopts.healthCheckFunc
  1456  	if healthCheckFunc == nil {
  1457  		// The health package is not imported to set health check function.
  1458  		//
  1459  		// TODO: add a link to the health check doc in the error message.
  1460  		channelz.Error(logger, ac.channelz, "Health check is requested but health check function is not set.")
  1461  		return
  1462  	}
  1463  
  1464  	healthcheckManagingState = true
  1465  
  1466  	// Set up the health check helper functions.
  1467  	currentTr := ac.transport
  1468  	newStream := func(method string) (any, error) {
  1469  		ac.mu.Lock()
  1470  		if ac.transport != currentTr {
  1471  			ac.mu.Unlock()
  1472  			return nil, status.Error(codes.Canceled, "the provided transport is no longer valid to use")
  1473  		}
  1474  		ac.mu.Unlock()
  1475  		return newNonRetryClientStream(ctx, &StreamDesc{ServerStreams: true}, method, currentTr, ac)
  1476  	}
  1477  	setConnectivityState := func(s connectivity.State, lastErr error) {
  1478  		ac.mu.Lock()
  1479  		defer ac.mu.Unlock()
  1480  		if ac.transport != currentTr {
  1481  			return
  1482  		}
  1483  		ac.updateConnectivityState(s, lastErr)
  1484  	}
  1485  	// Start the health checking stream.
  1486  	go func() {
  1487  		err := ac.cc.dopts.healthCheckFunc(ctx, newStream, setConnectivityState, healthCheckConfig.ServiceName)
  1488  		if err != nil {
  1489  			if status.Code(err) == codes.Unimplemented {
  1490  				channelz.Error(logger, ac.channelz, "Subchannel health check is unimplemented at server side, thus health check is disabled")
  1491  			} else {
  1492  				channelz.Errorf(logger, ac.channelz, "Health checking failed: %v", err)
  1493  			}
  1494  		}
  1495  	}()
  1496  }
  1497  
  1498  func (ac *addrConn) resetConnectBackoff() {
  1499  	ac.mu.Lock()
  1500  	close(ac.resetBackoff)
  1501  	ac.backoffIdx = 0
  1502  	ac.resetBackoff = make(chan struct{})
  1503  	ac.mu.Unlock()
  1504  }
  1505  
  1506  // getReadyTransport returns the transport if ac's state is READY or nil if not.
  1507  func (ac *addrConn) getReadyTransport() transport.ClientTransport {
  1508  	ac.mu.Lock()
  1509  	defer ac.mu.Unlock()
  1510  	if ac.state == connectivity.Ready {
  1511  		return ac.transport
  1512  	}
  1513  	return nil
  1514  }
  1515  
  1516  // getTransport waits until the addrconn is ready and returns the transport.
  1517  // If the context expires first, returns an appropriate status.  If the
  1518  // addrConn is stopped first, returns an Unavailable status error.
  1519  func (ac *addrConn) getTransport(ctx context.Context) (transport.ClientTransport, error) {
  1520  	for ctx.Err() == nil {
  1521  		ac.mu.Lock()
  1522  		t, state, sc := ac.transport, ac.state, ac.stateChan
  1523  		ac.mu.Unlock()
  1524  		if state == connectivity.Ready {
  1525  			return t, nil
  1526  		}
  1527  		if state == connectivity.Shutdown {
  1528  			return nil, status.Errorf(codes.Unavailable, "SubConn shutting down")
  1529  		}
  1530  
  1531  		select {
  1532  		case <-ctx.Done():
  1533  		case <-sc:
  1534  		}
  1535  	}
  1536  	return nil, status.FromContextError(ctx.Err()).Err()
  1537  }
  1538  
  1539  // tearDown starts to tear down the addrConn.
  1540  //
  1541  // Note that tearDown doesn't remove ac from ac.cc.conns, so the addrConn struct
  1542  // will leak. In most cases, call cc.removeAddrConn() instead.
  1543  func (ac *addrConn) tearDown(err error) {
  1544  	ac.mu.Lock()
  1545  	if ac.state == connectivity.Shutdown {
  1546  		ac.mu.Unlock()
  1547  		return
  1548  	}
  1549  	curTr := ac.transport
  1550  	ac.transport = nil
  1551  	// We have to set the state to Shutdown before anything else to prevent races
  1552  	// between setting the state and logic that waits on context cancellation / etc.
  1553  	ac.updateConnectivityState(connectivity.Shutdown, nil)
  1554  	ac.cancel()
  1555  	ac.curAddr = resolver.Address{}
  1556  
  1557  	channelz.AddTraceEvent(logger, ac.channelz, 0, &channelz.TraceEvent{
  1558  		Desc:     "Subchannel deleted",
  1559  		Severity: channelz.CtInfo,
  1560  		Parent: &channelz.TraceEvent{
  1561  			Desc:     fmt.Sprintf("Subchannel(id:%d) deleted", ac.channelz.ID),
  1562  			Severity: channelz.CtInfo,
  1563  		},
  1564  	})
  1565  	// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add
  1566  	// trace reference to the entity being deleted, and thus prevent it from
  1567  	// being deleted right away.
  1568  	channelz.RemoveEntry(ac.channelz.ID)
  1569  	ac.mu.Unlock()
  1570  
  1571  	// We have to release the lock before the call to GracefulClose/Close here
  1572  	// because both of them call onClose(), which requires locking ac.mu.
  1573  	if curTr != nil {
  1574  		if err == errConnDrain {
  1575  			// Close the transport gracefully when the subConn is being shutdown.
  1576  			//
  1577  			// GracefulClose() may be executed multiple times if:
  1578  			// - multiple GoAway frames are received from the server
  1579  			// - there are concurrent name resolver or balancer triggered
  1580  			//   address removal and GoAway
  1581  			curTr.GracefulClose()
  1582  		} else {
  1583  			// Hard close the transport when the channel is entering idle or is
  1584  			// being shutdown. In the case where the channel is being shutdown,
  1585  			// closing of transports is also taken care of by cancelation of cc.ctx.
  1586  			// But in the case where the channel is entering idle, we need to
  1587  			// explicitly close the transports here. Instead of distinguishing
  1588  			// between these two cases, it is simpler to close the transport
  1589  			// unconditionally here.
  1590  			curTr.Close(err)
  1591  		}
  1592  	}
  1593  }
  1594  
  1595  type retryThrottler struct {
  1596  	max    float64
  1597  	thresh float64
  1598  	ratio  float64
  1599  
  1600  	mu     sync.Mutex
  1601  	tokens float64 // TODO(dfawley): replace with atomic and remove lock.
  1602  }
  1603  
  1604  // throttle subtracts a retry token from the pool and returns whether a retry
  1605  // should be throttled (disallowed) based upon the retry throttling policy in
  1606  // the service config.
  1607  func (rt *retryThrottler) throttle() bool {
  1608  	if rt == nil {
  1609  		return false
  1610  	}
  1611  	rt.mu.Lock()
  1612  	defer rt.mu.Unlock()
  1613  	rt.tokens--
  1614  	if rt.tokens < 0 {
  1615  		rt.tokens = 0
  1616  	}
  1617  	return rt.tokens <= rt.thresh
  1618  }
  1619  
  1620  func (rt *retryThrottler) successfulRPC() {
  1621  	if rt == nil {
  1622  		return
  1623  	}
  1624  	rt.mu.Lock()
  1625  	defer rt.mu.Unlock()
  1626  	rt.tokens += rt.ratio
  1627  	if rt.tokens > rt.max {
  1628  		rt.tokens = rt.max
  1629  	}
  1630  }
  1631  
  1632  func (ac *addrConn) incrCallsStarted() {
  1633  	ac.channelz.ChannelMetrics.CallsStarted.Add(1)
  1634  	ac.channelz.ChannelMetrics.LastCallStartedTimestamp.Store(time.Now().UnixNano())
  1635  }
  1636  
  1637  func (ac *addrConn) incrCallsSucceeded() {
  1638  	ac.channelz.ChannelMetrics.CallsSucceeded.Add(1)
  1639  }
  1640  
  1641  func (ac *addrConn) incrCallsFailed() {
  1642  	ac.channelz.ChannelMetrics.CallsFailed.Add(1)
  1643  }
  1644  
  1645  // ErrClientConnTimeout indicates that the ClientConn cannot establish the
  1646  // underlying connections within the specified timeout.
  1647  //
  1648  // Deprecated: This error is never returned by grpc and should not be
  1649  // referenced by users.
  1650  var ErrClientConnTimeout = errors.New("grpc: timed out when dialing")
  1651  
  1652  // getResolver finds the scheme in the cc's resolvers or the global registry.
  1653  // scheme should always be lowercase (typically by virtue of url.Parse()
  1654  // performing proper RFC3986 behavior).
  1655  func (cc *ClientConn) getResolver(scheme string) resolver.Builder {
  1656  	for _, rb := range cc.dopts.resolvers {
  1657  		if scheme == rb.Scheme() {
  1658  			return rb
  1659  		}
  1660  	}
  1661  	return resolver.Get(scheme)
  1662  }
  1663  
  1664  func (cc *ClientConn) updateConnectionError(err error) {
  1665  	cc.lceMu.Lock()
  1666  	cc.lastConnectionError = err
  1667  	cc.lceMu.Unlock()
  1668  }
  1669  
  1670  func (cc *ClientConn) connectionError() error {
  1671  	cc.lceMu.Lock()
  1672  	defer cc.lceMu.Unlock()
  1673  	return cc.lastConnectionError
  1674  }
  1675  
  1676  // parseTargetAndFindResolver parses the user's dial target and stores the
  1677  // parsed target in `cc.parsedTarget`.
  1678  //
  1679  // The resolver to use is determined based on the scheme in the parsed target
  1680  // and the same is stored in `cc.resolverBuilder`.
  1681  //
  1682  // Doesn't grab cc.mu as this method is expected to be called only at Dial time.
  1683  func (cc *ClientConn) parseTargetAndFindResolver() error {
  1684  	channelz.Infof(logger, cc.channelz, "original dial target is: %q", cc.target)
  1685  
  1686  	var rb resolver.Builder
  1687  	parsedTarget, err := parseTarget(cc.target)
  1688  	if err != nil {
  1689  		channelz.Infof(logger, cc.channelz, "dial target %q parse failed: %v", cc.target, err)
  1690  	} else {
  1691  		channelz.Infof(logger, cc.channelz, "parsed dial target is: %#v", parsedTarget)
  1692  		rb = cc.getResolver(parsedTarget.URL.Scheme)
  1693  		if rb != nil {
  1694  			cc.parsedTarget = parsedTarget
  1695  			cc.resolverBuilder = rb
  1696  			return nil
  1697  		}
  1698  	}
  1699  
  1700  	// We are here because the user's dial target did not contain a scheme or
  1701  	// specified an unregistered scheme. We should fallback to the default
  1702  	// scheme, except when a custom dialer is specified in which case, we should
  1703  	// always use passthrough scheme. For either case, we need to respect any overridden
  1704  	// global defaults set by the user.
  1705  	defScheme := cc.dopts.defaultScheme
  1706  	if internal.UserSetDefaultScheme {
  1707  		defScheme = resolver.GetDefaultScheme()
  1708  	}
  1709  
  1710  	channelz.Infof(logger, cc.channelz, "fallback to scheme %q", defScheme)
  1711  	canonicalTarget := defScheme + ":///" + cc.target
  1712  
  1713  	parsedTarget, err = parseTarget(canonicalTarget)
  1714  	if err != nil {
  1715  		channelz.Infof(logger, cc.channelz, "dial target %q parse failed: %v", canonicalTarget, err)
  1716  		return err
  1717  	}
  1718  	channelz.Infof(logger, cc.channelz, "parsed dial target is: %+v", parsedTarget)
  1719  	rb = cc.getResolver(parsedTarget.URL.Scheme)
  1720  	if rb == nil {
  1721  		return fmt.Errorf("could not get resolver for default scheme: %q", parsedTarget.URL.Scheme)
  1722  	}
  1723  	cc.parsedTarget = parsedTarget
  1724  	cc.resolverBuilder = rb
  1725  	return nil
  1726  }
  1727  
  1728  // parseTarget uses RFC 3986 semantics to parse the given target into a
  1729  // resolver.Target struct containing url. Query params are stripped from the
  1730  // endpoint.
  1731  func parseTarget(target string) (resolver.Target, error) {
  1732  	u, err := url.Parse(target)
  1733  	if err != nil {
  1734  		return resolver.Target{}, err
  1735  	}
  1736  
  1737  	return resolver.Target{URL: *u}, nil
  1738  }
  1739  
  1740  // encodeAuthority escapes the authority string based on valid chars defined in
  1741  // https://datatracker.ietf.org/doc/html/rfc3986#section-3.2.
  1742  func encodeAuthority(authority string) string {
  1743  	const upperhex = "0123456789ABCDEF"
  1744  
  1745  	// Return for characters that must be escaped as per
  1746  	// Valid chars are mentioned here:
  1747  	// https://datatracker.ietf.org/doc/html/rfc3986#section-3.2
  1748  	shouldEscape := func(c byte) bool {
  1749  		// Alphanum are always allowed.
  1750  		if 'a' <= c && c <= 'z' || 'A' <= c && c <= 'Z' || '0' <= c && c <= '9' {
  1751  			return false
  1752  		}
  1753  		switch c {
  1754  		case '-', '_', '.', '~': // Unreserved characters
  1755  			return false
  1756  		case '!', '$', '&', '\'', '(', ')', '*', '+', ',', ';', '=': // Subdelim characters
  1757  			return false
  1758  		case ':', '[', ']', '@': // Authority related delimiters
  1759  			return false
  1760  		}
  1761  		// Everything else must be escaped.
  1762  		return true
  1763  	}
  1764  
  1765  	hexCount := 0
  1766  	for i := 0; i < len(authority); i++ {
  1767  		c := authority[i]
  1768  		if shouldEscape(c) {
  1769  			hexCount++
  1770  		}
  1771  	}
  1772  
  1773  	if hexCount == 0 {
  1774  		return authority
  1775  	}
  1776  
  1777  	required := len(authority) + 2*hexCount
  1778  	t := make([]byte, required)
  1779  
  1780  	j := 0
  1781  	// This logic is a barebones version of escape in the go net/url library.
  1782  	for i := 0; i < len(authority); i++ {
  1783  		switch c := authority[i]; {
  1784  		case shouldEscape(c):
  1785  			t[j] = '%'
  1786  			t[j+1] = upperhex[c>>4]
  1787  			t[j+2] = upperhex[c&15]
  1788  			j += 3
  1789  		default:
  1790  			t[j] = authority[i]
  1791  			j++
  1792  		}
  1793  	}
  1794  	return string(t)
  1795  }
  1796  
  1797  // Determine channel authority. The order of precedence is as follows:
  1798  // - user specified authority override using `WithAuthority` dial option
  1799  // - creds' notion of server name for the authentication handshake
  1800  // - endpoint from dial target of the form "scheme://[authority]/endpoint"
  1801  //
  1802  // Stores the determined authority in `cc.authority`.
  1803  //
  1804  // Returns a non-nil error if the authority returned by the transport
  1805  // credentials do not match the authority configured through the dial option.
  1806  //
  1807  // Doesn't grab cc.mu as this method is expected to be called only at Dial time.
  1808  func (cc *ClientConn) determineAuthority() error {
  1809  	dopts := cc.dopts
  1810  	// Historically, we had two options for users to specify the serverName or
  1811  	// authority for a channel. One was through the transport credentials
  1812  	// (either in its constructor, or through the OverrideServerName() method).
  1813  	// The other option (for cases where WithInsecure() dial option was used)
  1814  	// was to use the WithAuthority() dial option.
  1815  	//
  1816  	// A few things have changed since:
  1817  	// - `insecure` package with an implementation of the `TransportCredentials`
  1818  	//   interface for the insecure case
  1819  	// - WithAuthority() dial option support for secure credentials
  1820  	authorityFromCreds := ""
  1821  	if creds := dopts.copts.TransportCredentials; creds != nil && creds.Info().ServerName != "" {
  1822  		authorityFromCreds = creds.Info().ServerName
  1823  	}
  1824  	authorityFromDialOption := dopts.authority
  1825  	if (authorityFromCreds != "" && authorityFromDialOption != "") && authorityFromCreds != authorityFromDialOption {
  1826  		return fmt.Errorf("ClientConn's authority from transport creds %q and dial option %q don't match", authorityFromCreds, authorityFromDialOption)
  1827  	}
  1828  
  1829  	endpoint := cc.parsedTarget.Endpoint()
  1830  	if authorityFromDialOption != "" {
  1831  		cc.authority = authorityFromDialOption
  1832  	} else if authorityFromCreds != "" {
  1833  		cc.authority = authorityFromCreds
  1834  	} else if auth, ok := cc.resolverBuilder.(resolver.AuthorityOverrider); ok {
  1835  		cc.authority = auth.OverrideAuthority(cc.parsedTarget)
  1836  	} else if strings.HasPrefix(endpoint, ":") {
  1837  		cc.authority = "localhost" + endpoint
  1838  	} else {
  1839  		cc.authority = encodeAuthority(endpoint)
  1840  	}
  1841  	channelz.Infof(logger, cc.channelz, "Channel authority set to %q", cc.authority)
  1842  	return nil
  1843  }
  1844  

View as plain text