Merge pull request #70 from mjs/filter-nats-pending-limits

filter: Use nats_pending_max_mb

README.md

@@ -206,6 +206,11 @@ nats_subject_junkyard = "influx-spout-junk"
 # by the monitor).
 nats_subject_monitor = "influx-spout-monitor"
 
+# The maximum size that the pending buffer for the NATS subject that the filter
+# is reading from may become (in megabytes). Measurements will be dropped if
+# this limit is reached.
+nats_pending_max_mb = 200
+
 # The number of filter workers to spawn.
 workers = 8
 

filter/filter.go

@@ -91,6 +91,9 @@ func StartFilter(conf *config.Config) (_ *Filter, err error) {
 	if err != nil {
 		return nil, fmt.Errorf("NATS: failed to subscribe: %v", err)
 	}
+	if err := f.sub.SetPendingLimits(-1, conf.NATSPendingMaxMB*1024*1024); err != nil {
+		return nil, fmt.Errorf("NATS: failed to set pending limits: %v", err)
+	}
 
 	f.wg.Add(1)
 	go f.startStatistician(stats, rules)

filter/filter_medium_test.go

@@ -38,6 +38,7 @@ func testConfig() *config.Config {
 		NATSSubject:         []string{"filter-test"},
 		NATSSubjectMonitor:  "filter-test-monitor",
 		NATSSubjectJunkyard: "filter-junkyard",
+		NATSPendingMaxMB:    32,
 		Workers:             1,
 		MaxTimeDeltaSecs:    600,
 		Rule: []config.Rule{{

writer/writer.go

@@ -100,10 +100,10 @@ func StartWriter(c *config.Config) (_ *Writer, err error) {
 			jobs <- msg
 		})
 		if err != nil {
-			return nil, fmt.Errorf("subscription for %q failed: %v", subject, err)
+			return nil, fmt.Errorf("NATS: subscription for %q failed: %v", subject, err)
 		}
 		if err := sub.SetPendingLimits(-1, maxPendingBytes); err != nil {
-			return nil, fmt.Errorf("failed to set pending limits: %v", err)
+			return nil, fmt.Errorf("NATS: failed to set pending limits: %v", err)
 		}
 
 		w.wg.Add(1)