mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
ea72396b4d
* Remove Node from data model; standardize on Worker This commit does a lot of things, but in general it attempts to simplify the data model by getting rid of the Node and NodeRole models. Instead, these will use the Worker model, which itself has individual boolean fields for role types. Get rid of roleType in some FreeWorker methods rename NodeService to WorkerRegistry simplify the freeworker interface fix the tests * Remove DeleteWorker method from workerJobService
130 lines
2.6 KiB
Go
130 lines
2.6 KiB
Go
// Package poller provides the core Poller struct.
|
|
package poller
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/featurebasedb/featurebase/v3/dax"
|
|
"github.com/featurebasedb/featurebase/v3/logger"
|
|
)
|
|
|
|
// Poller maintains a list of nodes to poll. It also polls them.
|
|
type Poller struct {
|
|
mu sync.RWMutex
|
|
|
|
addressManager dax.AddressManager
|
|
|
|
workerRegistry dax.WorkerRegistry
|
|
|
|
nodePoller NodePoller
|
|
pollInterval time.Duration
|
|
|
|
stopping chan struct{}
|
|
|
|
logger logger.Logger
|
|
}
|
|
|
|
// New returns a new instance of Poller with default values.
|
|
func New(cfg Config) *Poller {
|
|
p := &Poller{
|
|
addressManager: dax.NewNopAddressManager(),
|
|
workerRegistry: dax.NewNopWorkerRegistry(),
|
|
nodePoller: NewNopNodePoller(),
|
|
pollInterval: time.Second,
|
|
logger: logger.NopLogger,
|
|
}
|
|
|
|
// Set config options.
|
|
if cfg.AddressManager != nil {
|
|
p.addressManager = cfg.AddressManager
|
|
}
|
|
if cfg.WorkerRegistry != nil {
|
|
p.workerRegistry = cfg.WorkerRegistry
|
|
}
|
|
if cfg.NodePoller != nil {
|
|
p.nodePoller = cfg.NodePoller
|
|
}
|
|
if cfg.PollInterval != 0 {
|
|
p.pollInterval = cfg.PollInterval
|
|
}
|
|
if cfg.Logger != nil {
|
|
p.logger = cfg.Logger
|
|
}
|
|
|
|
return p
|
|
}
|
|
|
|
func (p *Poller) Addresses() []dax.Address {
|
|
nodes, err := p.workerRegistry.Workers(context.Background())
|
|
if err != nil {
|
|
p.logger.Errorf("POLLER: unable to get nodes from node service: %v", err)
|
|
}
|
|
|
|
addrs := make([]dax.Address, 0, len(nodes))
|
|
for _, node := range nodes {
|
|
addrs = append(addrs, node.Address)
|
|
}
|
|
|
|
return addrs
|
|
}
|
|
|
|
// Run starts the polling goroutine.
|
|
func (p *Poller) Run() error {
|
|
// Set up the stopping channel here in case the controller restarts and runs
|
|
// the Poller again.
|
|
p.stopping = make(chan struct{})
|
|
|
|
p.run()
|
|
return nil
|
|
}
|
|
|
|
func (p *Poller) run() {
|
|
ticker := time.NewTicker(p.pollInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
// Wait for tick or a close.
|
|
select {
|
|
case <-p.stopping:
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
|
|
p.pollAll()
|
|
}
|
|
|
|
}
|
|
|
|
// Stop stops the polling routine.
|
|
func (p *Poller) Stop() {
|
|
close(p.stopping)
|
|
}
|
|
|
|
func (p *Poller) pollAll() {
|
|
addrs := p.Addresses()
|
|
|
|
ctx := context.Background()
|
|
|
|
toRemove := []dax.Address{}
|
|
|
|
for _, addr := range addrs {
|
|
up := p.nodePoller.Poll(addr)
|
|
if !up {
|
|
p.logger.Printf("poller removing %s", addr)
|
|
toRemove = append(toRemove, addr)
|
|
}
|
|
}
|
|
|
|
if len(toRemove) > 0 {
|
|
p.logger.Debugf("POLLER: removing addresses: %v", toRemove)
|
|
start := time.Now()
|
|
err := p.addressManager.RemoveAddresses(ctx, toRemove...)
|
|
if err != nil {
|
|
p.logger.Printf("POLLER: error removing %s: %v", toRemove, err)
|
|
}
|
|
p.logger.Debugf("POLLER removing %v complete: %s", toRemove, time.Since(start))
|
|
}
|
|
|
|
}
|