1/* 2 * 3 * Copyright 2017 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 19package grpc 20 21import ( 22 "fmt" 23 "strings" 24 "sync" 25 26 "google.golang.org/grpc/balancer" 27 "google.golang.org/grpc/credentials" 28 "google.golang.org/grpc/internal/channelz" 29 "google.golang.org/grpc/internal/grpcsync" 30 "google.golang.org/grpc/resolver" 31 "google.golang.org/grpc/serviceconfig" 32) 33 34// ccResolverWrapper is a wrapper on top of cc for resolvers. 35// It implements resolver.ClientConn interface. 36type ccResolverWrapper struct { 37 cc *ClientConn 38 resolverMu sync.Mutex 39 resolver resolver.Resolver 40 done *grpcsync.Event 41 curState resolver.State 42} 43 44// newCCResolverWrapper uses the resolver.Builder to build a Resolver and 45// returns a ccResolverWrapper object which wraps the newly built resolver. 46func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) { 47 ccr := &ccResolverWrapper{ 48 cc: cc, 49 done: grpcsync.NewEvent(), 50 } 51 52 var credsClone credentials.TransportCredentials 53 if creds := cc.dopts.copts.TransportCredentials; creds != nil { 54 credsClone = creds.Clone() 55 } 56 rbo := resolver.BuildOptions{ 57 DisableServiceConfig: cc.dopts.disableServiceConfig, 58 DialCreds: credsClone, 59 CredsBundle: cc.dopts.copts.CredsBundle, 60 Dialer: cc.dopts.copts.Dialer, 61 } 62 63 var err error 64 // We need to hold the lock here while we assign to the ccr.resolver field 65 // to guard against a data race caused by the following code path, 66 // rb.Build-->ccr.ReportError-->ccr.poll-->ccr.resolveNow, would end up 67 // accessing ccr.resolver which is being assigned here. 68 ccr.resolverMu.Lock() 69 defer ccr.resolverMu.Unlock() 70 ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo) 71 if err != nil { 72 return nil, err 73 } 74 return ccr, nil 75} 76 77func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOptions) { 78 ccr.resolverMu.Lock() 79 if !ccr.done.HasFired() { 80 ccr.resolver.ResolveNow(o) 81 } 82 ccr.resolverMu.Unlock() 83} 84 85func (ccr *ccResolverWrapper) close() { 86 ccr.resolverMu.Lock() 87 ccr.resolver.Close() 88 ccr.done.Fire() 89 ccr.resolverMu.Unlock() 90} 91 92func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error { 93 if ccr.done.HasFired() { 94 return nil 95 } 96 channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s) 97 if channelz.IsOn() { 98 ccr.addChannelzTraceEvent(s) 99 } 100 ccr.curState = s 101 if err := ccr.cc.updateResolverState(ccr.curState, nil); err == balancer.ErrBadResolverState { 102 return balancer.ErrBadResolverState 103 } 104 return nil 105} 106 107func (ccr *ccResolverWrapper) ReportError(err error) { 108 if ccr.done.HasFired() { 109 return 110 } 111 channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: reporting error to cc: %v", err) 112 ccr.cc.updateResolverState(resolver.State{}, err) 113} 114 115// NewAddress is called by the resolver implementation to send addresses to gRPC. 116func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) { 117 if ccr.done.HasFired() { 118 return 119 } 120 channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending new addresses to cc: %v", addrs) 121 if channelz.IsOn() { 122 ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig}) 123 } 124 ccr.curState.Addresses = addrs 125 ccr.cc.updateResolverState(ccr.curState, nil) 126} 127 128// NewServiceConfig is called by the resolver implementation to send service 129// configs to gRPC. 130func (ccr *ccResolverWrapper) NewServiceConfig(sc string) { 131 if ccr.done.HasFired() { 132 return 133 } 134 channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: got new service config: %v", sc) 135 if ccr.cc.dopts.disableServiceConfig { 136 channelz.Info(logger, ccr.cc.channelzID, "Service config lookups disabled; ignoring config") 137 return 138 } 139 scpr := parseServiceConfig(sc) 140 if scpr.Err != nil { 141 channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: error parsing service config: %v", scpr.Err) 142 return 143 } 144 if channelz.IsOn() { 145 ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr}) 146 } 147 ccr.curState.ServiceConfig = scpr 148 ccr.cc.updateResolverState(ccr.curState, nil) 149} 150 151func (ccr *ccResolverWrapper) ParseServiceConfig(scJSON string) *serviceconfig.ParseResult { 152 return parseServiceConfig(scJSON) 153} 154 155func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) { 156 var updates []string 157 var oldSC, newSC *ServiceConfig 158 var oldOK, newOK bool 159 if ccr.curState.ServiceConfig != nil { 160 oldSC, oldOK = ccr.curState.ServiceConfig.Config.(*ServiceConfig) 161 } 162 if s.ServiceConfig != nil { 163 newSC, newOK = s.ServiceConfig.Config.(*ServiceConfig) 164 } 165 if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) { 166 updates = append(updates, "service config updated") 167 } 168 if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 { 169 updates = append(updates, "resolver returned an empty address list") 170 } else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 { 171 updates = append(updates, "resolver returned new addresses") 172 } 173 channelz.AddTraceEvent(logger, ccr.cc.channelzID, 0, &channelz.TraceEventDesc{ 174 Desc: fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")), 175 Severity: channelz.CtInfo, 176 }) 177} 178