diff --git a/signal/server/send_tracker.go b/signal/server/send_tracker.go new file mode 100644 index 000000000..3721b946a --- /dev/null +++ b/signal/server/send_tracker.go @@ -0,0 +1,77 @@ +package server + +import ( + "context" + "sync" + "time" + + log "github.com/sirupsen/logrus" +) + +const sendRateLogInterval = 5 * time.Minute + +// sendRateTracker tracks per-key message counts and logs the busiest peers periodically. +type sendRateTracker struct { + mu sync.Mutex + counts map[string]int64 +} + +func newSendRateTracker() *sendRateTracker { + return &sendRateTracker{counts: make(map[string]int64)} +} + +func (t *sendRateTracker) increment(key string) { + t.mu.Lock() + t.counts[key]++ + t.mu.Unlock() +} + +// resetAndSnapshot atomically returns current counts and resets the tracker. +func (t *sendRateTracker) resetAndSnapshot() map[string]int64 { + t.mu.Lock() + snap := t.counts + t.counts = make(map[string]int64, len(snap)) + t.mu.Unlock() + return snap +} + +// logSendRates periodically logs peers that have at least half the rate of the busiest peer. +func (t *sendRateTracker) logSendRates(ctx context.Context) { + ticker := time.NewTicker(sendRateLogInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + snap := t.resetAndSnapshot() + if len(snap) == 0 { + continue + } + + var maxCount int64 + for _, count := range snap { + if count > maxCount { + maxCount = count + } + } + + threshold := int64(float64(maxCount) * 0.95) + intervalMin := sendRateLogInterval.Minutes() + + log.Debugf("send rate stats: %d unique peers in last %.0fs, max rate %.1f msg/min", + len(snap), sendRateLogInterval.Seconds(), float64(maxCount)/intervalMin) + logged := 0 + for key, count := range snap { + if count >= threshold { + log.Debugf("peer [%s] %.1f msg/min", key, float64(count)/intervalMin) + logged++ + if logged >= 100 { + break + } + } + } + } + } +} diff --git a/signal/server/send_tracker_test.go b/signal/server/send_tracker_test.go new file mode 100644 index 000000000..12f3df5f2 --- /dev/null +++ b/signal/server/send_tracker_test.go @@ -0,0 +1,56 @@ +package server + +import ( + "sync" + "testing" +) + +func TestSendRateTracker_Increment(t *testing.T) { + tracker := newSendRateTracker() + + tracker.increment("peer-a") + tracker.increment("peer-a") + tracker.increment("peer-b") + + snap := tracker.resetAndSnapshot() + if snap["peer-a"] != 2 { + t.Errorf("expected peer-a count 2, got %d", snap["peer-a"]) + } + if snap["peer-b"] != 1 { + t.Errorf("expected peer-b count 1, got %d", snap["peer-b"]) + } +} + +func TestSendRateTracker_ResetAndSnapshot_Resets(t *testing.T) { + tracker := newSendRateTracker() + tracker.increment("peer-a") + + snap1 := tracker.resetAndSnapshot() + if snap1["peer-a"] != 1 { + t.Fatalf("expected 1, got %d", snap1["peer-a"]) + } + + snap2 := tracker.resetAndSnapshot() + if len(snap2) != 0 { + t.Errorf("expected empty snapshot after reset, got %v", snap2) + } +} + +func TestSendRateTracker_ConcurrentIncrement(t *testing.T) { + tracker := newSendRateTracker() + + var wg sync.WaitGroup + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + tracker.increment("peer-x") + }() + } + wg.Wait() + + snap := tracker.resetAndSnapshot() + if snap["peer-x"] != 100 { + t.Errorf("expected 100, got %d", snap["peer-x"]) + } +} diff --git a/signal/server/signal.go b/signal/server/signal.go index c46df56d2..6fc980c1f 100644 --- a/signal/server/signal.go +++ b/signal/server/signal.go @@ -59,6 +59,8 @@ type Server struct { successHeader metadata.MD sendTimeout time.Duration + + sendTracker *sendRateTracker } // NewServer creates a new Signal server @@ -86,8 +88,11 @@ func NewServer(ctx context.Context, meter metric.Meter, metricsPrefix ...string) metrics: appMetrics, successHeader: metadata.Pairs(proto.HeaderRegistered, "1"), sendTimeout: sTimeout, + sendTracker: newSendRateTracker(), } + go s.sendTracker.logSendRates(ctx) + return s, nil } @@ -95,6 +100,8 @@ func NewServer(ctx context.Context, meter metric.Meter, metricsPrefix ...string) func (s *Server) Send(ctx context.Context, msg *proto.EncryptedMessage) (*proto.EncryptedMessage, error) { log.Tracef("received a new message to send from peer [%s] to peer [%s]", msg.Key, msg.RemoteKey) + s.sendTracker.increment(msg.Key) + if _, found := s.registry.Get(msg.RemoteKey); found { s.forwardMessageToPeer(ctx, msg) return &proto.EncryptedMessage{}, nil