vendor/google.golang.org/grpc/balancer/base/balancer.go

   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         "context"
  23
  24         "google.golang.org/grpc/balancer"
  25         "google.golang.org/grpc/connectivity"
  26         "google.golang.org/grpc/grpclog"
  27         "google.golang.org/grpc/resolver"
  28 )
  29
  30 type baseBuilder struct {
  31         name          string
  32         pickerBuilder PickerBuilder
  33         config        Config
  34 }
  35
  36 func (bb *baseBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
  37         return &baseBalancer{
  38                 cc:            cc,
  39                 pickerBuilder: bb.pickerBuilder,
  40
  41                 subConns: make(map[resolver.Address]balancer.SubConn),
  42                 scStates: make(map[balancer.SubConn]connectivity.State),
  43                 csEvltr:  &balancer.ConnectivityStateEvaluator{},
  44                 // Initialize picker to a picker that always return
  45                 // ErrNoSubConnAvailable, because when state of a SubConn changes, we
  46                 // may call UpdateBalancerState with this picker.
  47                 picker: NewErrPicker(balancer.ErrNoSubConnAvailable),
  48                 config: bb.config,
  49         }
  50 }
  51
  52 func (bb *baseBuilder) Name() string {
  53         return bb.name
  54 }
  55
  56 type baseBalancer struct {
  57         cc            balancer.ClientConn
  58         pickerBuilder PickerBuilder
  59
  60         csEvltr *balancer.ConnectivityStateEvaluator
  61         state   connectivity.State
  62
  63         subConns map[resolver.Address]balancer.SubConn
  64         scStates map[balancer.SubConn]connectivity.State
  65         picker   balancer.Picker
  66         config   Config
  67 }
  68
  69 func (b *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) {
  70         if err != nil {
  71                 grpclog.Infof("base.baseBalancer: HandleResolvedAddrs called with error %v", err)
  72                 return
  73         }
  74         grpclog.Infoln("base.baseBalancer: got new resolved addresses: ", addrs)
  75         // addrsSet is the set converted from addrs, it's used for quick lookup of an address.
  76         addrsSet := make(map[resolver.Address]struct{})
  77         for _, a := range addrs {
  78                 addrsSet[a] = struct{}{}
  79                 if _, ok := b.subConns[a]; !ok {
  80                         // a is a new address (not existing in b.subConns).
  81                         sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
  82                         if err != nil {
  83                                 grpclog.Warningf("base.baseBalancer: failed to create new SubConn: %v", err)
  84                                 continue
  85                         }
  86                         b.subConns[a] = sc
  87                         b.scStates[sc] = connectivity.Idle
  88                         sc.Connect()
  89                 }
  90         }
  91         for a, sc := range b.subConns {
  92                 // a was removed by resolver.
  93                 if _, ok := addrsSet[a]; !ok {
  94                         b.cc.RemoveSubConn(sc)
  95                         delete(b.subConns, a)
  96                         // Keep the state of this sc in b.scStates until sc's state becomes Shutdown.
  97                         // The entry will be deleted in HandleSubConnStateChange.
  98                 }
  99         }
 100 }
 101
 102 // regeneratePicker takes a snapshot of the balancer, and generates a picker
 103 // from it. The picker is
 104 //  - errPicker with ErrTransientFailure if the balancer is in TransientFailure,
 105 //  - built by the pickerBuilder with all READY SubConns otherwise.
 106 func (b *baseBalancer) regeneratePicker() {
 107         if b.state == connectivity.TransientFailure {
 108                 b.picker = NewErrPicker(balancer.ErrTransientFailure)
 109                 return
 110         }
 111         readySCs := make(map[resolver.Address]balancer.SubConn)
 112
 113         // Filter out all ready SCs from full subConn map.
 114         for addr, sc := range b.subConns {
 115                 if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
 116                         readySCs[addr] = sc
 117                 }
 118         }
 119         b.picker = b.pickerBuilder.Build(readySCs)
 120 }
 121
 122 func (b *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
 123         grpclog.Infof("base.baseBalancer: handle SubConn state change: %p, %v", sc, s)
 124         oldS, ok := b.scStates[sc]
 125         if !ok {
 126                 grpclog.Infof("base.baseBalancer: got state changes for an unknown SubConn: %p, %v", sc, s)
 127                 return
 128         }
 129         b.scStates[sc] = s
 130         switch s {
 131         case connectivity.Idle:
 132                 sc.Connect()
 133         case connectivity.Shutdown:
 134                 // When an address was removed by resolver, b called RemoveSubConn but
 135                 // kept the sc's state in scStates. Remove state for this sc here.
 136                 delete(b.scStates, sc)
 137         }
 138
 139         oldAggrState := b.state
 140         b.state = b.csEvltr.RecordTransition(oldS, s)
 141
 142         // Regenerate picker when one of the following happens:
 143         //  - this sc became ready from not-ready
 144         //  - this sc became not-ready from ready
 145         //  - the aggregated state of balancer became TransientFailure from non-TransientFailure
 146         //  - the aggregated state of balancer became non-TransientFailure from TransientFailure
 147         if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
 148                 (b.state == connectivity.TransientFailure) != (oldAggrState == connectivity.TransientFailure) {
 149                 b.regeneratePicker()
 150         }
 151
 152         b.cc.UpdateBalancerState(b.state, b.picker)
 153 }
 154
 155 // Close is a nop because base balancer doesn't have internal state to clean up,
 156 // and it doesn't need to call RemoveSubConn for the SubConns.
 157 func (b *baseBalancer) Close() {
 158 }
 159
 160 // NewErrPicker returns a picker that always returns err on Pick().
 161 func NewErrPicker(err error) balancer.Picker {
 162         return &errPicker{err: err}
 163 }
 164
 165 type errPicker struct {
 166         err error // Pick() always returns this err.
 167 }
 168
 169 func (p *errPicker) Pick(ctx context.Context, opts balancer.PickOptions) (balancer.SubConn, func(balancer.DoneInfo), error) {
 170         return nil, nil, p.err
 171 }