blob: edfda866c001cfc0b2c08b3925af93173dda7424 [file] [log] [blame]
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -07001/*
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"
Arjun E K57a7fcb2020-01-30 06:44:45 +000024 "sync"
25 "time"
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070026
Arjun E K57a7fcb2020-01-30 06:44:45 +000027 "google.golang.org/grpc/balancer"
28 "google.golang.org/grpc/credentials"
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070029 "google.golang.org/grpc/internal/channelz"
Arjun E K57a7fcb2020-01-30 06:44:45 +000030 "google.golang.org/grpc/internal/grpcsync"
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070031 "google.golang.org/grpc/resolver"
Arjun E K57a7fcb2020-01-30 06:44:45 +000032 "google.golang.org/grpc/serviceconfig"
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070033)
34
35// ccResolverWrapper is a wrapper on top of cc for resolvers.
Arjun E K57a7fcb2020-01-30 06:44:45 +000036// It implements resolver.ClientConn interface.
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070037type ccResolverWrapper struct {
Arjun E K57a7fcb2020-01-30 06:44:45 +000038 cc *ClientConn
39 resolverMu sync.Mutex
40 resolver resolver.Resolver
41 done *grpcsync.Event
42 curState resolver.State
43
44 pollingMu sync.Mutex
45 polling chan struct{}
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070046}
47
Arjun E K57a7fcb2020-01-30 06:44:45 +000048// newCCResolverWrapper uses the resolver.Builder to build a Resolver and
49// returns a ccResolverWrapper object which wraps the newly built resolver.
50func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
51 ccr := &ccResolverWrapper{
52 cc: cc,
53 done: grpcsync.NewEvent(),
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070054 }
55
Arjun E K57a7fcb2020-01-30 06:44:45 +000056 var credsClone credentials.TransportCredentials
57 if creds := cc.dopts.copts.TransportCredentials; creds != nil {
58 credsClone = creds.Clone()
59 }
60 rbo := resolver.BuildOptions{
61 DisableServiceConfig: cc.dopts.disableServiceConfig,
62 DialCreds: credsClone,
63 CredsBundle: cc.dopts.copts.CredsBundle,
64 Dialer: cc.dopts.copts.Dialer,
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070065 }
66
67 var err error
Arjun E K57a7fcb2020-01-30 06:44:45 +000068 // We need to hold the lock here while we assign to the ccr.resolver field
69 // to guard against a data race caused by the following code path,
70 // rb.Build-->ccr.ReportError-->ccr.poll-->ccr.resolveNow, would end up
71 // accessing ccr.resolver which is being assigned here.
72 ccr.resolverMu.Lock()
73 defer ccr.resolverMu.Unlock()
74 ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070075 if err != nil {
76 return nil, err
77 }
78 return ccr, nil
79}
80
Arjun E K57a7fcb2020-01-30 06:44:45 +000081func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOptions) {
82 ccr.resolverMu.Lock()
83 if !ccr.done.HasFired() {
84 ccr.resolver.ResolveNow(o)
85 }
86 ccr.resolverMu.Unlock()
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070087}
88
89func (ccr *ccResolverWrapper) close() {
Arjun E K57a7fcb2020-01-30 06:44:45 +000090 ccr.resolverMu.Lock()
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070091 ccr.resolver.Close()
Arjun E K57a7fcb2020-01-30 06:44:45 +000092 ccr.done.Fire()
93 ccr.resolverMu.Unlock()
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -070094}
95
Arjun E K57a7fcb2020-01-30 06:44:45 +000096// poll begins or ends asynchronous polling of the resolver based on whether
97// err is ErrBadResolverState.
98func (ccr *ccResolverWrapper) poll(err error) {
99 ccr.pollingMu.Lock()
100 defer ccr.pollingMu.Unlock()
101 if err != balancer.ErrBadResolverState {
102 // stop polling
103 if ccr.polling != nil {
104 close(ccr.polling)
105 ccr.polling = nil
106 }
107 return
108 }
109 if ccr.polling != nil {
110 // already polling
111 return
112 }
113 p := make(chan struct{})
114 ccr.polling = p
115 go func() {
116 for i := 0; ; i++ {
117 ccr.resolveNow(resolver.ResolveNowOptions{})
118 t := time.NewTimer(ccr.cc.dopts.resolveNowBackoff(i))
119 select {
120 case <-p:
121 t.Stop()
122 return
123 case <-ccr.done.Done():
124 // Resolver has been closed.
125 t.Stop()
126 return
127 case <-t.C:
128 select {
129 case <-p:
130 return
131 default:
132 }
133 // Timer expired; re-resolve.
134 }
135 }
136 }()
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700137}
138
139func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
Arjun E K57a7fcb2020-01-30 06:44:45 +0000140 if ccr.done.HasFired() {
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700141 return
142 }
amit.ghosh258d14c2020-10-02 15:13:38 +0200143 channelz.Infof(ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s)
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700144 if channelz.IsOn() {
145 ccr.addChannelzTraceEvent(s)
146 }
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700147 ccr.curState = s
Arjun E K57a7fcb2020-01-30 06:44:45 +0000148 ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
149}
150
151func (ccr *ccResolverWrapper) ReportError(err error) {
152 if ccr.done.HasFired() {
153 return
154 }
amit.ghosh258d14c2020-10-02 15:13:38 +0200155 channelz.Warningf(ccr.cc.channelzID, "ccResolverWrapper: reporting error to cc: %v", err)
Arjun E K57a7fcb2020-01-30 06:44:45 +0000156 ccr.poll(ccr.cc.updateResolverState(resolver.State{}, err))
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700157}
158
159// NewAddress is called by the resolver implementation to send addresses to gRPC.
160func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
Arjun E K57a7fcb2020-01-30 06:44:45 +0000161 if ccr.done.HasFired() {
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700162 return
163 }
amit.ghosh258d14c2020-10-02 15:13:38 +0200164 channelz.Infof(ccr.cc.channelzID, "ccResolverWrapper: sending new addresses to cc: %v", addrs)
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700165 if channelz.IsOn() {
166 ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
167 }
168 ccr.curState.Addresses = addrs
Arjun E K57a7fcb2020-01-30 06:44:45 +0000169 ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700170}
171
172// NewServiceConfig is called by the resolver implementation to send service
173// configs to gRPC.
174func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
Arjun E K57a7fcb2020-01-30 06:44:45 +0000175 if ccr.done.HasFired() {
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700176 return
177 }
amit.ghosh258d14c2020-10-02 15:13:38 +0200178 channelz.Infof(ccr.cc.channelzID, "ccResolverWrapper: got new service config: %v", sc)
Arjun E K57a7fcb2020-01-30 06:44:45 +0000179 if ccr.cc.dopts.disableServiceConfig {
amit.ghosh258d14c2020-10-02 15:13:38 +0200180 channelz.Info(ccr.cc.channelzID, "Service config lookups disabled; ignoring config")
Arjun E K57a7fcb2020-01-30 06:44:45 +0000181 return
182 }
183 scpr := parseServiceConfig(sc)
184 if scpr.Err != nil {
amit.ghosh258d14c2020-10-02 15:13:38 +0200185 channelz.Warningf(ccr.cc.channelzID, "ccResolverWrapper: error parsing service config: %v", scpr.Err)
Arjun E K57a7fcb2020-01-30 06:44:45 +0000186 ccr.poll(balancer.ErrBadResolverState)
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700187 return
188 }
189 if channelz.IsOn() {
Arjun E K57a7fcb2020-01-30 06:44:45 +0000190 ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr})
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700191 }
Arjun E K57a7fcb2020-01-30 06:44:45 +0000192 ccr.curState.ServiceConfig = scpr
193 ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
194}
195
196func (ccr *ccResolverWrapper) ParseServiceConfig(scJSON string) *serviceconfig.ParseResult {
197 return parseServiceConfig(scJSON)
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700198}
199
200func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
201 var updates []string
Arjun E K57a7fcb2020-01-30 06:44:45 +0000202 var oldSC, newSC *ServiceConfig
203 var oldOK, newOK bool
204 if ccr.curState.ServiceConfig != nil {
205 oldSC, oldOK = ccr.curState.ServiceConfig.Config.(*ServiceConfig)
206 }
207 if s.ServiceConfig != nil {
208 newSC, newOK = s.ServiceConfig.Config.(*ServiceConfig)
209 }
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700210 if oldOK != newOK || (oldOK && newOK && oldSC.rawJSONString != newSC.rawJSONString) {
211 updates = append(updates, "service config updated")
212 }
213 if len(ccr.curState.Addresses) > 0 && len(s.Addresses) == 0 {
214 updates = append(updates, "resolver returned an empty address list")
215 } else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
216 updates = append(updates, "resolver returned new addresses")
217 }
amit.ghosh258d14c2020-10-02 15:13:38 +0200218 channelz.AddTraceEvent(ccr.cc.channelzID, 0, &channelz.TraceEventDesc{
Matteo Scandoloa6a3aee2019-11-26 13:30:14 -0700219 Desc: fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
220 Severity: channelz.CtINFO,
221 })
222}