Repository: nats-io/nats-server
Pull request #2994 (2022-04-01 21:34:42)

[CHANGED] Rate limit (some) similar warnings
Some warnings, especially when dealing with JS limits that were
printed on a per-message basis, are now limited to ~1 per second
if the content of the warning is already found in a map.

This is also for "client" warnings, but the client porting of the
warning is not taken into account so that helps with reducing logging
for similar content, but coming from different clients.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/server/client.go b/server/client.go
index 4e563909..bea7fea7 100644
--- a/server/client.go
+++ b/server/client.go
@@ -5325,6 +5325,15 @@ func (c *client) Warnf(format string, v ...interface{}) {
 	c.srv.Warnf(format, v...)
 }
 
+func (c *client) RateLimitWarnf(format string, v ...interface{}) {
+	// Do the check before adding the client info to the format...
+	statement := fmt.Sprintf(format, v...)
+	if _, loaded := c.srv.rateLimitLogging.LoadOrStore(statement, time.Now()); loaded {
+		return
+	}
+	c.Warnf("%s", statement)
+}
+
 // Set the very first PING to a lower interval to capture the initial RTT.
 // After that the PING interval will be set to the user defined value.
 // Client lock should be held.
diff --git a/server/jetstream.go b/server/jetstream.go
index 9a74cafc..a6cbbfca 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -2269,7 +2269,7 @@ func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client,
 	t.mu.Unlock()
 
 	if atLimit {
-		c.Warnf("JetStream could not create stream for account %q on subject %q, at limit", acc.Name, subject)
+		c.RateLimitWarnf("JetStream could not create stream for account %q on subject %q, at limit", acc.Name, subject)
 		return
 	}
 
