// Copyright 2018 The etcd 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 balancer implements client balancer. package balancer import ( "strconv" "sync" "time" "go.etcd.io/etcd/clientv3/balancer/connectivity" "go.etcd.io/etcd/clientv3/balancer/picker" "go.uber.org/zap" "google.golang.org/grpc/balancer" grpcconnectivity "google.golang.org/grpc/connectivity" "google.golang.org/grpc/resolver" _ "google.golang.org/grpc/resolver/dns" // register DNS resolver _ "google.golang.org/grpc/resolver/passthrough" // register passthrough resolver ) // Config defines balancer configurations. type Config struct { // Policy configures balancer policy. Policy picker.Policy // Picker implements gRPC picker. // Leave empty if "Policy" field is not custom. // TODO: currently custom policy is not supported. // Picker picker.Picker // Name defines an additional name for balancer. // Useful for balancer testing to avoid register conflicts. // If empty, defaults to policy name. Name string // Logger configures balancer logging. // If nil, logs are discarded. Logger *zap.Logger } // RegisterBuilder creates and registers a builder. Since this function calls balancer.Register, it // must be invoked at initialization time. func RegisterBuilder(cfg Config) { bb := &builder{cfg} balancer.Register(bb) bb.cfg.Logger.Debug( "registered balancer", zap.String("policy", bb.cfg.Policy.String()), zap.String("name", bb.cfg.Name), ) } type builder struct { cfg Config } // Build is called initially when creating "ccBalancerWrapper". // "grpc.Dial" is called to this client connection. // Then, resolved addresses will be handled via "HandleResolvedAddrs". func (b *builder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer { bb := &baseBalancer{ id: strconv.FormatInt(time.Now().UnixNano(), 36), policy: b.cfg.Policy, name: b.cfg.Name, lg: b.cfg.Logger, addrToSc: make(map[resolver.Address]balancer.SubConn), scToAddr: make(map[balancer.SubConn]resolver.Address), scToSt: make(map[balancer.SubConn]grpcconnectivity.State), currentConn: nil, connectivityRecorder: connectivity.New(b.cfg.Logger), // initialize picker always returns "ErrNoSubConnAvailable" picker: picker.NewErr(balancer.ErrNoSubConnAvailable), } // TODO: support multiple connections bb.mu.Lock() bb.currentConn = cc bb.mu.Unlock() bb.lg.Info( "built balancer", zap.String("balancer-id", bb.id), zap.String("policy", bb.policy.String()), zap.String("resolver-target", cc.Target()), ) return bb } // Name implements "grpc/balancer.Builder" interface. func (b *builder) Name() string { return b.cfg.Name } // Balancer defines client balancer interface. type Balancer interface { // Balancer is called on specified client connection. Client initiates gRPC // connection with "grpc.Dial(addr, grpc.WithBalancerName)", and then those resolved // addresses are passed to "grpc/balancer.Balancer.HandleResolvedAddrs". // For each resolved address, balancer calls "balancer.ClientConn.NewSubConn". // "grpc/balancer.Balancer.HandleSubConnStateChange" is called when connectivity state // changes, thus requires failover logic in this method. balancer.Balancer // Picker calls "Pick" for every client request. picker.Picker } type baseBalancer struct { id string policy picker.Policy name string lg *zap.Logger mu sync.RWMutex addrToSc map[resolver.Address]balancer.SubConn scToAddr map[balancer.SubConn]resolver.Address scToSt map[balancer.SubConn]grpcconnectivity.State currentConn balancer.ClientConn connectivityRecorder connectivity.Recorder picker picker.Picker } // HandleResolvedAddrs implements "grpc/balancer.Balancer" interface. // gRPC sends initial or updated resolved addresses from "Build". func (bb *baseBalancer) HandleResolvedAddrs(addrs []resolver.Address, err error) { if err != nil { bb.lg.Warn("HandleResolvedAddrs called with error", zap.String("balancer-id", bb.id), zap.Error(err)) return } bb.lg.Info("resolved", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.Strings("addresses", addrsToStrings(addrs)), ) bb.mu.Lock() defer bb.mu.Unlock() resolved := make(map[resolver.Address]struct{}) for _, addr := range addrs { resolved[addr] = struct{}{} if _, ok := bb.addrToSc[addr]; !ok { sc, err := bb.currentConn.NewSubConn([]resolver.Address{addr}, balancer.NewSubConnOptions{}) if err != nil { bb.lg.Warn("NewSubConn failed", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.Error(err), zap.String("address", addr.Addr)) continue } bb.lg.Info("created subconn", zap.String("address", addr.Addr)) bb.addrToSc[addr] = sc bb.scToAddr[sc] = addr bb.scToSt[sc] = grpcconnectivity.Idle sc.Connect() } } for addr, sc := range bb.addrToSc { if _, ok := resolved[addr]; !ok { // was removed by resolver or failed to create subconn bb.currentConn.RemoveSubConn(sc) delete(bb.addrToSc, addr) bb.lg.Info( "removed subconn", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.String("address", addr.Addr), zap.String("subconn", scToString(sc)), ) // Keep the state of this sc in bb.scToSt until sc's state becomes Shutdown. // The entry will be deleted in HandleSubConnStateChange. // (DO NOT) delete(bb.scToAddr, sc) // (DO NOT) delete(bb.scToSt, sc) } } } // HandleSubConnStateChange implements "grpc/balancer.Balancer" interface. func (bb *baseBalancer) HandleSubConnStateChange(sc balancer.SubConn, s grpcconnectivity.State) { bb.mu.Lock() defer bb.mu.Unlock() old, ok := bb.scToSt[sc] if !ok { bb.lg.Warn( "state change for an unknown subconn", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.String("subconn", scToString(sc)), zap.Int("subconn-size", len(bb.scToAddr)), zap.String("state", s.String()), ) return } bb.lg.Info( "state changed", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.Bool("connected", s == grpcconnectivity.Ready), zap.String("subconn", scToString(sc)), zap.Int("subconn-size", len(bb.scToAddr)), zap.String("address", bb.scToAddr[sc].Addr), zap.String("old-state", old.String()), zap.String("new-state", s.String()), ) bb.scToSt[sc] = s switch s { case grpcconnectivity.Idle: sc.Connect() case grpcconnectivity.Shutdown: // When an address was removed by resolver, b called RemoveSubConn but // kept the sc's state in scToSt. Remove state for this sc here. delete(bb.scToAddr, sc) delete(bb.scToSt, sc) } oldAggrState := bb.connectivityRecorder.GetCurrentState() bb.connectivityRecorder.RecordTransition(old, s) // Update balancer picker when one of the following happens: // - this sc became ready from not-ready // - this sc became not-ready from ready // - the aggregated state of balancer became TransientFailure from non-TransientFailure // - the aggregated state of balancer became non-TransientFailure from TransientFailure if (s == grpcconnectivity.Ready) != (old == grpcconnectivity.Ready) || (bb.connectivityRecorder.GetCurrentState() == grpcconnectivity.TransientFailure) != (oldAggrState == grpcconnectivity.TransientFailure) { bb.updatePicker() } bb.currentConn.UpdateBalancerState(bb.connectivityRecorder.GetCurrentState(), bb.picker) } func (bb *baseBalancer) updatePicker() { if bb.connectivityRecorder.GetCurrentState() == grpcconnectivity.TransientFailure { bb.picker = picker.NewErr(balancer.ErrTransientFailure) bb.lg.Info( "updated picker to transient error picker", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.String("policy", bb.policy.String()), ) return } // only pass ready subconns to picker scToAddr := make(map[balancer.SubConn]resolver.Address) for addr, sc := range bb.addrToSc { if st, ok := bb.scToSt[sc]; ok && st == grpcconnectivity.Ready { scToAddr[sc] = addr } } bb.picker = picker.New(picker.Config{ Policy: bb.policy, Logger: bb.lg, SubConnToResolverAddress: scToAddr, }) bb.lg.Info( "updated picker", zap.String("picker", bb.picker.String()), zap.String("balancer-id", bb.id), zap.String("policy", bb.policy.String()), zap.Strings("subconn-ready", scsToStrings(scToAddr)), zap.Int("subconn-size", len(scToAddr)), ) } // Close implements "grpc/balancer.Balancer" interface. // 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 (bb *baseBalancer) Close() { // TODO }