| /* |
| * |
| * Copyright 2017 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 base |
| |
| import ( |
| "errors" |
| "fmt" |
| |
| "google.golang.org/grpc/balancer" |
| "google.golang.org/grpc/connectivity" |
| "google.golang.org/grpc/grpclog" |
| "google.golang.org/grpc/resolver" |
| ) |
| |
| var logger = grpclog.Component("balancer") |
| |
| type baseBuilder struct { |
| name string |
| pickerBuilder PickerBuilder |
| config Config |
| } |
| |
| func (bb *baseBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer { |
| bal := &baseBalancer{ |
| cc: cc, |
| pickerBuilder: bb.pickerBuilder, |
| |
| subConns: make(map[resolver.Address]balancer.SubConn), |
| scStates: make(map[balancer.SubConn]connectivity.State), |
| csEvltr: &balancer.ConnectivityStateEvaluator{}, |
| config: bb.config, |
| } |
| // Initialize picker to a picker that always returns |
| // ErrNoSubConnAvailable, because when state of a SubConn changes, we |
| // may call UpdateState with this picker. |
| bal.picker = NewErrPicker(balancer.ErrNoSubConnAvailable) |
| return bal |
| } |
| |
| func (bb *baseBuilder) Name() string { |
| return bb.name |
| } |
| |
| type baseBalancer struct { |
| cc balancer.ClientConn |
| pickerBuilder PickerBuilder |
| |
| csEvltr *balancer.ConnectivityStateEvaluator |
| state connectivity.State |
| |
| subConns map[resolver.Address]balancer.SubConn |
| scStates map[balancer.SubConn]connectivity.State |
| picker balancer.Picker |
| config Config |
| |
| resolverErr error // the last error reported by the resolver; cleared on successful resolution |
| connErr error // the last connection error; cleared upon leaving TransientFailure |
| } |
| |
| func (b *baseBalancer) ResolverError(err error) { |
| b.resolverErr = err |
| if len(b.subConns) == 0 { |
| b.state = connectivity.TransientFailure |
| } |
| |
| if b.state != connectivity.TransientFailure { |
| // The picker will not change since the balancer does not currently |
| // report an error. |
| return |
| } |
| b.regeneratePicker() |
| b.cc.UpdateState(balancer.State{ |
| ConnectivityState: b.state, |
| Picker: b.picker, |
| }) |
| } |
| |
| func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error { |
| // TODO: handle s.ResolverState.ServiceConfig? |
| if logger.V(2) { |
| logger.Info("base.baseBalancer: got new ClientConn state: ", s) |
| } |
| // Successful resolution; clear resolver error and ensure we return nil. |
| b.resolverErr = nil |
| // addrsSet is the set converted from addrs, it's used for quick lookup of an address. |
| addrsSet := make(map[resolver.Address]struct{}) |
| for _, a := range s.ResolverState.Addresses { |
| addrsSet[a] = struct{}{} |
| if _, ok := b.subConns[a]; !ok { |
| // a is a new address (not existing in b.subConns). |
| sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck}) |
| if err != nil { |
| logger.Warningf("base.baseBalancer: failed to create new SubConn: %v", err) |
| continue |
| } |
| b.subConns[a] = sc |
| b.scStates[sc] = connectivity.Idle |
| sc.Connect() |
| } |
| } |
| for a, sc := range b.subConns { |
| // a was removed by resolver. |
| if _, ok := addrsSet[a]; !ok { |
| b.cc.RemoveSubConn(sc) |
| delete(b.subConns, a) |
| // Keep the state of this sc in b.scStates until sc's state becomes Shutdown. |
| // The entry will be deleted in UpdateSubConnState. |
| } |
| } |
| // If resolver state contains no addresses, return an error so ClientConn |
| // will trigger re-resolve. Also records this as an resolver error, so when |
| // the overall state turns transient failure, the error message will have |
| // the zero address information. |
| if len(s.ResolverState.Addresses) == 0 { |
| b.ResolverError(errors.New("produced zero addresses")) |
| return balancer.ErrBadResolverState |
| } |
| return nil |
| } |
| |
| // mergeErrors builds an error from the last connection error and the last |
| // resolver error. Must only be called if b.state is TransientFailure. |
| func (b *baseBalancer) mergeErrors() error { |
| // connErr must always be non-nil unless there are no SubConns, in which |
| // case resolverErr must be non-nil. |
| if b.connErr == nil { |
| return fmt.Errorf("last resolver error: %v", b.resolverErr) |
| } |
| if b.resolverErr == nil { |
| return fmt.Errorf("last connection error: %v", b.connErr) |
| } |
| return fmt.Errorf("last connection error: %v; last resolver error: %v", b.connErr, b.resolverErr) |
| } |
| |
| // regeneratePicker takes a snapshot of the balancer, and generates a picker |
| // from it. The picker is |
| // - errPicker if the balancer is in TransientFailure, |
| // - built by the pickerBuilder with all READY SubConns otherwise. |
| func (b *baseBalancer) regeneratePicker() { |
| if b.state == connectivity.TransientFailure { |
| b.picker = NewErrPicker(b.mergeErrors()) |
| return |
| } |
| readySCs := make(map[balancer.SubConn]SubConnInfo) |
| |
| // Filter out all ready SCs from full subConn map. |
| for addr, sc := range b.subConns { |
| if st, ok := b.scStates[sc]; ok && st == connectivity.Ready { |
| readySCs[sc] = SubConnInfo{Address: addr} |
| } |
| } |
| b.picker = b.pickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs}) |
| } |
| |
| func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) { |
| s := state.ConnectivityState |
| if logger.V(2) { |
| logger.Infof("base.baseBalancer: handle SubConn state change: %p, %v", sc, s) |
| } |
| oldS, ok := b.scStates[sc] |
| if !ok { |
| if logger.V(2) { |
| logger.Infof("base.baseBalancer: got state changes for an unknown SubConn: %p, %v", sc, s) |
| } |
| return |
| } |
| if oldS == connectivity.TransientFailure && s == connectivity.Connecting { |
| // Once a subconn enters TRANSIENT_FAILURE, ignore subsequent |
| // CONNECTING transitions to prevent the aggregated state from being |
| // always CONNECTING when many backends exist but are all down. |
| return |
| } |
| b.scStates[sc] = s |
| switch s { |
| case connectivity.Idle: |
| sc.Connect() |
| case connectivity.Shutdown: |
| // When an address was removed by resolver, b called RemoveSubConn but |
| // kept the sc's state in scStates. Remove state for this sc here. |
| delete(b.scStates, sc) |
| case connectivity.TransientFailure: |
| // Save error to be reported via picker. |
| b.connErr = state.ConnectionError |
| } |
| |
| b.state = b.csEvltr.RecordTransition(oldS, s) |
| |
| // Regenerate picker when one of the following happens: |
| // - this sc entered or left ready |
| // - the aggregated state of balancer is TransientFailure |
| // (may need to update error message) |
| if (s == connectivity.Ready) != (oldS == connectivity.Ready) || |
| b.state == connectivity.TransientFailure { |
| b.regeneratePicker() |
| } |
| |
| b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker}) |
| } |
| |
| // Close is a nop because base balancer doesn't have internal state to clean up, |
| // and it doesn't need to call RemoveSubConn for the SubConns. |
| func (b *baseBalancer) Close() { |
| } |
| |
| // NewErrPicker returns a Picker that always returns err on Pick(). |
| func NewErrPicker(err error) balancer.Picker { |
| return &errPicker{err: err} |
| } |
| |
| // NewErrPickerV2 is temporarily defined for backward compatibility reasons. |
| // |
| // Deprecated: use NewErrPicker instead. |
| var NewErrPickerV2 = NewErrPicker |
| |
| type errPicker struct { |
| err error // Pick() always returns this err. |
| } |
| |
| func (p *errPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) { |
| return balancer.PickResult{}, p.err |
| } |