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 19 package base
20 21 import (
22 "errors"
23 "fmt"
24 25 "google.golang.org/grpc/balancer"
26 "google.golang.org/grpc/connectivity"
27 "google.golang.org/grpc/grpclog"
28 "google.golang.org/grpc/resolver"
29 )
30 31 var logger = grpclog.Component("balancer")
32 33 type baseBuilder struct {
34 name string
35 pickerBuilder PickerBuilder
36 config Config
37 }
38 39 func (bb *baseBuilder) Build(cc balancer.ClientConn, _ balancer.BuildOptions) balancer.Balancer {
40 bal := &baseBalancer{
41 cc: cc,
42 pickerBuilder: bb.pickerBuilder,
43 44 subConns: resolver.NewAddressMapV2[balancer.SubConn](),
45 scStates: make(map[balancer.SubConn]connectivity.State),
46 csEvltr: &balancer.ConnectivityStateEvaluator{},
47 config: bb.config,
48 state: connectivity.Connecting,
49 }
50 // Initialize picker to a picker that always returns
51 // ErrNoSubConnAvailable, because when state of a SubConn changes, we
52 // may call UpdateState with this picker.
53 bal.picker = NewErrPicker(balancer.ErrNoSubConnAvailable)
54 return bal
55 }
56 57 func (bb *baseBuilder) Name() string {
58 return bb.name
59 }
60 61 type baseBalancer struct {
62 cc balancer.ClientConn
63 pickerBuilder PickerBuilder
64 65 csEvltr *balancer.ConnectivityStateEvaluator
66 state connectivity.State
67 68 subConns *resolver.AddressMapV2[balancer.SubConn]
69 scStates map[balancer.SubConn]connectivity.State
70 picker balancer.Picker
71 config Config
72 73 resolverErr error // the last error reported by the resolver; cleared on successful resolution
74 connErr error // the last connection error; cleared upon leaving TransientFailure
75 }
76 77 func (b *baseBalancer) ResolverError(err error) {
78 b.resolverErr = err
79 if b.subConns.Len() == 0 {
80 b.state = connectivity.TransientFailure
81 }
82 83 if b.state != connectivity.TransientFailure {
84 // The picker will not change since the balancer does not currently
85 // report an error.
86 return
87 }
88 b.regeneratePicker()
89 b.cc.UpdateState(balancer.State{
90 ConnectivityState: b.state,
91 Picker: b.picker,
92 })
93 }
94 95 func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
96 // TODO: handle s.ResolverState.ServiceConfig?
97 if logger.V(2) {
98 logger.Info("base.baseBalancer: got new ClientConn state: ", s)
99 }
100 // Successful resolution; clear resolver error and ensure we return nil.
101 b.resolverErr = nil
102 // addrsSet is the set converted from addrs, it's used for quick lookup of an address.
103 addrsSet := resolver.NewAddressMapV2[any]()
104 for _, a := range s.ResolverState.Addresses {
105 addrsSet.Set(a, nil)
106 if _, ok := b.subConns.Get(a); !ok {
107 // a is a new address (not existing in b.subConns).
108 var sc balancer.SubConn
109 opts := balancer.NewSubConnOptions{
110 HealthCheckEnabled: b.config.HealthCheck,
111 StateListener: func(scs balancer.SubConnState) { b.updateSubConnState(sc, scs) },
112 }
113 sc, err := b.cc.NewSubConn([]resolver.Address{a}, opts)
114 if err != nil {
115 logger.Warningf("base.baseBalancer: failed to create new SubConn: %v", err)
116 continue
117 }
118 b.subConns.Set(a, sc)
119 b.scStates[sc] = connectivity.Idle
120 b.csEvltr.RecordTransition(connectivity.Shutdown, connectivity.Idle)
121 sc.Connect()
122 }
123 }
124 for _, a := range b.subConns.Keys() {
125 sc, _ := b.subConns.Get(a)
126 // a was removed by resolver.
127 if _, ok := addrsSet.Get(a); !ok {
128 sc.Shutdown()
129 b.subConns.Delete(a)
130 // Keep the state of this sc in b.scStates until sc's state becomes Shutdown.
131 // The entry will be deleted in updateSubConnState.
132 }
133 }
134 // If resolver state contains no addresses, return an error so ClientConn
135 // will trigger re-resolve. Also records this as a resolver error, so when
136 // the overall state turns transient failure, the error message will have
137 // the zero address information.
138 if len(s.ResolverState.Addresses) == 0 {
139 b.ResolverError(errors.New("produced zero addresses"))
140 return balancer.ErrBadResolverState
141 }
142 143 b.regeneratePicker()
144 b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
145 return nil
146 }
147 148 // mergeErrors builds an error from the last connection error and the last
149 // resolver error. Must only be called if b.state is TransientFailure.
150 func (b *baseBalancer) mergeErrors() error {
151 // connErr must always be non-nil unless there are no SubConns, in which
152 // case resolverErr must be non-nil.
153 if b.connErr == nil {
154 return fmt.Errorf("last resolver error: %v", b.resolverErr)
155 }
156 if b.resolverErr == nil {
157 return fmt.Errorf("last connection error: %v", b.connErr)
158 }
159 return fmt.Errorf("last connection error: %v; last resolver error: %v", b.connErr, b.resolverErr)
160 }
161 162 // regeneratePicker takes a snapshot of the balancer, and generates a picker
163 // from it. The picker is
164 // - errPicker if the balancer is in TransientFailure,
165 // - built by the pickerBuilder with all READY SubConns otherwise.
166 func (b *baseBalancer) regeneratePicker() {
167 if b.state == connectivity.TransientFailure {
168 b.picker = NewErrPicker(b.mergeErrors())
169 return
170 }
171 readySCs := make(map[balancer.SubConn]SubConnInfo)
172 173 // Filter out all ready SCs from full subConn map.
174 for _, addr := range b.subConns.Keys() {
175 sc, _ := b.subConns.Get(addr)
176 if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
177 readySCs[sc] = SubConnInfo{Address: addr}
178 }
179 }
180 b.picker = b.pickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs})
181 }
182 183 // UpdateSubConnState is a nop because a StateListener is always set in NewSubConn.
184 func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
185 logger.Errorf("base.baseBalancer: UpdateSubConnState(%v, %+v) called unexpectedly", sc, state)
186 }
187 188 func (b *baseBalancer) updateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
189 s := state.ConnectivityState
190 if logger.V(2) {
191 logger.Infof("base.baseBalancer: handle SubConn state change: %p, %v", sc, s)
192 }
193 oldS, ok := b.scStates[sc]
194 if !ok {
195 if logger.V(2) {
196 logger.Infof("base.baseBalancer: got state changes for an unknown SubConn: %p, %v", sc, s)
197 }
198 return
199 }
200 if oldS == connectivity.TransientFailure &&
201 (s == connectivity.Connecting || s == connectivity.Idle) {
202 // Once a subconn enters TRANSIENT_FAILURE, ignore subsequent IDLE or
203 // CONNECTING transitions to prevent the aggregated state from being
204 // always CONNECTING when many backends exist but are all down.
205 if s == connectivity.Idle {
206 sc.Connect()
207 }
208 return
209 }
210 b.scStates[sc] = s
211 switch s {
212 case connectivity.Idle:
213 sc.Connect()
214 case connectivity.Shutdown:
215 // When an address was removed by resolver, b called Shutdown but kept
216 // the sc's state in scStates. Remove state for this sc here.
217 delete(b.scStates, sc)
218 case connectivity.TransientFailure:
219 // Save error to be reported via picker.
220 b.connErr = state.ConnectionError
221 }
222 223 b.state = b.csEvltr.RecordTransition(oldS, s)
224 225 // Regenerate picker when one of the following happens:
226 // - this sc entered or left ready
227 // - the aggregated state of balancer is TransientFailure
228 // (may need to update error message)
229 if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
230 b.state == connectivity.TransientFailure {
231 b.regeneratePicker()
232 }
233 b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
234 }
235 236 // Close is a nop because base balancer doesn't have internal state to clean up,
237 // and it doesn't need to call Shutdown for the SubConns.
238 func (b *baseBalancer) Close() {
239 }
240 241 // ExitIdle is a nop because the base balancer attempts to stay connected to
242 // all SubConns at all times.
243 func (b *baseBalancer) ExitIdle() {
244 }
245 246 // NewErrPicker returns a Picker that always returns err on Pick().
247 func NewErrPicker(err error) balancer.Picker {
248 return &errPicker{err: err}
249 }
250 251 // NewErrPickerV2 is temporarily defined for backward compatibility reasons.
252 //
253 // Deprecated: use NewErrPicker instead.
254 var NewErrPickerV2 = NewErrPicker
255 256 type errPicker struct {
257 err error // Pick() always returns this err.
258 }
259 260 func (p *errPicker) Pick(balancer.PickInfo) (balancer.PickResult, error) {
261 return balancer.PickResult{}, p.err
262 }
263