threshold monitoring for beacon processes

chain/beacon/node.go

@@ -8,6 +8,8 @@ import (
 	"sync"
 	"time"
 
+	dmetrics "github.com/drand/drand/metrics"
+
 	clock "github.com/jonboulle/clockwork"
 
 	"github.com/drand/drand/chain"
@@ -45,8 +47,9 @@ type Handler struct {
 	// keeps the cryptographic info (group share etc)
 	crypto *vault.Vault
 	// main logic that treats incoming packet / new beacons created
-	chain  *chainStore
-	ticker *ticker
+	chain            *chainStore
+	ticker           *ticker
+	thresholdMonitor *dmetrics.ThresholdMonitor
 
 	ctx       context.Context
 	ctxCancel context.CancelFunc
@@ -87,16 +90,17 @@ func NewHandler(c net.ProtocolClient, s chain.Store, conf *Config, l log.Logger,
 	ctx, ctxCancel := context.WithCancel(context.Background())
 
 	handler := &Handler{
-		conf:      conf,
-		client:    c,
-		crypto:    v,
-		chain:     store,
-		ticker:    ticker,
-		addr:      addr,
-		ctx:       ctx,
-		ctxCancel: ctxCancel,
-		l:         l,
-		version:   version,
+		conf:             conf,
+		client:           c,
+		crypto:           v,
+		chain:            store,
+		ticker:           ticker,
+		addr:             addr,
+		ctx:              ctx,
+		ctxCancel:        ctxCancel,
+		l:                l,
+		version:          version,
+		thresholdMonitor: dmetrics.NewThresholdMonitor(conf.Group.ID, l, conf.Group.Threshold),
 	}
 	return handler, nil
 }
@@ -200,6 +204,7 @@ func (h *Handler) Start() error {
 	h.started = true
 	h.Unlock()
 
+	h.thresholdMonitor.Start()
 	_, tTime := chain.NextRound(h.conf.Clock.Now().Unix(), h.conf.Group.Period, h.conf.Group.GenesisTime)
 	h.l.Infow("", "beacon", "start", "scheme", h.crypto.Name)
 	go h.run(tTime)
@@ -218,6 +223,7 @@ func (h *Handler) Catchup() {
 	h.Unlock()
 
 	nRound, tTime := chain.NextRound(h.conf.Clock.Now().Unix(), h.conf.Group.Period, h.conf.Group.GenesisTime)
+	h.thresholdMonitor.Start()
 	go h.run(tTime)
 	h.chain.RunSync(nRound, nil)
 }
@@ -270,6 +276,7 @@ func (h *Handler) TransitionNewGroup(newShare *key.Share, newGroup *key.Group) {
 			return
 		}
 		h.crypto.SetInfo(newGroup, newShare)
+		h.thresholdMonitor.UpdateThreshold(newGroup.Threshold)
 		h.chain.RemoveCallback("transition")
 	})
 }
@@ -457,6 +464,7 @@ func (h *Handler) broadcastNextPartial(ctx context.Context, current roundInfo, u
 			h.l.Debugw("sending partial", "round", round, "to", i.Address())
 			err := h.client.PartialBeacon(ctx, &i, packet)
 			if err != nil {
+				h.thresholdMonitor.ReportFailure(i.Address())
 				h.l.Errorw("error sending partial", "round", round, "err", err, "to", i.Address())
 				return
 			}
@@ -476,6 +484,7 @@ func (h *Handler) Stop() {
 
 	h.ticker.Stop()
 	h.chain.Stop()
+	h.thresholdMonitor.Stop()
 
 	h.stopped = true
 	h.running = false

go.mod

@@ -141,6 +141,7 @@ require (
 	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/stretchr/objx v0.5.0 // indirect
 	github.com/uber/jaeger-client-go v2.30.0+incompatible // indirect
 	github.com/uber/jaeger-lib v2.4.1+incompatible // indirect
 	github.com/weaveworks/promrus v1.2.0 // indirect

metrics/threshold_monitor.go

@@ -0,0 +1,104 @@
+package metrics
+
+import (
+	"context"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/drand/drand/log"
+)
+
+type ThresholdMonitor struct {
+	lock              sync.RWMutex
+	log               log.Logger
+	beaconID          string
+	threshold         int
+	failedConnections map[string]bool
+	ctx               context.Context
+	cancel            func()
+	period            time.Duration
+}
+
+func NewThresholdMonitor(beaconID string, l log.Logger, threshold int) *ThresholdMonitor {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &ThresholdMonitor{
+		lock:              sync.RWMutex{},
+		log:               l,
+		beaconID:          beaconID,
+		threshold:         threshold,
+		failedConnections: make(map[string]bool),
+		ctx:               ctx,
+		cancel:            cancel,
+		period:            1 * time.Minute,
+	}
+}
+
+func (t *ThresholdMonitor) Start() {
+	t.log.Infow("starting threshold monitor", "beaconID", t.beaconID)
+
+	go func() {
+		for {
+			select {
+			case <-t.ctx.Done():
+				t.log.Infow("ending threshold monitor", "beaconID", t.beaconID)
+				return
+			default:
+				t.lock.RLock()
+				var failingNodes []string
+				for address := range t.failedConnections {
+					failingNodes = append(failingNodes, address)
+				}
+
+				if len(failingNodes) >= t.threshold {
+					t.log.Errorw(
+						"failed connections crossed threshold in the last minute",
+						"beaconID", t.beaconID,
+						"threshold", t.threshold,
+						"failures", len(failingNodes),
+						"nodes", strings.Join(failingNodes, ","),
+					)
+				} else if len(failingNodes) >= t.threshold/2 {
+					t.log.Warnw(
+						"failed connections crossed half threshold in the last minute",
+						"beaconID", t.beaconID,
+						"threshold", t.threshold,
+						"failures", len(failingNodes),
+						"nodes", strings.Join(failingNodes, ","),
+					)
+				} else {
+					t.log.Debugw(
+						"threshold monitor healthy",
+						"threshold", t.threshold,
+						"beaconID", t.beaconID,
+						"failures", len(failingNodes),
+						"nodes", strings.Join(failingNodes, ","),
+					)
+				}
+				t.lock.RUnlock()
+
+				t.lock.Lock()
+				t.failedConnections = make(map[string]bool)
+				t.lock.Unlock()
+
+				time.Sleep(t.period)
+			}
+		}
+	}()
+}
+
+func (t *ThresholdMonitor) Stop() {
+	t.cancel()
+}
+
+func (t *ThresholdMonitor) ReportFailure(addr string) {
+	t.lock.Lock()
+	t.failedConnections[addr] = true
+	t.lock.Unlock()
+}
+
+func (t *ThresholdMonitor) UpdateThreshold(newThreshold int) {
+	t.lock.Lock()
+	t.threshold = newThreshold
+	t.lock.Unlock()
+}