Periodically refresh ring information
Reviewers: sivakk, samar
@@ -2,6 +2,9 @@ package membership
import (
"sync"
+ "time"
+
+ "code.uber.internal/devexp/minions/common"
"github.com/dgryski/go-farm"
"github.com/uber-common/bark"
@@ -11,18 +14,27 @@ import (
"github.com/uber/ringpop-go/swim"
)
+const (
// RoleKey label is set by every single service as soon as it bootstraps its
// ringpop instance. The data for this key is the service name
-const RoleKey = "serviceName"
+ RoleKey = "serviceName"
+ defaultRefreshInterval = time.Second * 20
+)
type ringpopServiceResolver struct {
service string
+ isStarted bool
+ isStopped bool
rp *ringpop.Ringpop
- ring *hashring.HashRing
+ shutdownCh chan struct{}
+ shutdownWG sync.WaitGroup
+ logger bark.Logger
+
ringLock sync.RWMutex
- listeners map[string]chan<- *ChangedEvent
+ ring *hashring.HashRing
+
listenerLock sync.RWMutex
- logger bark.Logger
+ listeners map[string]chan<- *ChangedEvent
}
var _ ServiceResolver = (*ringpopServiceResolver)(nil)
@@ -34,6 +46,7 @@ func newRingpopServiceResolver(service string, rp *ringpop.Ringpop, logger bark.
logger: logger.WithFields(bark.Fields{"component": "ServiceResolver", RoleKey: service}),
ring: hashring.New(farm.Fingerprint32, 1),
listeners: make(map[string]chan<- *ChangedEvent),
+ shutdownCh: make(chan struct{}),
}
}
@@ -42,6 +55,10 @@ func (r *ringpopServiceResolver) Start() error {
r.ringLock.Lock()
defer r.ringLock.Unlock()
+ if r.isStarted {
+ return nil
+ }
+
r.rp.AddListener(r)
addrs, err := r.rp.GetReachableMembers(swim.MemberWithLabelAndValue(RoleKey, r.service))
if err != nil {
@@ -53,19 +70,34 @@ func (r *ringpopServiceResolver) Start() error {
r.ring.AddMembers(NewHostInfo(addr, labels))
}
+ r.shutdownWG.Add(1)
+ go r.refreshRingWorker()
+
+ r.isStarted = true
return nil
}
// Stop stops the oracle
func (r *ringpopServiceResolver) Stop() error {
r.ringLock.Lock()
+ defer r.ringLock.Unlock()
r.listenerLock.Lock()
defer r.listenerLock.Unlock()
- defer r.ringLock.Unlock()
+ if r.isStopped {
+ return nil
+ }
+
+ if r.isStarted {
r.rp.RemoveListener(r)
r.ring = hashring.New(farm.Fingerprint32, 1)
r.listeners = make(map[string]chan<- *ChangedEvent)
+ close(r.shutdownCh)
+ }
+
+ if success := common.AwaitWaitGroup(&r.shutdownWG, time.Minute); !success {
+ r.logger.Warn("service resolver timed out on shutdown.")
+ }
return nil
}
@@ -162,6 +194,22 @@ func (r *ringpopServiceResolver) emitEvent(rpEvent events.RingChangedEvent) {
}
}
+func (r *ringpopServiceResolver) refreshRingWorker() {
+ defer r.shutdownWG.Done()
+
+ refreshTicker := time.NewTicker(defaultRefreshInterval)
+ defer refreshTicker.Stop()
+
+ for {
+ select {
+ case <-r.shutdownCh:
+ return
+ case <-refreshTicker.C:
+ r.refresh()
+ }
+ }
+}
+
func (r *ringpopServiceResolver) getLabelsMap() map[string]string {
labels := make(map[string]string)
labels[RoleKey] = r.service
