2019-04-13 12:47:57 +00:00
/ *
*
* Copyright 2014 gRPC authors .
*
* Licensed under the Apache License , Version 2.0 ( the "License" ) ;
* you may not use this file except in compliance with the License .
* You may obtain a copy of the License at
*
* http : //www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing , software
* distributed under the License is distributed on an "AS IS" BASIS ,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND , either express or implied .
* See the License for the specific language governing permissions and
* limitations under the License .
*
* /
package grpc
import (
"context"
"errors"
"fmt"
"math"
"net"
"reflect"
"strings"
"sync"
"sync/atomic"
"time"
"google.golang.org/grpc/balancer"
_ "google.golang.org/grpc/balancer/roundrobin" // To register roundrobin.
"google.golang.org/grpc/codes"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/grpclog"
"google.golang.org/grpc/internal/backoff"
"google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/internal/envconfig"
"google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/resolver"
_ "google.golang.org/grpc/resolver/dns" // To register dns resolver.
_ "google.golang.org/grpc/resolver/passthrough" // To register passthrough resolver.
"google.golang.org/grpc/status"
)
const (
// minimum time to give a connection to complete
minConnectTimeout = 20 * time . Second
// must match grpclbName in grpclb/grpclb.go
grpclbName = "grpclb"
)
var (
// ErrClientConnClosing indicates that the operation is illegal because
// the ClientConn is closing.
//
// Deprecated: this error should not be relied upon by users; use the status
// code of Canceled instead.
ErrClientConnClosing = status . Error ( codes . Canceled , "grpc: the client connection is closing" )
// errConnDrain indicates that the connection starts to be drained and does not accept any new RPCs.
errConnDrain = errors . New ( "grpc: the connection is drained" )
// errConnClosing indicates that the connection is closing.
errConnClosing = errors . New ( "grpc: the connection is closing" )
// errBalancerClosed indicates that the balancer is closed.
errBalancerClosed = errors . New ( "grpc: balancer is closed" )
// invalidDefaultServiceConfigErrPrefix is used to prefix the json parsing error for the default
// service config.
invalidDefaultServiceConfigErrPrefix = "grpc: the provided default service config is invalid"
)
// The following errors are returned from Dial and DialContext
var (
// errNoTransportSecurity indicates that there is no transport security
// being set for ClientConn. Users should either set one or explicitly
// call WithInsecure DialOption to disable security.
errNoTransportSecurity = errors . New ( "grpc: no transport security set (use grpc.WithInsecure() explicitly or set credentials)" )
// errTransportCredsAndBundle indicates that creds bundle is used together
// with other individual Transport Credentials.
errTransportCredsAndBundle = errors . New ( "grpc: credentials.Bundle may not be used with individual TransportCredentials" )
// errTransportCredentialsMissing indicates that users want to transmit security
// information (e.g., OAuth2 token) which requires secure connection on an insecure
// connection.
errTransportCredentialsMissing = errors . New ( "grpc: the credentials require transport level security (use grpc.WithTransportCredentials() to set)" )
// errCredentialsConflict indicates that grpc.WithTransportCredentials()
// and grpc.WithInsecure() are both called for a connection.
errCredentialsConflict = errors . New ( "grpc: transport credentials are set for an insecure connection (grpc.WithTransportCredentials() and grpc.WithInsecure() are both called)" )
)
const (
defaultClientMaxReceiveMessageSize = 1024 * 1024 * 4
defaultClientMaxSendMessageSize = math . MaxInt32
// http2IOBufSize specifies the buffer size for sending frames.
defaultWriteBufSize = 32 * 1024
defaultReadBufSize = 32 * 1024
)
// Dial creates a client connection to the given target.
func Dial ( target string , opts ... DialOption ) ( * ClientConn , error ) {
return DialContext ( context . Background ( ) , target , opts ... )
}
// DialContext creates a client connection to the given target. By default, it's
// a non-blocking dial (the function won't wait for connections to be
// established, and connecting happens in the background). To make it a blocking
// dial, use WithBlock() dial option.
//
// In the non-blocking case, the ctx does not act against the connection. It
// only controls the setup steps.
//
// In the blocking case, ctx can be used to cancel or expire the pending
// connection. Once this function returns, the cancellation and expiration of
// ctx will be noop. Users should call ClientConn.Close to terminate all the
// pending operations after this function returns.
//
// The target name syntax is defined in
// https://github.com/grpc/grpc/blob/master/doc/naming.md.
// e.g. to use dns resolver, a "dns:///" prefix should be applied to the target.
func DialContext ( ctx context . Context , target string , opts ... DialOption ) ( conn * ClientConn , err error ) {
cc := & ClientConn {
target : target ,
csMgr : & connectivityStateManager { } ,
conns : make ( map [ * addrConn ] struct { } ) ,
dopts : defaultDialOptions ( ) ,
blockingpicker : newPickerWrapper ( ) ,
czData : new ( channelzData ) ,
firstResolveEvent : grpcsync . NewEvent ( ) ,
}
cc . retryThrottler . Store ( ( * retryThrottler ) ( nil ) )
cc . ctx , cc . cancel = context . WithCancel ( context . Background ( ) )
for _ , opt := range opts {
opt . apply ( & cc . dopts )
}
2019-06-27 11:30:45 +00:00
chainUnaryClientInterceptors ( cc )
chainStreamClientInterceptors ( cc )
2019-04-13 12:47:57 +00:00
defer func ( ) {
if err != nil {
cc . Close ( )
}
} ( )
if channelz . IsOn ( ) {
if cc . dopts . channelzParentID != 0 {
cc . channelzID = channelz . RegisterChannel ( & channelzChannel { cc } , cc . dopts . channelzParentID , target )
channelz . AddTraceEvent ( cc . channelzID , & channelz . TraceEventDesc {
Desc : "Channel Created" ,
Severity : channelz . CtINFO ,
Parent : & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Nested Channel(id:%d) created" , cc . channelzID ) ,
Severity : channelz . CtINFO ,
} ,
} )
} else {
cc . channelzID = channelz . RegisterChannel ( & channelzChannel { cc } , 0 , target )
channelz . AddTraceEvent ( cc . channelzID , & channelz . TraceEventDesc {
Desc : "Channel Created" ,
Severity : channelz . CtINFO ,
} )
}
cc . csMgr . channelzID = cc . channelzID
}
if ! cc . dopts . insecure {
if cc . dopts . copts . TransportCredentials == nil && cc . dopts . copts . CredsBundle == nil {
return nil , errNoTransportSecurity
}
if cc . dopts . copts . TransportCredentials != nil && cc . dopts . copts . CredsBundle != nil {
return nil , errTransportCredsAndBundle
}
} else {
if cc . dopts . copts . TransportCredentials != nil || cc . dopts . copts . CredsBundle != nil {
return nil , errCredentialsConflict
}
for _ , cd := range cc . dopts . copts . PerRPCCredentials {
if cd . RequireTransportSecurity ( ) {
return nil , errTransportCredentialsMissing
}
}
}
if cc . dopts . defaultServiceConfigRawJSON != nil {
sc , err := parseServiceConfig ( * cc . dopts . defaultServiceConfigRawJSON )
if err != nil {
return nil , fmt . Errorf ( "%s: %v" , invalidDefaultServiceConfigErrPrefix , err )
}
cc . dopts . defaultServiceConfig = sc
}
cc . mkp = cc . dopts . copts . KeepaliveParams
if cc . dopts . copts . Dialer == nil {
cc . dopts . copts . Dialer = newProxyDialer (
func ( ctx context . Context , addr string ) ( net . Conn , error ) {
network , addr := parseDialTarget ( addr )
return ( & net . Dialer { } ) . DialContext ( ctx , network , addr )
} ,
)
}
if cc . dopts . copts . UserAgent != "" {
cc . dopts . copts . UserAgent += " " + grpcUA
} else {
cc . dopts . copts . UserAgent = grpcUA
}
if cc . dopts . timeout > 0 {
var cancel context . CancelFunc
ctx , cancel = context . WithTimeout ( ctx , cc . dopts . timeout )
defer cancel ( )
}
defer func ( ) {
select {
case <- ctx . Done ( ) :
conn , err = nil , ctx . Err ( )
default :
}
} ( )
scSet := false
if cc . dopts . scChan != nil {
// Try to get an initial service config.
select {
case sc , ok := <- cc . dopts . scChan :
if ok {
cc . sc = & sc
scSet = true
}
default :
}
}
if cc . dopts . bs == nil {
cc . dopts . bs = backoff . Exponential {
MaxDelay : DefaultBackoffConfig . MaxDelay ,
}
}
if cc . dopts . resolverBuilder == nil {
// Only try to parse target when resolver builder is not already set.
cc . parsedTarget = parseTarget ( cc . target )
grpclog . Infof ( "parsed scheme: %q" , cc . parsedTarget . Scheme )
cc . dopts . resolverBuilder = resolver . Get ( cc . parsedTarget . Scheme )
if cc . dopts . resolverBuilder == nil {
// If resolver builder is still nil, the parsed target's scheme is
// not registered. Fallback to default resolver and set Endpoint to
// the original target.
grpclog . Infof ( "scheme %q not registered, fallback to default scheme" , cc . parsedTarget . Scheme )
cc . parsedTarget = resolver . Target {
Scheme : resolver . GetDefaultScheme ( ) ,
Endpoint : target ,
}
cc . dopts . resolverBuilder = resolver . Get ( cc . parsedTarget . Scheme )
}
} else {
cc . parsedTarget = resolver . Target { Endpoint : target }
}
creds := cc . dopts . copts . TransportCredentials
if creds != nil && creds . Info ( ) . ServerName != "" {
cc . authority = creds . Info ( ) . ServerName
} else if cc . dopts . insecure && cc . dopts . authority != "" {
cc . authority = cc . dopts . authority
} else {
// Use endpoint from "scheme://authority/endpoint" as the default
// authority for ClientConn.
cc . authority = cc . parsedTarget . Endpoint
}
if cc . dopts . scChan != nil && ! scSet {
// Blocking wait for the initial service config.
select {
case sc , ok := <- cc . dopts . scChan :
if ok {
cc . sc = & sc
}
case <- ctx . Done ( ) :
return nil , ctx . Err ( )
}
}
if cc . dopts . scChan != nil {
go cc . scWatcher ( )
}
var credsClone credentials . TransportCredentials
if creds := cc . dopts . copts . TransportCredentials ; creds != nil {
credsClone = creds . Clone ( )
}
cc . balancerBuildOpts = balancer . BuildOptions {
DialCreds : credsClone ,
CredsBundle : cc . dopts . copts . CredsBundle ,
Dialer : cc . dopts . copts . Dialer ,
ChannelzParentID : cc . channelzID ,
2019-06-27 11:30:45 +00:00
Target : cc . parsedTarget ,
2019-04-13 12:47:57 +00:00
}
// Build the resolver.
rWrapper , err := newCCResolverWrapper ( cc )
if err != nil {
return nil , fmt . Errorf ( "failed to build resolver: %v" , err )
}
cc . mu . Lock ( )
cc . resolverWrapper = rWrapper
cc . mu . Unlock ( )
// A blocking dial blocks until the clientConn is ready.
if cc . dopts . block {
for {
s := cc . GetState ( )
if s == connectivity . Ready {
break
} else if cc . dopts . copts . FailOnNonTempDialError && s == connectivity . TransientFailure {
if err = cc . blockingpicker . connectionError ( ) ; err != nil {
terr , ok := err . ( interface {
Temporary ( ) bool
} )
if ok && ! terr . Temporary ( ) {
return nil , err
}
}
}
if ! cc . WaitForStateChange ( ctx , s ) {
// ctx got timeout or canceled.
return nil , ctx . Err ( )
}
}
}
return cc , nil
}
2019-06-27 11:30:45 +00:00
// chainUnaryClientInterceptors chains all unary client interceptors into one.
func chainUnaryClientInterceptors ( cc * ClientConn ) {
interceptors := cc . dopts . chainUnaryInts
// Prepend dopts.unaryInt to the chaining interceptors if it exists, since unaryInt will
// be executed before any other chained interceptors.
if cc . dopts . unaryInt != nil {
interceptors = append ( [ ] UnaryClientInterceptor { cc . dopts . unaryInt } , interceptors ... )
}
var chainedInt UnaryClientInterceptor
if len ( interceptors ) == 0 {
chainedInt = nil
} else if len ( interceptors ) == 1 {
chainedInt = interceptors [ 0 ]
} else {
chainedInt = func ( ctx context . Context , method string , req , reply interface { } , cc * ClientConn , invoker UnaryInvoker , opts ... CallOption ) error {
return interceptors [ 0 ] ( ctx , method , req , reply , cc , getChainUnaryInvoker ( interceptors , 0 , invoker ) , opts ... )
}
}
cc . dopts . unaryInt = chainedInt
}
// getChainUnaryInvoker recursively generate the chained unary invoker.
func getChainUnaryInvoker ( interceptors [ ] UnaryClientInterceptor , curr int , finalInvoker UnaryInvoker ) UnaryInvoker {
if curr == len ( interceptors ) - 1 {
return finalInvoker
}
return func ( ctx context . Context , method string , req , reply interface { } , cc * ClientConn , opts ... CallOption ) error {
return interceptors [ curr + 1 ] ( ctx , method , req , reply , cc , getChainUnaryInvoker ( interceptors , curr + 1 , finalInvoker ) , opts ... )
}
}
// chainStreamClientInterceptors chains all stream client interceptors into one.
func chainStreamClientInterceptors ( cc * ClientConn ) {
interceptors := cc . dopts . chainStreamInts
// Prepend dopts.streamInt to the chaining interceptors if it exists, since streamInt will
// be executed before any other chained interceptors.
if cc . dopts . streamInt != nil {
interceptors = append ( [ ] StreamClientInterceptor { cc . dopts . streamInt } , interceptors ... )
}
var chainedInt StreamClientInterceptor
if len ( interceptors ) == 0 {
chainedInt = nil
} else if len ( interceptors ) == 1 {
chainedInt = interceptors [ 0 ]
} else {
chainedInt = func ( ctx context . Context , desc * StreamDesc , cc * ClientConn , method string , streamer Streamer , opts ... CallOption ) ( ClientStream , error ) {
return interceptors [ 0 ] ( ctx , desc , cc , method , getChainStreamer ( interceptors , 0 , streamer ) , opts ... )
}
}
cc . dopts . streamInt = chainedInt
}
// getChainStreamer recursively generate the chained client stream constructor.
func getChainStreamer ( interceptors [ ] StreamClientInterceptor , curr int , finalStreamer Streamer ) Streamer {
if curr == len ( interceptors ) - 1 {
return finalStreamer
}
return func ( ctx context . Context , desc * StreamDesc , cc * ClientConn , method string , opts ... CallOption ) ( ClientStream , error ) {
return interceptors [ curr + 1 ] ( ctx , desc , cc , method , getChainStreamer ( interceptors , curr + 1 , finalStreamer ) , opts ... )
}
}
2019-04-13 12:47:57 +00:00
// connectivityStateManager keeps the connectivity.State of ClientConn.
// This struct will eventually be exported so the balancers can access it.
type connectivityStateManager struct {
mu sync . Mutex
state connectivity . State
notifyChan chan struct { }
channelzID int64
}
// updateState updates the connectivity.State of ClientConn.
// If there's a change it notifies goroutines waiting on state change to
// happen.
func ( csm * connectivityStateManager ) updateState ( state connectivity . State ) {
csm . mu . Lock ( )
defer csm . mu . Unlock ( )
if csm . state == connectivity . Shutdown {
return
}
if csm . state == state {
return
}
csm . state = state
if channelz . IsOn ( ) {
channelz . AddTraceEvent ( csm . channelzID , & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Channel Connectivity change to %v" , state ) ,
Severity : channelz . CtINFO ,
} )
}
if csm . notifyChan != nil {
// There are other goroutines waiting on this channel.
close ( csm . notifyChan )
csm . notifyChan = nil
}
}
func ( csm * connectivityStateManager ) getState ( ) connectivity . State {
csm . mu . Lock ( )
defer csm . mu . Unlock ( )
return csm . state
}
func ( csm * connectivityStateManager ) getNotifyChan ( ) <- chan struct { } {
csm . mu . Lock ( )
defer csm . mu . Unlock ( )
if csm . notifyChan == nil {
csm . notifyChan = make ( chan struct { } )
}
return csm . notifyChan
}
// ClientConn represents a client connection to an RPC server.
type ClientConn struct {
ctx context . Context
cancel context . CancelFunc
target string
parsedTarget resolver . Target
authority string
dopts dialOptions
csMgr * connectivityStateManager
balancerBuildOpts balancer . BuildOptions
blockingpicker * pickerWrapper
mu sync . RWMutex
resolverWrapper * ccResolverWrapper
sc * ServiceConfig
conns map [ * addrConn ] struct { }
// Keepalive parameter can be updated if a GoAway is received.
mkp keepalive . ClientParameters
curBalancerName string
balancerWrapper * ccBalancerWrapper
retryThrottler atomic . Value
firstResolveEvent * grpcsync . Event
channelzID int64 // channelz unique identification number
czData * channelzData
}
// WaitForStateChange waits until the connectivity.State of ClientConn changes from sourceState or
// ctx expires. A true value is returned in former case and false in latter.
// This is an EXPERIMENTAL API.
func ( cc * ClientConn ) WaitForStateChange ( ctx context . Context , sourceState connectivity . State ) bool {
ch := cc . csMgr . getNotifyChan ( )
if cc . csMgr . getState ( ) != sourceState {
return true
}
select {
case <- ctx . Done ( ) :
return false
case <- ch :
return true
}
}
// GetState returns the connectivity.State of ClientConn.
// This is an EXPERIMENTAL API.
func ( cc * ClientConn ) GetState ( ) connectivity . State {
return cc . csMgr . getState ( )
}
func ( cc * ClientConn ) scWatcher ( ) {
for {
select {
case sc , ok := <- cc . dopts . scChan :
if ! ok {
return
}
cc . mu . Lock ( )
// TODO: load balance policy runtime change is ignored.
// We may revisit this decision in the future.
cc . sc = & sc
cc . mu . Unlock ( )
case <- cc . ctx . Done ( ) :
return
}
}
}
// waitForResolvedAddrs blocks until the resolver has provided addresses or the
// context expires. Returns nil unless the context expires first; otherwise
// returns a status error based on the context.
func ( cc * ClientConn ) waitForResolvedAddrs ( ctx context . Context ) error {
// This is on the RPC path, so we use a fast path to avoid the
// more-expensive "select" below after the resolver has returned once.
if cc . firstResolveEvent . HasFired ( ) {
return nil
}
select {
case <- cc . firstResolveEvent . Done ( ) :
return nil
case <- ctx . Done ( ) :
return status . FromContextError ( ctx . Err ( ) ) . Err ( )
case <- cc . ctx . Done ( ) :
return ErrClientConnClosing
}
}
// gRPC should resort to default service config when:
// * resolver service config is disabled
// * or, resolver does not return a service config or returns an invalid one.
func ( cc * ClientConn ) fallbackToDefaultServiceConfig ( sc string ) bool {
if cc . dopts . disableServiceConfig {
return true
}
// The logic below is temporary, will be removed once we change the resolver.State ServiceConfig field type.
// Right now, we assume that empty service config string means resolver does not return a config.
if sc == "" {
return true
}
// TODO: the logic below is temporary. Once we finish the logic to validate service config
// in resolver, we will replace the logic below.
_ , err := parseServiceConfig ( sc )
return err != nil
}
func ( cc * ClientConn ) updateResolverState ( s resolver . State ) error {
cc . mu . Lock ( )
defer cc . mu . Unlock ( )
// Check if the ClientConn is already closed. Some fields (e.g.
// balancerWrapper) are set to nil when closing the ClientConn, and could
// cause nil pointer panic if we don't have this check.
if cc . conns == nil {
return nil
}
if cc . fallbackToDefaultServiceConfig ( s . ServiceConfig ) {
if cc . dopts . defaultServiceConfig != nil && cc . sc == nil {
cc . applyServiceConfig ( cc . dopts . defaultServiceConfig )
}
} else {
// TODO: the parsing logic below will be moved inside resolver.
sc , err := parseServiceConfig ( s . ServiceConfig )
if err != nil {
return err
}
if cc . sc == nil || cc . sc . rawJSONString != s . ServiceConfig {
cc . applyServiceConfig ( sc )
}
}
// update the service config that will be sent to balancer.
if cc . sc != nil {
s . ServiceConfig = cc . sc . rawJSONString
}
if cc . dopts . balancerBuilder == nil {
// Only look at balancer types and switch balancer if balancer dial
// option is not set.
var isGRPCLB bool
for _ , a := range s . Addresses {
if a . Type == resolver . GRPCLB {
isGRPCLB = true
break
}
}
var newBalancerName string
// TODO: use new loadBalancerConfig field with appropriate priority.
if isGRPCLB {
newBalancerName = grpclbName
} else if cc . sc != nil && cc . sc . LB != nil {
newBalancerName = * cc . sc . LB
} else {
newBalancerName = PickFirstBalancerName
}
cc . switchBalancer ( newBalancerName )
} else if cc . balancerWrapper == nil {
// Balancer dial option was set, and this is the first time handling
// resolved addresses. Build a balancer with dopts.balancerBuilder.
2019-06-27 11:30:45 +00:00
cc . curBalancerName = cc . dopts . balancerBuilder . Name ( )
2019-04-13 12:47:57 +00:00
cc . balancerWrapper = newCCBalancerWrapper ( cc , cc . dopts . balancerBuilder , cc . balancerBuildOpts )
}
cc . balancerWrapper . updateResolverState ( s )
cc . firstResolveEvent . Fire ( )
return nil
}
// switchBalancer starts the switching from current balancer to the balancer
// with the given name.
//
// It will NOT send the current address list to the new balancer. If needed,
// caller of this function should send address list to the new balancer after
// this function returns.
//
// Caller must hold cc.mu.
func ( cc * ClientConn ) switchBalancer ( name string ) {
if strings . ToLower ( cc . curBalancerName ) == strings . ToLower ( name ) {
return
}
grpclog . Infof ( "ClientConn switching balancer to %q" , name )
if cc . dopts . balancerBuilder != nil {
grpclog . Infoln ( "ignoring balancer switching: Balancer DialOption used instead" )
return
}
if cc . balancerWrapper != nil {
cc . balancerWrapper . close ( )
}
builder := balancer . Get ( name )
if channelz . IsOn ( ) {
if builder == nil {
channelz . AddTraceEvent ( cc . channelzID , & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Channel switches to new LB policy %q due to fallback from invalid balancer name" , PickFirstBalancerName ) ,
Severity : channelz . CtWarning ,
} )
} else {
channelz . AddTraceEvent ( cc . channelzID , & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Channel switches to new LB policy %q" , name ) ,
Severity : channelz . CtINFO ,
} )
}
}
if builder == nil {
grpclog . Infof ( "failed to get balancer builder for: %v, using pick_first instead" , name )
builder = newPickfirstBuilder ( )
}
cc . curBalancerName = builder . Name ( )
cc . balancerWrapper = newCCBalancerWrapper ( cc , builder , cc . balancerBuildOpts )
}
func ( cc * ClientConn ) handleSubConnStateChange ( sc balancer . SubConn , s connectivity . State ) {
cc . mu . Lock ( )
if cc . conns == nil {
cc . mu . Unlock ( )
return
}
// TODO(bar switching) send updates to all balancer wrappers when balancer
// gracefully switching is supported.
cc . balancerWrapper . handleSubConnStateChange ( sc , s )
cc . mu . Unlock ( )
}
// newAddrConn creates an addrConn for addrs and adds it to cc.conns.
//
// Caller needs to make sure len(addrs) > 0.
func ( cc * ClientConn ) newAddrConn ( addrs [ ] resolver . Address , opts balancer . NewSubConnOptions ) ( * addrConn , error ) {
ac := & addrConn {
cc : cc ,
addrs : addrs ,
scopts : opts ,
dopts : cc . dopts ,
czData : new ( channelzData ) ,
resetBackoff : make ( chan struct { } ) ,
}
ac . ctx , ac . cancel = context . WithCancel ( cc . ctx )
// Track ac in cc. This needs to be done before any getTransport(...) is called.
cc . mu . Lock ( )
if cc . conns == nil {
cc . mu . Unlock ( )
return nil , ErrClientConnClosing
}
if channelz . IsOn ( ) {
ac . channelzID = channelz . RegisterSubChannel ( ac , cc . channelzID , "" )
channelz . AddTraceEvent ( ac . channelzID , & channelz . TraceEventDesc {
Desc : "Subchannel Created" ,
Severity : channelz . CtINFO ,
Parent : & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Subchannel(id:%d) created" , ac . channelzID ) ,
Severity : channelz . CtINFO ,
} ,
} )
}
cc . conns [ ac ] = struct { } { }
cc . mu . Unlock ( )
return ac , nil
}
// removeAddrConn removes the addrConn in the subConn from clientConn.
// It also tears down the ac with the given error.
func ( cc * ClientConn ) removeAddrConn ( ac * addrConn , err error ) {
cc . mu . Lock ( )
if cc . conns == nil {
cc . mu . Unlock ( )
return
}
delete ( cc . conns , ac )
cc . mu . Unlock ( )
ac . tearDown ( err )
}
func ( cc * ClientConn ) channelzMetric ( ) * channelz . ChannelInternalMetric {
return & channelz . ChannelInternalMetric {
State : cc . GetState ( ) ,
Target : cc . target ,
CallsStarted : atomic . LoadInt64 ( & cc . czData . callsStarted ) ,
CallsSucceeded : atomic . LoadInt64 ( & cc . czData . callsSucceeded ) ,
CallsFailed : atomic . LoadInt64 ( & cc . czData . callsFailed ) ,
LastCallStartedTimestamp : time . Unix ( 0 , atomic . LoadInt64 ( & cc . czData . lastCallStartedTime ) ) ,
}
}
// Target returns the target string of the ClientConn.
// This is an EXPERIMENTAL API.
func ( cc * ClientConn ) Target ( ) string {
return cc . target
}
func ( cc * ClientConn ) incrCallsStarted ( ) {
atomic . AddInt64 ( & cc . czData . callsStarted , 1 )
atomic . StoreInt64 ( & cc . czData . lastCallStartedTime , time . Now ( ) . UnixNano ( ) )
}
func ( cc * ClientConn ) incrCallsSucceeded ( ) {
atomic . AddInt64 ( & cc . czData . callsSucceeded , 1 )
}
func ( cc * ClientConn ) incrCallsFailed ( ) {
atomic . AddInt64 ( & cc . czData . callsFailed , 1 )
}
// connect starts creating a transport.
// It does nothing if the ac is not IDLE.
// TODO(bar) Move this to the addrConn section.
func ( ac * addrConn ) connect ( ) error {
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return errConnClosing
}
if ac . state != connectivity . Idle {
ac . mu . Unlock ( )
return nil
}
ac . updateConnectivityState ( connectivity . Connecting )
ac . mu . Unlock ( )
// Start a goroutine connecting to the server asynchronously.
go ac . resetTransport ( )
return nil
}
// tryUpdateAddrs tries to update ac.addrs with the new addresses list.
//
// It checks whether current connected address of ac is in the new addrs list.
// - If true, it updates ac.addrs and returns true. The ac will keep using
// the existing connection.
// - If false, it does nothing and returns false.
func ( ac * addrConn ) tryUpdateAddrs ( addrs [ ] resolver . Address ) bool {
ac . mu . Lock ( )
defer ac . mu . Unlock ( )
grpclog . Infof ( "addrConn: tryUpdateAddrs curAddr: %v, addrs: %v" , ac . curAddr , addrs )
if ac . state == connectivity . Shutdown {
ac . addrs = addrs
return true
}
// Unless we're busy reconnecting already, let's reconnect from the top of
// the list.
if ac . state != connectivity . Ready {
return false
}
var curAddrFound bool
for _ , a := range addrs {
if reflect . DeepEqual ( ac . curAddr , a ) {
curAddrFound = true
break
}
}
grpclog . Infof ( "addrConn: tryUpdateAddrs curAddrFound: %v" , curAddrFound )
if curAddrFound {
ac . addrs = addrs
}
return curAddrFound
}
// GetMethodConfig gets the method config of the input method.
// If there's an exact match for input method (i.e. /service/method), we return
// the corresponding MethodConfig.
// If there isn't an exact match for the input method, we look for the default config
// under the service (i.e /service/). If there is a default MethodConfig for
// the service, we return it.
// Otherwise, we return an empty MethodConfig.
func ( cc * ClientConn ) GetMethodConfig ( method string ) MethodConfig {
// TODO: Avoid the locking here.
cc . mu . RLock ( )
defer cc . mu . RUnlock ( )
if cc . sc == nil {
return MethodConfig { }
}
m , ok := cc . sc . Methods [ method ]
if ! ok {
i := strings . LastIndex ( method , "/" )
m = cc . sc . Methods [ method [ : i + 1 ] ]
}
return m
}
func ( cc * ClientConn ) healthCheckConfig ( ) * healthCheckConfig {
cc . mu . RLock ( )
defer cc . mu . RUnlock ( )
if cc . sc == nil {
return nil
}
return cc . sc . healthCheckConfig
}
func ( cc * ClientConn ) getTransport ( ctx context . Context , failfast bool , method string ) ( transport . ClientTransport , func ( balancer . DoneInfo ) , error ) {
t , done , err := cc . blockingpicker . pick ( ctx , failfast , balancer . PickOptions {
FullMethodName : method ,
} )
if err != nil {
return nil , nil , toRPCErr ( err )
}
return t , done , nil
}
func ( cc * ClientConn ) applyServiceConfig ( sc * ServiceConfig ) error {
if sc == nil {
// should never reach here.
return fmt . Errorf ( "got nil pointer for service config" )
}
cc . sc = sc
if cc . sc . retryThrottling != nil {
newThrottler := & retryThrottler {
tokens : cc . sc . retryThrottling . MaxTokens ,
max : cc . sc . retryThrottling . MaxTokens ,
thresh : cc . sc . retryThrottling . MaxTokens / 2 ,
ratio : cc . sc . retryThrottling . TokenRatio ,
}
cc . retryThrottler . Store ( newThrottler )
} else {
cc . retryThrottler . Store ( ( * retryThrottler ) ( nil ) )
}
return nil
}
func ( cc * ClientConn ) resolveNow ( o resolver . ResolveNowOption ) {
cc . mu . RLock ( )
r := cc . resolverWrapper
cc . mu . RUnlock ( )
if r == nil {
return
}
go r . resolveNow ( o )
}
// ResetConnectBackoff wakes up all subchannels in transient failure and causes
// them to attempt another connection immediately. It also resets the backoff
// times used for subsequent attempts regardless of the current state.
//
// In general, this function should not be used. Typical service or network
// outages result in a reasonable client reconnection strategy by default.
// However, if a previously unavailable network becomes available, this may be
// used to trigger an immediate reconnect.
//
// This API is EXPERIMENTAL.
func ( cc * ClientConn ) ResetConnectBackoff ( ) {
cc . mu . Lock ( )
defer cc . mu . Unlock ( )
for ac := range cc . conns {
ac . resetConnectBackoff ( )
}
}
// Close tears down the ClientConn and all underlying connections.
func ( cc * ClientConn ) Close ( ) error {
defer cc . cancel ( )
cc . mu . Lock ( )
if cc . conns == nil {
cc . mu . Unlock ( )
return ErrClientConnClosing
}
conns := cc . conns
cc . conns = nil
cc . csMgr . updateState ( connectivity . Shutdown )
rWrapper := cc . resolverWrapper
cc . resolverWrapper = nil
bWrapper := cc . balancerWrapper
cc . balancerWrapper = nil
cc . mu . Unlock ( )
cc . blockingpicker . close ( )
if rWrapper != nil {
rWrapper . close ( )
}
if bWrapper != nil {
bWrapper . close ( )
}
for ac := range conns {
ac . tearDown ( ErrClientConnClosing )
}
if channelz . IsOn ( ) {
ted := & channelz . TraceEventDesc {
Desc : "Channel Deleted" ,
Severity : channelz . CtINFO ,
}
if cc . dopts . channelzParentID != 0 {
ted . Parent = & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Nested channel(id:%d) deleted" , cc . channelzID ) ,
Severity : channelz . CtINFO ,
}
}
channelz . AddTraceEvent ( cc . channelzID , ted )
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
// the entity being deleted, and thus prevent it from being deleted right away.
channelz . RemoveEntry ( cc . channelzID )
}
return nil
}
// addrConn is a network connection to a given address.
type addrConn struct {
ctx context . Context
cancel context . CancelFunc
cc * ClientConn
dopts dialOptions
acbw balancer . SubConn
scopts balancer . NewSubConnOptions
// transport is set when there's a viable transport (note: ac state may not be READY as LB channel
// health checking may require server to report healthy to set ac to READY), and is reset
// to nil when the current transport should no longer be used to create a stream (e.g. after GoAway
// is received, transport is closed, ac has been torn down).
transport transport . ClientTransport // The current transport.
mu sync . Mutex
curAddr resolver . Address // The current address.
addrs [ ] resolver . Address // All addresses that the resolver resolved to.
// Use updateConnectivityState for updating addrConn's connectivity state.
state connectivity . State
backoffIdx int // Needs to be stateful for resetConnectBackoff.
resetBackoff chan struct { }
channelzID int64 // channelz unique identification number.
czData * channelzData
}
// Note: this requires a lock on ac.mu.
func ( ac * addrConn ) updateConnectivityState ( s connectivity . State ) {
if ac . state == s {
return
}
updateMsg := fmt . Sprintf ( "Subchannel Connectivity change to %v" , s )
ac . state = s
if channelz . IsOn ( ) {
channelz . AddTraceEvent ( ac . channelzID , & channelz . TraceEventDesc {
Desc : updateMsg ,
Severity : channelz . CtINFO ,
} )
}
ac . cc . handleSubConnStateChange ( ac . acbw , s )
}
// adjustParams updates parameters used to create transports upon
// receiving a GoAway.
func ( ac * addrConn ) adjustParams ( r transport . GoAwayReason ) {
switch r {
case transport . GoAwayTooManyPings :
v := 2 * ac . dopts . copts . KeepaliveParams . Time
ac . cc . mu . Lock ( )
if v > ac . cc . mkp . Time {
ac . cc . mkp . Time = v
}
ac . cc . mu . Unlock ( )
}
}
func ( ac * addrConn ) resetTransport ( ) {
for i := 0 ; ; i ++ {
if i > 0 {
ac . cc . resolveNow ( resolver . ResolveNowOption { } )
}
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return
}
addrs := ac . addrs
backoffFor := ac . dopts . bs . Backoff ( ac . backoffIdx )
// This will be the duration that dial gets to finish.
dialDuration := minConnectTimeout
if ac . dopts . minConnectTimeout != nil {
dialDuration = ac . dopts . minConnectTimeout ( )
}
if dialDuration < backoffFor {
// Give dial more time as we keep failing to connect.
dialDuration = backoffFor
}
// We can potentially spend all the time trying the first address, and
// if the server accepts the connection and then hangs, the following
// addresses will never be tried.
//
// The spec doesn't mention what should be done for multiple addresses.
// https://github.com/grpc/grpc/blob/master/doc/connection-backoff.md#proposed-backoff-algorithm
connectDeadline := time . Now ( ) . Add ( dialDuration )
ac . mu . Unlock ( )
newTr , addr , reconnect , err := ac . tryAllAddrs ( addrs , connectDeadline )
if err != nil {
// After exhausting all addresses, the addrConn enters
// TRANSIENT_FAILURE.
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return
}
ac . updateConnectivityState ( connectivity . TransientFailure )
// Backoff.
b := ac . resetBackoff
ac . mu . Unlock ( )
timer := time . NewTimer ( backoffFor )
select {
case <- timer . C :
ac . mu . Lock ( )
ac . backoffIdx ++
ac . mu . Unlock ( )
case <- b :
timer . Stop ( )
case <- ac . ctx . Done ( ) :
timer . Stop ( )
return
}
continue
}
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
newTr . Close ( )
ac . mu . Unlock ( )
return
}
ac . curAddr = addr
ac . transport = newTr
ac . backoffIdx = 0
healthCheckConfig := ac . cc . healthCheckConfig ( )
// LB channel health checking is only enabled when all the four requirements below are met:
// 1. it is not disabled by the user with the WithDisableHealthCheck DialOption,
// 2. the internal.HealthCheckFunc is set by importing the grpc/healthcheck package,
// 3. a service config with non-empty healthCheckConfig field is provided,
// 4. the current load balancer allows it.
hctx , hcancel := context . WithCancel ( ac . ctx )
healthcheckManagingState := false
if ! ac . cc . dopts . disableHealthCheck && healthCheckConfig != nil && ac . scopts . HealthCheckEnabled {
if ac . cc . dopts . healthCheckFunc == nil {
// TODO: add a link to the health check doc in the error message.
grpclog . Error ( "the client side LB channel health check function has not been set." )
} else {
// TODO(deklerk) refactor to just return transport
go ac . startHealthCheck ( hctx , newTr , addr , healthCheckConfig . ServiceName )
healthcheckManagingState = true
}
}
if ! healthcheckManagingState {
ac . updateConnectivityState ( connectivity . Ready )
}
ac . mu . Unlock ( )
// Block until the created transport is down. And when this happens,
// we restart from the top of the addr list.
<- reconnect . Done ( )
hcancel ( )
// Need to reconnect after a READY, the addrConn enters
// TRANSIENT_FAILURE.
//
// This will set addrConn to TRANSIENT_FAILURE for a very short period
// of time, and turns CONNECTING. It seems reasonable to skip this, but
// READY-CONNECTING is not a valid transition.
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return
}
ac . updateConnectivityState ( connectivity . TransientFailure )
ac . mu . Unlock ( )
}
}
// tryAllAddrs tries to creates a connection to the addresses, and stop when at the
// first successful one. It returns the transport, the address and a Event in
// the successful case. The Event fires when the returned transport disconnects.
func ( ac * addrConn ) tryAllAddrs ( addrs [ ] resolver . Address , connectDeadline time . Time ) ( transport . ClientTransport , resolver . Address , * grpcsync . Event , error ) {
for _ , addr := range addrs {
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return nil , resolver . Address { } , nil , errConnClosing
}
ac . updateConnectivityState ( connectivity . Connecting )
ac . transport = nil
ac . cc . mu . RLock ( )
ac . dopts . copts . KeepaliveParams = ac . cc . mkp
ac . cc . mu . RUnlock ( )
copts := ac . dopts . copts
if ac . scopts . CredsBundle != nil {
copts . CredsBundle = ac . scopts . CredsBundle
}
ac . mu . Unlock ( )
if channelz . IsOn ( ) {
channelz . AddTraceEvent ( ac . channelzID , & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Subchannel picks a new address %q to connect" , addr . Addr ) ,
Severity : channelz . CtINFO ,
} )
}
newTr , reconnect , err := ac . createTransport ( addr , copts , connectDeadline )
if err == nil {
return newTr , addr , reconnect , nil
}
ac . cc . blockingpicker . updateConnectionError ( err )
}
// Couldn't connect to any address.
return nil , resolver . Address { } , nil , fmt . Errorf ( "couldn't connect to any address" )
}
// createTransport creates a connection to addr. It returns the transport and a
// Event in the successful case. The Event fires when the returned transport
// disconnects.
func ( ac * addrConn ) createTransport ( addr resolver . Address , copts transport . ConnectOptions , connectDeadline time . Time ) ( transport . ClientTransport , * grpcsync . Event , error ) {
prefaceReceived := make ( chan struct { } )
onCloseCalled := make ( chan struct { } )
reconnect := grpcsync . NewEvent ( )
target := transport . TargetInfo {
Addr : addr . Addr ,
Metadata : addr . Metadata ,
Authority : ac . cc . authority ,
}
onGoAway := func ( r transport . GoAwayReason ) {
ac . mu . Lock ( )
ac . adjustParams ( r )
ac . mu . Unlock ( )
reconnect . Fire ( )
}
onClose := func ( ) {
close ( onCloseCalled )
reconnect . Fire ( )
}
onPrefaceReceipt := func ( ) {
close ( prefaceReceived )
}
connectCtx , cancel := context . WithDeadline ( ac . ctx , connectDeadline )
defer cancel ( )
if channelz . IsOn ( ) {
copts . ChannelzParentID = ac . channelzID
}
newTr , err := transport . NewClientTransport ( connectCtx , ac . cc . ctx , target , copts , onPrefaceReceipt , onGoAway , onClose )
if err != nil {
// newTr is either nil, or closed.
grpclog . Warningf ( "grpc: addrConn.createTransport failed to connect to %v. Err :%v. Reconnecting..." , addr , err )
return nil , nil , err
}
if ac . dopts . reqHandshake == envconfig . RequireHandshakeOn {
select {
case <- time . After ( connectDeadline . Sub ( time . Now ( ) ) ) :
// We didn't get the preface in time.
newTr . Close ( )
grpclog . Warningf ( "grpc: addrConn.createTransport failed to connect to %v: didn't receive server preface in time. Reconnecting..." , addr )
return nil , nil , errors . New ( "timed out waiting for server handshake" )
case <- prefaceReceived :
// We got the preface - huzzah! things are good.
case <- onCloseCalled :
// The transport has already closed - noop.
return nil , nil , errors . New ( "connection closed" )
// TODO(deklerk) this should bail on ac.ctx.Done(). Add a test and fix.
}
}
return newTr , reconnect , nil
}
func ( ac * addrConn ) startHealthCheck ( ctx context . Context , newTr transport . ClientTransport , addr resolver . Address , serviceName string ) {
// Set up the health check helper functions
newStream := func ( ) ( interface { } , error ) {
return ac . newClientStream ( ctx , & StreamDesc { ServerStreams : true } , "/grpc.health.v1.Health/Watch" , newTr )
}
firstReady := true
reportHealth := func ( ok bool ) {
ac . mu . Lock ( )
defer ac . mu . Unlock ( )
if ac . transport != newTr {
return
}
if ok {
if firstReady {
firstReady = false
ac . curAddr = addr
}
ac . updateConnectivityState ( connectivity . Ready )
} else {
ac . updateConnectivityState ( connectivity . TransientFailure )
}
}
err := ac . cc . dopts . healthCheckFunc ( ctx , newStream , reportHealth , serviceName )
if err != nil {
if status . Code ( err ) == codes . Unimplemented {
if channelz . IsOn ( ) {
channelz . AddTraceEvent ( ac . channelzID , & channelz . TraceEventDesc {
Desc : "Subchannel health check is unimplemented at server side, thus health check is disabled" ,
Severity : channelz . CtError ,
} )
}
grpclog . Error ( "Subchannel health check is unimplemented at server side, thus health check is disabled" )
} else {
grpclog . Errorf ( "HealthCheckFunc exits with unexpected error %v" , err )
}
}
}
func ( ac * addrConn ) resetConnectBackoff ( ) {
ac . mu . Lock ( )
close ( ac . resetBackoff )
ac . backoffIdx = 0
ac . resetBackoff = make ( chan struct { } )
ac . mu . Unlock ( )
}
// getReadyTransport returns the transport if ac's state is READY.
// Otherwise it returns nil, false.
// If ac's state is IDLE, it will trigger ac to connect.
func ( ac * addrConn ) getReadyTransport ( ) ( transport . ClientTransport , bool ) {
ac . mu . Lock ( )
if ac . state == connectivity . Ready && ac . transport != nil {
t := ac . transport
ac . mu . Unlock ( )
return t , true
}
var idle bool
if ac . state == connectivity . Idle {
idle = true
}
ac . mu . Unlock ( )
// Trigger idle ac to connect.
if idle {
ac . connect ( )
}
return nil , false
}
// tearDown starts to tear down the addrConn.
// TODO(zhaoq): Make this synchronous to avoid unbounded memory consumption in
// some edge cases (e.g., the caller opens and closes many addrConn's in a
// tight loop.
// tearDown doesn't remove ac from ac.cc.conns.
func ( ac * addrConn ) tearDown ( err error ) {
ac . mu . Lock ( )
if ac . state == connectivity . Shutdown {
ac . mu . Unlock ( )
return
}
curTr := ac . transport
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 cancelation / etc.
ac . updateConnectivityState ( connectivity . Shutdown )
ac . cancel ( )
ac . curAddr = resolver . Address { }
if err == errConnDrain && curTr != nil {
// GracefulClose(...) may be executed multiple times when
// i) receiving multiple GoAway frames from the server; or
// ii) there are concurrent name resolver/Balancer triggered
// address removal and GoAway.
// We have to unlock and re-lock here because GracefulClose => Close => onClose, which requires locking ac.mu.
ac . mu . Unlock ( )
curTr . GracefulClose ( )
ac . mu . Lock ( )
}
if channelz . IsOn ( ) {
channelz . AddTraceEvent ( ac . channelzID , & channelz . TraceEventDesc {
Desc : "Subchannel Deleted" ,
Severity : channelz . CtINFO ,
Parent : & channelz . TraceEventDesc {
Desc : fmt . Sprintf ( "Subchanel(id:%d) deleted" , ac . channelzID ) ,
Severity : channelz . CtINFO ,
} ,
} )
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
// the entity beng deleted, and thus prevent it from being deleted right away.
channelz . RemoveEntry ( ac . channelzID )
}
ac . mu . Unlock ( )
}
func ( ac * addrConn ) getState ( ) connectivity . State {
ac . mu . Lock ( )
defer ac . mu . Unlock ( )
return ac . state
}
func ( ac * addrConn ) ChannelzMetric ( ) * channelz . ChannelInternalMetric {
ac . mu . Lock ( )
addr := ac . curAddr . Addr
ac . mu . Unlock ( )
return & channelz . ChannelInternalMetric {
State : ac . getState ( ) ,
Target : addr ,
CallsStarted : atomic . LoadInt64 ( & ac . czData . callsStarted ) ,
CallsSucceeded : atomic . LoadInt64 ( & ac . czData . callsSucceeded ) ,
CallsFailed : atomic . LoadInt64 ( & ac . czData . callsFailed ) ,
LastCallStartedTimestamp : time . Unix ( 0 , atomic . LoadInt64 ( & ac . czData . lastCallStartedTime ) ) ,
}
}
func ( ac * addrConn ) incrCallsStarted ( ) {
atomic . AddInt64 ( & ac . czData . callsStarted , 1 )
atomic . StoreInt64 ( & ac . czData . lastCallStartedTime , time . Now ( ) . UnixNano ( ) )
}
func ( ac * addrConn ) incrCallsSucceeded ( ) {
atomic . AddInt64 ( & ac . czData . callsSucceeded , 1 )
}
func ( ac * addrConn ) incrCallsFailed ( ) {
atomic . AddInt64 ( & ac . czData . callsFailed , 1 )
}
type retryThrottler struct {
max float64
thresh float64
ratio float64
mu sync . Mutex
tokens float64 // TODO(dfawley): replace with atomic and remove lock.
}
// throttle subtracts a retry token from the pool and returns whether a retry
// should be throttled (disallowed) based upon the retry throttling policy in
// the service config.
func ( rt * retryThrottler ) throttle ( ) bool {
if rt == nil {
return false
}
rt . mu . Lock ( )
defer rt . mu . Unlock ( )
rt . tokens --
if rt . tokens < 0 {
rt . tokens = 0
}
return rt . tokens <= rt . thresh
}
func ( rt * retryThrottler ) successfulRPC ( ) {
if rt == nil {
return
}
rt . mu . Lock ( )
defer rt . mu . Unlock ( )
rt . tokens += rt . ratio
if rt . tokens > rt . max {
rt . tokens = rt . max
}
}
type channelzChannel struct {
cc * ClientConn
}
func ( c * channelzChannel ) ChannelzMetric ( ) * channelz . ChannelInternalMetric {
return c . cc . channelzMetric ( )
}
// ErrClientConnTimeout indicates that the ClientConn cannot establish the
// underlying connections within the specified timeout.
//
// Deprecated: This error is never returned by grpc and should not be
// referenced by users.
var ErrClientConnTimeout = errors . New ( "grpc: timed out when dialing" )