Repository: jumptrading/influx-spout (go)

Issue:
Make stats publish interval configurable
The statistician publish interval is currently fixed at 3s in each component.

Pull request description:
Make stats publish interval configurable
It was fixed at 3s but can now be set via new `stats_interval` configuration option.

Fixes #35.

Patch:
diff --git a/README.md b/README.md
index e67246d..a827407 100644
--- a/README.md
+++ b/README.md
@@ -159,6 +159,10 @@ read_buffer_size = "4MB"
 # by the monitor).
 nats_subject_monitor = "influx-spout-monitor"
 
+# The listener will publish its internal metrics to the monitor component at
+this interval.
+stats_interval = "3s"
+
 # The listener will serve Kubernetes liveness and readiness probes on this port
 # at /healthz and /readyz. Set to 0 (the default) to disable probes support.
 probe_port = 0
@@ -211,6 +215,10 @@ batch_max_size = "1MB"
 # consumption by the monitor).
 nats_subject_monitor = "influx-spout-monitor"
 
+# The HTTP listener will publish its internal metrics to the monitor component
+# at this interval.
+stats_interval = "3s"
+
 # The HTTP listener will serve Kubernetes liveness and readiness probes on this
 # port at /healthz and /readyz. Set to 0 (the default) to disable probes support.
 probe_port = 0
@@ -258,6 +266,10 @@ nats_max_pending_size = "200MB"
 # The number of filter workers to spawn.
 workers = 8
 
+# The filter will publish its internal metrics to the monitor component at this
+# interval.
+stats_interval = "3s"
+
 # The filter will serve Kubernetes liveness and readiness probes on this port at
 # /healthz and /readyz. Set to 0 (the default) to disable probes support.
 probe_port = 0
@@ -374,6 +386,10 @@ nats_max_pending_size = "200MB"
 # by the monitor).
 nats_subject_monitor = "influx-spout-monitor"
 
+# The writer will publish its internal metrics to the monitor component at this
+# interval.
+stats_interval = "3s"
+
 # The writer will serve Kubernetes liveness and readiness probes on this port at
 # /healthz and /readyz. Set to 0 (the default) to disable probes support.
 probe_port = 0
diff --git a/config/config.go b/config/config.go
index 08009ff..0cc5a6f 100644
--- a/config/config.go
+++ b/config/config.go
@@ -52,6 +52,7 @@ type Config struct {
 	NATSMaxPendingSize  datasize.ByteSize `toml:"nats_max_pending_size"`
 	Rule                []Rule            `toml:"rule"`
 	MaxTimeDelta        Duration          `toml:"max_time_delta"`
+	StatsInterval       Duration          `toml:"stats_interval"`
 	ProbePort           int               `toml:"probe_port"`
 	PprofPort           int               `toml:"pprof_port"`
 	Debug               bool              `toml:"debug"`
@@ -80,6 +81,7 @@ func newDefaultConfig() *Config {
 		ReadBufferSize:      4 * datasize.MB,
 		NATSMaxPendingSize:  200 * datasize.MB,
 		MaxTimeDelta:        Duration{10 * time.Minute},
+		StatsInterval:       Duration{3 * time.Second},
 		ProbePort:           0,
 		PprofPort:           0,
 	}
diff --git a/filter/filter.go b/filter/filter.go
index 3eabf8c..d108b13 100644
--- a/filter/filter.go
+++ b/filter/filter.go
@@ -187,7 +187,7 @@ func (f *Filter) startStatistician(st *stats.Stats, ruleSt *stats.AnonStats, rul
 		}
 
 		select {
-		case <-time.After(3 * time.Second):
+		case <-time.After(f.c.StatsInterval.Duration):
 		case <-f.stop:
 			return
 		}
diff --git a/listener/listener.go b/listener/listener.go
index 0799c67..8b0f80c 100644
--- a/listener/listener.go
+++ b/listener/listener.go
@@ -47,8 +47,6 @@ const (
 	maxUDPDatagramSize = 65536
 )
 
-var statsInterval = 3 * time.Second
-
 // StartListener initialises a listener, starts its statistician
 // goroutine and runs it's main loop. It never returns.
 //
@@ -418,7 +416,7 @@ func (l *Listener) startStatistician() {
 		lines := stats.SnapshotToPrometheus(l.stats.Snapshot(), time.Now(), labels)
 		l.nc.Publish(l.c.NATSSubjectMonitor, lines)
 		select {
-		case <-time.After(statsInterval):
+		case <-time.After(l.c.StatsInterval.Duration):
 		case <-l.stop:
 			return
 		}
diff --git a/writer/writer.go b/writer/writer.go
index c993d7a..afb56f7 100644
--- a/writer/writer.go
+++ b/writer/writer.go
@@ -273,7 +273,7 @@ func (w *Writer) startStatistician(subs []*nats.Subscription) {
 		w.nc.Flush()
 
 		select {
-		case <-time.After(3 * time.Second):
+		case <-time.After(w.c.StatsInterval.Duration):
 		case <-w.stop:
 			return
 		}


Test patch:
diff --git a/config/config_small_test.go b/config/config_small_test.go
index e186030..0f2b2cb 100644
--- a/config/config_small_test.go
+++ b/config/config_small_test.go
@@ -57,6 +57,8 @@ read_buffer_size = 43210
 nats_max_pending_size = "100MB"
 max_time_delta = "789s"
 
+stats_interval = "5s"
+
 probe_port = 6789
 pprof_port = 5432
 `
@@ -83,6 +85,7 @@ pprof_port = 5432
 	assert.Equal(t, "spout-monitor", conf.NATSSubjectMonitor, "Monitor subject must match")
 	assert.Equal(t, "nats://localhost:4222", conf.NATSAddress, "Address must match")
 
+	assert.Equal(t, 5*time.Second, conf.StatsInterval.Duration)
 	assert.Equal(t, 6789, conf.ProbePort)
 	assert.Equal(t, 5432, conf.PprofPort)
 }
@@ -109,6 +112,7 @@ func TestAllDefaults(t *testing.T) {
 	assert.Equal(t, 4*datasize.MB, conf.ReadBufferSize)
 	assert.Equal(t, 200*datasize.MB, conf.NATSMaxPendingSize)
 	assert.Equal(t, 10*time.Minute, conf.MaxTimeDelta.Duration)
+	assert.Equal(t, 3*time.Second, conf.StatsInterval.Duration)
 	assert.Equal(t, 0, conf.ProbePort)
 	assert.Equal(t, 0, conf.PprofPort)
 	assert.Equal(t, false, conf.Debug)
diff --git a/listener/listener_medium_test.go b/listener/listener_medium_test.go
index 51ef692..a292e82 100644
--- a/listener/listener_medium_test.go
+++ b/listener/listener_medium_test.go
@@ -56,12 +56,6 @@ var (
 	numLines = len(poetry)
 )
 
-func init() {
-	// Make the statistician report more often during tests (default
-	// is 3s). This makes the tests run faster.
-	statsInterval = 500 * time.Millisecond
-}
-
 func testConfig() *config.Config {
 	return &config.Config{
 		Mode:               "listener",
@@ -76,6 +70,10 @@ func testConfig() *config.Config {
 
 		Port:      listenPort,
 		ProbePort: probePort,
+
+		// Make the statistician report more often during tests. This
+		// makes the tests run faster.
+		StatsInterval: config.Duration{250 * time.Millisecond},
 	}
 }