@@ -2280,7 +2280,7 @@ func (t *streamTemplate) processInboundTemplateMsg(_ *subscription, pc *client,
 	mset, err := acc.addStream(&cfg)
 	if err != nil {
 		acc.validateStreams(t)
-		c.Warnf("JetStream could not create stream for account %q on subject %q", acc.Name, subject)
+		c.RateLimitWarnf("JetStream could not create stream for account %q on subject %q: %v", acc.Name, subject, err)
 		return
 	}
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index fd6f2367..494a8053 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5131,7 +5131,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	if !ok {
 		jsa.mu.RUnlock()
 		err := fmt.Errorf("no JetStream resource limits found account: %q", jsa.acc().Name)
-		s.Warnf(err.Error())
+		s.RateLimitWarnf(err.Error())
 		if canRespond {
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSNoLimitsError()
@@ -5161,7 +5161,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	// If we have exceeded our account limits go ahead and return.
 	if exceeded {
 		err := fmt.Errorf("JetStream resource limits exceeded for account: %q", jsa.acc().Name)
-		s.Warnf(err.Error())
+		s.RateLimitWarnf(err.Error())
 		if canRespond {
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSAccountResourcesExceededError()
@@ -5174,7 +5174,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	// Check msgSize if we have a limit set there. Again this works if it goes through but better to be pre-emptive.
 	if maxMsgSize >= 0 && (len(hdr)+len(msg)) > maxMsgSize {
 		err := fmt.Errorf("JetStream message size exceeds limits for '%s > %s'", jsa.acc().Name, mset.cfg.Name)
-		s.Warnf(err.Error())
+		s.RateLimitWarnf(err.Error())
 		if canRespond {
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSStreamMessageExceedsMaximumError()
@@ -5188,7 +5188,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	// Again this works if it goes through but better to be pre-emptive.
 	if len(hdr) > math.MaxUint16 {
 		err := fmt.Errorf("JetStream header size exceeds limits for '%s > %s'", jsa.acc().Name, mset.cfg.Name)
-		s.Warnf(err.Error())
+		s.RateLimitWarnf(err.Error())
 		if canRespond {
 			var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
 			resp.Error = NewJSStreamHeaderExceedsMaximumError()
diff --git a/server/log.go b/server/log.go
index 2bfe25c2..4b9193c0 100644
--- a/server/log.go
+++ b/server/log.go
@@ -14,9 +14,11 @@
 package server
 
 import (
+	"fmt"
 	"io"
 	"os"
 	"sync/atomic"
+	"time"
 
 	srvlog "github.com/nats-io/nats-server/v2/logger"
 )
@@ -204,6 +206,14 @@ func (s *Server) Warnf(format string, v ...interface{}) {
 	}, format, v...)
 }
 
+func (s *Server) RateLimitWarnf(format string, v ...interface{}) {
+	statement := fmt.Sprintf(format, v...)
+	if _, loaded := s.rateLimitLogging.LoadOrStore(statement, time.Now()); loaded {
+		return
+	}
+	s.Warnf("%s", statement)
+}
+
 // Fatalf logs a fatal error
 func (s *Server) Fatalf(format string, v ...interface{}) {
 	s.executeLogCall(func(logger Logger, format string, v ...interface{}) {
diff --git a/server/server.go b/server/server.go
index 1056eef1..b226db2d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -271,6 +271,10 @@ type Server struct {
 
 	// IPQueues map
 	ipQueues sync.Map
+
+	// To limit logging frequency
+	rateLimitLogging   sync.Map
+	rateLimitLoggingCh chan time.Duration
 }
 
 // For tracking JS nodes.
@@ -357,19 +361,20 @@ func NewServer(opts *Options) (*Server, error) {
 	now := time.Now().UTC()
 
 	s := &Server{
-		kp:           kp,
-		configFile:   opts.ConfigFile,
-		info:         info,
-		prand:        rand.New(rand.NewSource(time.Now().UnixNano())),
-		opts:         opts,
-		done:         make(chan bool, 1),
-		start:        now,
-		configTime:   now,
-		gwLeafSubs:   NewSublistWithCache(),
-		httpBasePath: httpBasePath,
-		eventIds:     nuid.New(),
-		routesToSelf: make(map[string]struct{}),
-		httpReqStats: make(map[string]uint64), // Used to track HTTP requests
+		kp:                 kp,
+		configFile:         opts.ConfigFile,
+		info:               info,
+		prand:              rand.New(rand.NewSource(time.Now().UnixNano())),
+		opts:               opts,
+		done:               make(chan bool, 1),
+		start:              now,
+		configTime:         now,
+		gwLeafSubs:         NewSublistWithCache(),
+		httpBasePath:       httpBasePath,
+		eventIds:           nuid.New(),
+		routesToSelf:       make(map[string]struct{}),
+		httpReqStats:       make(map[string]uint64), // Used to track HTTP requests
+		rateLimitLoggingCh: make(chan time.Duration, 1),
 	}
 
 	if opts.TLSRateLimit > 0 {
@@ -1601,6 +1606,8 @@ func (s *Server) Start() {
 	s.grRunning = true
 	s.grMu.Unlock()
 
+	s.startRateLimitLogExpiration()
+
 	// Pprof http endpoint for the profiler.
 	if opts.ProfPort != 0 {
 		s.StartProfiler()
@@ -3593,3 +3600,39 @@ func (s *Server) updateRemoteSubscription(acc *Account, sub *subscription, delta
 
 	s.updateLeafNodes(acc, sub, delta)
 }
+
+func (s *Server) startRateLimitLogExpiration() {
+	interval := time.Second
+	s.startGoRoutine(func() {
+		defer s.grWG.Done()
+
+		ticker := time.NewTicker(time.Second)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-s.quitCh:
+				return
+			case interval = <-s.rateLimitLoggingCh:
+				ticker.Reset(interval)
+			case <-ticker.C:
+				s.rateLimitLogging.Range(func(k, v interface{}) bool {
+					start := v.(time.Time)
+					if time.Since(start) >= interval {
+						s.rateLimitLogging.Delete(k)
+					}
+					return true
+				})
+			}
+		}
+	})
+}
+
+func (s *Server) changeRateLimitLogInterval(d time.Duration) {
+	if d <= 0 {
+		return
+	}
+	select {
+	case s.rateLimitLoggingCh <- d:
+	default:
+	}
+}
diff --git a/server/stream.go b/server/stream.go
index 99a2ec90..fe245cc1 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1625,7 +1625,7 @@ func (mset *stream) processInboundMirrorMsg(m *inMsg) bool {
 				mset.retryMirrorConsumer()
 			}
 		} else {
-			s.Warnf("Got error processing JetStream mirror msg: %v", err)
+			s.RateLimitWarnf("Got error processing JetStream mirror msg: %v", err)
 		}
 		if strings.Contains(err.Error(), "no space left") {
 			s.Errorf("JetStream out of space, will be DISABLED")
@@ -2236,7 +2236,7 @@ func (mset *stream) processInboundSourceMsg(si *sourceInfo, m *inMsg) bool {
 			mset.cancelSourceConsumer(si.iname)
 			mset.retrySourceConsumer(si.iname)
 		} else {
-			s.Warnf("JetStream got an error processing inbound source msg: %v", err)
+			s.RateLimitWarnf("JetStream got an error processing inbound source msg: %v", err)
 		}
 		if strings.Contains(err.Error(), "no space left") {
 			s.Errorf("JetStream out of space, will be DISABLED")
@@ -3138,7 +3138,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 			response, _ = json.Marshal(resp)
 		}
 	} else if exceeded, apiErr := jsa.limitsExceeded(stype, tierName); exceeded {
-		s.Warnf("JetStream resource limits exceeded for account: %q", accName)
+		s.RateLimitWarnf("JetStream resource limits exceeded for account: %q", accName)
 		if canRespond {
 			resp.PubAck = &PubAck{Stream: name}
 			if apiErr == nil {


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 534e05a1..e7c07f42 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -8199,7 +8199,7 @@ func TestJetStreamPanicDecodingConsumerState(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "JSC", 3)
 	defer c.shutdown()
 
-	rch := make(chan struct{}, 1)
+	rch := make(chan struct{}, 2)
 	nc, js := jsClientConnect(t, c.randomServer(),
 		nats.ReconnectWait(50*time.Millisecond),
 		nats.MaxReconnects(-1),
diff --git a/server/server_test.go b/server/server_test.go
index 820d4d7e..1ccdbb50 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -1966,3 +1966,110 @@ func TestServerLogsConfigurationFile(t *testing.T) {
 		t.Fatalf("Config file location was not reported in log: %s", log)
 	}
 }
+
+func TestServerRateLimitLogging(t *testing.T) {
+	s := RunServer(DefaultOptions())
+	defer s.Shutdown()
+
+	s.changeRateLimitLogInterval(100 * time.Millisecond)
+
+	l := &captureWarnLogger{warn: make(chan string, 100)}
+	s.SetLogger(l, false, false)
+
+	s.RateLimitWarnf("Warning number 1")
+	s.RateLimitWarnf("Warning number 2")
+	s.RateLimitWarnf("Warning number 1")
+	s.RateLimitWarnf("Warning number 2")
+
+	checkLog := func(c1, c2 *client) {
+		t.Helper()
+
+		nb1 := "Warning number 1"
+		nb2 := "Warning number 2"
+		gotOne := 0
+		gotTwo := 0
+		for done := false; !done; {
+			select {
+			case w := <-l.warn:
+				if strings.Contains(w, nb1) {
+					gotOne++
+				} else if strings.Contains(w, nb2) {
+					gotTwo++
+				}
+			case <-time.After(150 * time.Millisecond):
+				done = true
+			}
+		}
+		if gotOne != 1 {
+			t.Fatalf("Should have had only 1 warning for nb1, got %v", gotOne)
+		}
+		if gotTwo != 1 {
+			t.Fatalf("Should have had only 1 warning for nb2, got %v", gotTwo)
+		}
+
+		// Wait for more than the expiration interval
+		time.Sleep(200 * time.Millisecond)
+		if c1 == nil {
+			s.RateLimitWarnf(nb1)
+		} else {
+			c1.RateLimitWarnf(nb1)
+			c2.RateLimitWarnf(nb1)
+		}
+		gotOne = 0
+		for {
+			select {
+			case w := <-l.warn:
+				if strings.Contains(w, nb1) {
+					gotOne++
+				}
+			case <-time.After(200 * time.Millisecond):
+				if gotOne == 0 {
+					t.Fatalf("Warning was still suppressed")
+				} else if gotOne > 1 {
+					t.Fatalf("Should have had only 1 warning for nb1, got %v", gotOne)
+				} else {
+					// OK! we are done
+					return
+				}
+			}
+		}
+	}
+
+	checkLog(nil, nil)
+
+	nc1 := natsConnect(t, s.ClientURL(), nats.Name("c1"))
+	defer nc1.Close()
+	nc2 := natsConnect(t, s.ClientURL(), nats.Name("c2"))
+	defer nc2.Close()
+
+	var c1 *client
+	var c2 *client
+	s.mu.Lock()
+	for _, cli := range s.clients {
+		cli.mu.Lock()
+		switch cli.opts.Name {
+		case "c1":
+			c1 = cli
+		case "c2":
+			c2 = cli
+		}
+		cli.mu.Unlock()
+		if c1 != nil && c2 != nil {
+			break
+		}
+	}
+	s.mu.Unlock()
+	if c1 == nil || c2 == nil {
+		t.Fatal("Did not find the clients")
+	}
+
+	// Wait for more than the expiration interval
+	time.Sleep(200 * time.Millisecond)
+
+	c1.RateLimitWarnf("Warning number 1")
+	c1.RateLimitWarnf("Warning number 2")
+	c2.RateLimitWarnf("Warning number 1")
+	c2.RateLimitWarnf("Warning number 2")
+
+	checkLog(c1, c2)
+}