blob: 042c949b70806bbcde16f203b39c18b00be7daf2 [file] [log] [blame]
// Copyright 2017 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 grpcproxy
import (
"context"
"math"
"sync"
"github.com/coreos/etcd/clientv3"
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
"golang.org/x/time/rate"
"google.golang.org/grpc"
)
const (
lostLeaderKey = "__lostleader" // watched to detect leader loss
retryPerSecond = 10
)
type leader struct {
ctx context.Context
w clientv3.Watcher
// mu protects leaderc updates.
mu sync.RWMutex
leaderc chan struct{}
disconnc chan struct{}
donec chan struct{}
}
func newLeader(ctx context.Context, w clientv3.Watcher) *leader {
l := &leader{
ctx: clientv3.WithRequireLeader(ctx),
w: w,
leaderc: make(chan struct{}),
disconnc: make(chan struct{}),
donec: make(chan struct{}),
}
// begin assuming leader is lost
close(l.leaderc)
go l.recvLoop()
return l
}
func (l *leader) recvLoop() {
defer close(l.donec)
limiter := rate.NewLimiter(rate.Limit(retryPerSecond), retryPerSecond)
rev := int64(math.MaxInt64 - 2)
for limiter.Wait(l.ctx) == nil {
wch := l.w.Watch(l.ctx, lostLeaderKey, clientv3.WithRev(rev), clientv3.WithCreatedNotify())
cresp, ok := <-wch
if !ok {
l.loseLeader()
continue
}
if cresp.Err() != nil {
l.loseLeader()
if rpctypes.ErrorDesc(cresp.Err()) == grpc.ErrClientConnClosing.Error() {
close(l.disconnc)
return
}
continue
}
l.gotLeader()
<-wch
l.loseLeader()
}
}
func (l *leader) loseLeader() {
l.mu.RLock()
defer l.mu.RUnlock()
select {
case <-l.leaderc:
default:
close(l.leaderc)
}
}
// gotLeader will force update the leadership status to having a leader.
func (l *leader) gotLeader() {
l.mu.Lock()
defer l.mu.Unlock()
select {
case <-l.leaderc:
l.leaderc = make(chan struct{})
default:
}
}
func (l *leader) disconnectNotify() <-chan struct{} { return l.disconnc }
func (l *leader) stopNotify() <-chan struct{} { return l.donec }
// lostNotify returns a channel that is closed if there has been
// a leader loss not yet followed by a leader reacquire.
func (l *leader) lostNotify() <-chan struct{} {
l.mu.RLock()
defer l.mu.RUnlock()
return l.leaderc
}