Repository: atlassian/gostatsd (go)

Issue:
Per metric-type expiry interval
How I can send to back the previous value, if the gauge is not updated at the next flush?


Pull request description:
Adds support for per metric-type expiry intervals.
Closes #300.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8cbb673..376c321 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+21.0.0
+------
+- Breaks out expiry interval in to per metric type expiry intervals.  See [README.md](README.md) for details.
+
 20.3.2
 ------
 - Use the current timestamp when sending to NewRelic backend, rather than the last updated timestamp of the metric, which may be old.
@@ -8,7 +12,7 @@
 
 20.3.0
 ------
-- Added experimental support for generating latency histograms for timers, more details in [README.md]
+- Added experimental support for generating latency histograms for timers, more details in [README.md](README.md).
 
 20.2.0
 ------
diff --git a/README.md b/README.md
index 7a29cba..5750e48 100644
--- a/README.md
+++ b/README.md
@@ -76,6 +76,12 @@ following configuration options:
   may cause blocking in the pipeline (back pressure).  A UDP only receiver will never use more than the number of
   configured parsers (`--max-parsers` option).  Defaults to the value of `--max-parsers`, but may require tuning for
   HTTP based servers.
+- `expiry-interval`: interval before metrics are expired, see `Metric expiry and persistence` section.  Defaults
+   to `5m`.  0 to disable, -1 for immediate.
+- `expiry-interval-counter`: interval before counters are expired, defaults to the value of `expiry-interval`.
+- `expiry-interval-gauge`: interval before gauges are expired, defaults to the value of `expiry-interval`.
+- `expiry-interval-set`: interval before sets are expired, defaults to the value of `expiry-interval`.
+- `expiry-interval-timer`: interval before timers are expired, defaults to the value of `expiry-interval`.
 - `flush-interval`: duration for how long to batch metrics before flushing. Should be an order of magnitude less than
   the upstream flush interval. Defaults to `1s`.
 - `transport`: see [TRANSPORT.md](TRANSPORT.md) for how to configure the transport.
@@ -83,6 +89,25 @@ following configuration options:
 - `custom-headers` : a map of strings that are added to each request sent to allow for additional network routing / request inspection.
   Not required, default is empty. Example: `--custom-headers='{"region" : "us-east-1", "service" : "event-producer"}'`
 
+
+Metric expiry and persistence
+-----------------------------
+After a metric has been sent to the server, the server will continue to send the metric to the configured backend until
+it expires, even if no additional metrics are sent from the client.  The value sent depends on the metric type:
+
+- `counter`: sends 0 for both rate and count
+- `gauge`: sends the last received value.
+- `set`: sends 0
+- `timer`: sends non-percentile values of 0.  Percentile values are not sent at all (see issue #135)
+
+Setting an expiry interval of 0 will persist metrics forever.  If metrics are not carefully controlled in such an
+environment, the server may run out of memory or overload the backend receiving the metrics.  Setting a negative expiry
+interval will result in metrics not being persisted at all.
+
+Each metric type has its own interval, which is configured using the following precedence (from highest to lowest):
+`expiry-interval-<type>` > `expiry-interval` > default (5 minutes).
+
+
 Configuring HTTP servers
 ------------------------
 The service supports multiple HTTP servers, with different configurations for different requirements.  All http servers
diff --git a/cmd/gostatsd/main.go b/cmd/gostatsd/main.go
index 9f6cd91..57c94a3 100644
--- a/cmd/gostatsd/main.go
+++ b/cmd/gostatsd/main.go
@@ -138,33 +138,43 @@ func constructServer(v *viper.Viper) (*statsd.Server, []gostatsd.Runnable, error
 	if err != nil {
 		return nil, nil, err
 	}
+
+	// Set defaults for expiry from the main expiry setting
+	v.SetDefault(gostatsd.ParamExpiryIntervalCounter, v.GetDuration(gostatsd.ParamExpiryInterval))
+	v.SetDefault(gostatsd.ParamExpiryIntervalGauge, v.GetDuration(gostatsd.ParamExpiryInterval))
+	v.SetDefault(gostatsd.ParamExpiryIntervalSet, v.GetDuration(gostatsd.ParamExpiryInterval))
+	v.SetDefault(gostatsd.ParamExpiryIntervalTimer, v.GetDuration(gostatsd.ParamExpiryInterval))
+
 	// Create server
 	return &statsd.Server{
-		Backends:            backendsList,
-		CachedInstances:     cachedInstances,
-		InternalTags:        v.GetStringSlice(gostatsd.ParamInternalTags),
-		InternalNamespace:   v.GetString(gostatsd.ParamInternalNamespace),
-		DefaultTags:         v.GetStringSlice(gostatsd.ParamDefaultTags),
-		Hostname:            v.GetString(gostatsd.ParamHostname),
-		SelfIP:              selfIP,
-		ExpiryInterval:      v.GetDuration(gostatsd.ParamExpiryInterval),
-		FlushInterval:       v.GetDuration(gostatsd.ParamFlushInterval),
-		IgnoreHost:          v.GetBool(gostatsd.ParamIgnoreHost),
-		MaxReaders:          v.GetInt(gostatsd.ParamMaxReaders),
-		MaxParsers:          v.GetInt(gostatsd.ParamMaxParsers),
-		MaxWorkers:          v.GetInt(gostatsd.ParamMaxWorkers),
-		MaxQueueSize:        v.GetInt(gostatsd.ParamMaxQueueSize),
-		MaxConcurrentEvents: v.GetInt(gostatsd.ParamMaxConcurrentEvents),
-		EstimatedTags:       v.GetInt(gostatsd.ParamEstimatedTags),
-		MetricsAddr:         v.GetString(gostatsd.ParamMetricsAddr),
-		Namespace:           v.GetString(gostatsd.ParamNamespace),
-		StatserType:         v.GetString(gostatsd.ParamStatserType),
-		PercentThreshold:    pt,
-		HeartbeatEnabled:    v.GetBool(gostatsd.ParamHeartbeatEnabled),
-		ReceiveBatchSize:    v.GetInt(gostatsd.ParamReceiveBatchSize),
-		ConnPerReader:       v.GetBool(gostatsd.ParamConnPerReader),
-		ServerMode:          v.GetString(gostatsd.ParamServerMode),
-		LogRawMetric:        v.GetBool(gostatsd.ParamLogRawMetric),
+		Backends:              backendsList,
+		CachedInstances:       cachedInstances,
+		InternalTags:          v.GetStringSlice(gostatsd.ParamInternalTags),
+		InternalNamespace:     v.GetString(gostatsd.ParamInternalNamespace),
+		DefaultTags:           v.GetStringSlice(gostatsd.ParamDefaultTags),
+		Hostname:              v.GetString(gostatsd.ParamHostname),
+		SelfIP:                selfIP,
+		ExpiryIntervalCounter: v.GetDuration(gostatsd.ParamExpiryIntervalCounter),
+		ExpiryIntervalGauge:   v.GetDuration(gostatsd.ParamExpiryIntervalGauge),
+		ExpiryIntervalSet:     v.GetDuration(gostatsd.ParamExpiryIntervalSet),
+		ExpiryIntervalTimer:   v.GetDuration(gostatsd.ParamExpiryIntervalTimer),
+		FlushInterval:         v.GetDuration(gostatsd.ParamFlushInterval),
+		IgnoreHost:            v.GetBool(gostatsd.ParamIgnoreHost),
+		MaxReaders:            v.GetInt(gostatsd.ParamMaxReaders),
+		MaxParsers:            v.GetInt(gostatsd.ParamMaxParsers),
+		MaxWorkers:            v.GetInt(gostatsd.ParamMaxWorkers),
+		MaxQueueSize:          v.GetInt(gostatsd.ParamMaxQueueSize),
+		MaxConcurrentEvents:   v.GetInt(gostatsd.ParamMaxConcurrentEvents),
+		EstimatedTags:         v.GetInt(gostatsd.ParamEstimatedTags),
+		MetricsAddr:           v.GetString(gostatsd.ParamMetricsAddr),
+		Namespace:             v.GetString(gostatsd.ParamNamespace),
+		StatserType:           v.GetString(gostatsd.ParamStatserType),
+		PercentThreshold:      pt,
+		HeartbeatEnabled:      v.GetBool(gostatsd.ParamHeartbeatEnabled),
+		ReceiveBatchSize:      v.GetInt(gostatsd.ParamReceiveBatchSize),
+		ConnPerReader:         v.GetBool(gostatsd.ParamConnPerReader),
+		ServerMode:            v.GetString(gostatsd.ParamServerMode),
+		LogRawMetric:          v.GetBool(gostatsd.ParamLogRawMetric),
 		HeartbeatTags: gostatsd.Tags{
 			fmt.Sprintf("version:%s", Version),
 			fmt.Sprintf("commit:%s", GitCommit),
diff --git a/defaults_and_params.go b/defaults_and_params.go
index e175aea..a29cacb 100644
--- a/defaults_and_params.go
+++ b/defaults_and_params.go
@@ -108,6 +108,14 @@ const (
 	ParamInternalNamespace = "internal-namespace"
 	// ParamExpiryInterval is the name of parameter with expiry interval for metrics.
 	ParamExpiryInterval = "expiry-interval"
+	// ParamExpiryIntervalCounter is the name of parameter which overrides counter expiry interval for metrics.
+	ParamExpiryIntervalCounter = "expiry-interval-counter"
+	// ParamExpiryIntervalGauge is the name of parameter with overrides gauge expiry interval for metrics.
+	ParamExpiryIntervalGauge = "expiry-interval-gauge"
+	// ParamExpiryIntervalSet is the name of parameter with overrides set expiry interval for metrics.
+	ParamExpiryIntervalSet = "expiry-interval-set"
+	// ParamExpiryIntervalTimer is the name of parameter with overrides timer expiry interval for metrics.
+	ParamExpiryIntervalTimer = "expiry-interval-timer"
 	// ParamFlushInterval is the name of parameter with metrics flush interval.
 	ParamFlushInterval = "flush-interval"
 	// ParamIgnoreHost is the name of parameter indicating if the source should be used as the host
@@ -161,7 +169,11 @@ const (
 // AddFlags adds flags to the specified FlagSet.
 func AddFlags(fs *pflag.FlagSet) {
 	fs.String(ParamCloudProvider, "", "If set, use the cloud provider to retrieve metadata about the sender")
-	fs.Duration(ParamExpiryInterval, DefaultExpiryInterval, "After how long do we expire metrics (0 to disable)")
+	fs.Duration(ParamExpiryInterval, DefaultExpiryInterval, "After how long do we expire metrics (0 to disable, -1 for immediate)")
+	fs.Duration(ParamExpiryIntervalCounter, DefaultExpiryInterval, "Overrides "+ParamExpiryInterval+" for counters")
+	fs.Duration(ParamExpiryIntervalGauge, DefaultExpiryInterval, "Overrides "+ParamExpiryInterval+" for gauges")
+	fs.Duration(ParamExpiryIntervalSet, DefaultExpiryInterval, "Overrides "+ParamExpiryInterval+" for sets")
+	fs.Duration(ParamExpiryIntervalTimer, DefaultExpiryInterval, "Overrides "+ParamExpiryInterval+" for timers")
 	fs.Duration(ParamFlushInterval, DefaultFlushInterval, "How often to flush metrics to the backends")
 	fs.Bool(ParamIgnoreHost, DefaultIgnoreHost, "Ignore the source for populating the hostname field of metrics")
 	fs.Int(ParamMaxReaders, DefaultMaxReaders, "Maximum number of socket readers")
diff --git a/pkg/statsd/aggregator.go b/pkg/statsd/aggregator.go
index 0d6d917..91b728e 100644
--- a/pkg/statsd/aggregator.go
+++ b/pkg/statsd/aggregator.go
@@ -23,21 +23,36 @@ type percentStruct struct {
 
 // MetricAggregator aggregates metrics.
 type MetricAggregator struct {
-	metricsReceived    uint64
-	metricMapsReceived uint64
-	expiryInterval     time.Duration // How often to expire metrics
-	percentThresholds  map[float64]percentStruct
-	now                func() time.Time // Returns current time. Useful for testing.
-	statser            stats.Statser
-	disabledSubtypes   gostatsd.TimerSubtypes
-	histogramLimit     uint32
-	metricMap          *gostatsd.MetricMap
+	metricsReceived       uint64
+	metricMapsReceived    uint64
+	expiryIntervalCounter time.Duration // How often to expire counters
+	expiryIntervalGauge   time.Duration // How often to expire gauges
+	expiryIntervalSet     time.Duration // How often to expire sets
+	expiryIntervalTimer   time.Duration // How often to expire timers
+	percentThresholds     map[float64]percentStruct
+	now                   func() time.Time // Returns current time. Useful for testing.
+	statser               stats.Statser
+	disabledSubtypes      gostatsd.TimerSubtypes
+	histogramLimit        uint32
+	metricMap             *gostatsd.MetricMap
 }
 
 // NewMetricAggregator creates a new MetricAggregator object.
-func NewMetricAggregator(percentThresholds []float64, expiryInterval time.Duration, disabled gostatsd.TimerSubtypes, histogramLimit uint32) *MetricAggregator {
+func NewMetricAggregator(
+	percentThresholds []float64,
+	expiryIntervalCounter time.Duration,
+	expiryIntervalGauge time.Duration,
+	expiryIntervalSet time.Duration,
+	expiryIntervalTimer time.Duration,
+	disabled gostatsd.TimerSubtypes,
+	histogramLimit uint32,
+) *MetricAggregator {
 	a := MetricAggregator{
-		expiryInterval:    expiryInterval,
+		expiryIntervalCounter: expiryIntervalCounter,
+		expiryIntervalGauge:   expiryIntervalGauge,
+		expiryIntervalSet:     expiryIntervalSet,
+		expiryIntervalTimer:   expiryIntervalTimer,
+
 		percentThresholds: make(map[float64]percentStruct, len(percentThresholds)),
 		now:               time.Now,
 		statser:           stats.NewNullStatser(), // Will probably be replaced via RunMetrics
@@ -187,8 +202,8 @@ func (a *MetricAggregator) Process(f ProcessFunc) {
 	f(a.metricMap)
 }
 
-func (a *MetricAggregator) isExpired(now, ts gostatsd.Nanotime) bool {
-	return a.expiryInterval != 0 && time.Duration(now-ts) > a.expiryInterval
+func isExpired(interval time.Duration, now, ts gostatsd.Nanotime) bool {
+	return interval != 0 && time.Duration(now-ts) > interval
 }
 
 func deleteMetric(key, tagsKey string, metrics gostatsd.AggregatedMetrics) {
@@ -205,7 +220,7 @@ func (a *MetricAggregator) Reset() {
 	nowNano := gostatsd.Nanotime(a.now().UnixNano())
 
 	a.metricMap.Counters.Each(func(key, tagsKey string, counter gostatsd.Counter) {
-		if a.isExpired(nowNano, counter.Timestamp) {
+		if isExpired(a.expiryIntervalCounter, nowNano, counter.Timestamp) {
 			deleteMetric(key, tagsKey, a.metricMap.Counters)
 		} else {
 			a.metricMap.Counters[key][tagsKey] = gostatsd.Counter{
@@ -217,7 +232,7 @@ func (a *MetricAggregator) Reset() {
 	})
 
 	a.metricMap.Timers.Each(func(key, tagsKey string, timer gostatsd.Timer) {
-		if a.isExpired(nowNano, timer.Timestamp) {
+		if isExpired(a.expiryIntervalTimer, nowNano, timer.Timestamp) {
 			deleteMetric(key, tagsKey, a.metricMap.Timers)
 		} else {
 			if hasHistogramTag(timer) {
@@ -240,14 +255,14 @@ func (a *MetricAggregator) Reset() {
 	})
 
 	a.metricMap.Gauges.Each(func(key, tagsKey string, gauge gostatsd.Gauge) {
-		if a.isExpired(nowNano, gauge.Timestamp) {
+		if isExpired(a.expiryIntervalGauge, nowNano, gauge.Timestamp) {
 			deleteMetric(key, tagsKey, a.metricMap.Gauges)
 		}
 		// No reset for gauges, they keep the last value until expiration
 	})
 
 	a.metricMap.Sets.Each(func(key, tagsKey string, set gostatsd.Set) {
-		if a.isExpired(nowNano, set.Timestamp) {
+		if isExpired(a.expiryIntervalSet, nowNano, set.Timestamp) {
 			deleteMetric(key, tagsKey, a.metricMap.Sets)
 		} else {
 			a.metricMap.Sets[key][tagsKey] = gostatsd.Set{
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index 87a3f4a..7026c5d 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -8,14 +8,15 @@ import (
 	"time"
 
 	"github.com/ash2k/stager"
-	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/stats"
-	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/web"
 	"github.com/libp2p/go-reuseport"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"golang.org/x/time/rate"
+
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/stats"
+	"github.com/atlassian/gostatsd/pkg/transport"
+	"github.com/atlassian/gostatsd/pkg/web"
 )
 
 // Server encapsulates all of the parameters necessary for starting up
@@ -26,7 +27,10 @@ type Server struct {
 	InternalTags              gostatsd.Tags
 	InternalNamespace         string
 	DefaultTags               gostatsd.Tags
-	ExpiryInterval            time.Duration
+	ExpiryIntervalCounter     time.Duration
+	ExpiryIntervalGauge       time.Duration
+	ExpiryIntervalSet         time.Duration
+	ExpiryIntervalTimer       time.Duration
 	FlushInterval             time.Duration
 	MaxReaders                int
 	MaxParsers                int
@@ -88,10 +92,13 @@ func (s *Server) createStandaloneSink() (gostatsd.PipelineHandler, []gostatsd.Ru
 
 	// Create the backend handler
 	factory := agrFactory{
-		percentThresholds: s.PercentThreshold,
-		expiryInterval:    s.ExpiryInterval,
-		disabledSubtypes:  s.DisabledSubTypes,
-		histogramLimit:    s.HistogramLimit,
+		percentThresholds:     s.PercentThreshold,
+		expiryIntervalCounter: s.ExpiryIntervalCounter,
+		expiryIntervalGauge:   s.ExpiryIntervalGauge,
+		expiryIntervalSet:     s.ExpiryIntervalSet,
+		expiryIntervalTimer:   s.ExpiryIntervalTimer,
+		disabledSubtypes:      s.DisabledSubTypes,
+		histogramLimit:        s.HistogramLimit,
 	}
 
 	backendHandler := NewBackendHandler(s.Backends, uint(s.MaxConcurrentEvents), s.MaxWorkers, s.MaxQueueSize, &factory)
@@ -244,12 +251,23 @@ func sendStopEvent(handler gostatsd.PipelineHandler, selfIP gostatsd.IP, hostnam
 }
 
 type agrFactory struct {
-	percentThresholds []float64
-	expiryInterval    time.Duration
-	disabledSubtypes  gostatsd.TimerSubtypes
-	histogramLimit    uint32
+	percentThresholds     []float64
+	expiryIntervalCounter time.Duration
+	expiryIntervalGauge   time.Duration
+	expiryIntervalSet     time.Duration
+	expiryIntervalTimer   time.Duration
+	disabledSubtypes      gostatsd.TimerSubtypes
+	histogramLimit        uint32
 }
 
 func (af *agrFactory) Create() Aggregator {
-	return NewMetricAggregator(af.percentThresholds, af.expiryInterval, af.disabledSubtypes, af.histogramLimit)
+	return NewMetricAggregator(
+		af.percentThresholds,
+		af.expiryIntervalCounter,
+		af.expiryIntervalGauge,
+		af.expiryIntervalSet,
+		af.expiryIntervalTimer,
+		af.disabledSubtypes,
+		af.histogramLimit,
+	)
 }


Test patch:
diff --git a/cmd/tester/main.go b/cmd/tester/main.go
index 002a421..724229d 100644
--- a/cmd/tester/main.go
+++ b/cmd/tester/main.go
@@ -8,12 +8,13 @@ import (
 	"runtime/pprof"
 	"time"
 
-	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/fakesocket"
-	"github.com/atlassian/gostatsd/pkg/statsd"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
+
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/fakesocket"
+	"github.com/atlassian/gostatsd/pkg/statsd"
 )
 
 var (
@@ -37,15 +38,18 @@ func main() {
 	}
 	if s.Benchmark != 0 {
 		server := statsd.Server{
-			DefaultTags:      gostatsd.DefaultTags,
-			ExpiryInterval:   gostatsd.DefaultExpiryInterval,
-			FlushInterval:    gostatsd.DefaultFlushInterval,
-			MaxReaders:       gostatsd.DefaultMaxReaders,
-			MaxWorkers:       gostatsd.DefaultMaxWorkers,
-			MaxQueueSize:     gostatsd.DefaultMaxQueueSize,
-			PercentThreshold: gostatsd.DefaultPercentThreshold,
-			ReceiveBatchSize: gostatsd.DefaultReceiveBatchSize,
-			Viper:            viper.New(),
+			DefaultTags:           gostatsd.DefaultTags,
+			ExpiryIntervalCounter: gostatsd.DefaultExpiryInterval,
+			ExpiryIntervalGauge:   gostatsd.DefaultExpiryInterval,
+			ExpiryIntervalSet:     gostatsd.DefaultExpiryInterval,
+			ExpiryIntervalTimer:   gostatsd.DefaultExpiryInterval,
+			FlushInterval:         gostatsd.DefaultFlushInterval,
+			MaxReaders:            gostatsd.DefaultMaxReaders,
+			MaxWorkers:            gostatsd.DefaultMaxWorkers,
+			MaxQueueSize:          gostatsd.DefaultMaxQueueSize,
+			PercentThreshold:      gostatsd.DefaultPercentThreshold,
+			ReceiveBatchSize:      gostatsd.DefaultReceiveBatchSize,
+			Viper:                 viper.New(),
 		}
 		ctx, cancelFunc := context.WithTimeout(context.Background(), time.Duration(s.Benchmark)*time.Second)
 		defer cancelFunc()
diff --git a/defaults_and_params_test.go b/defaults_and_params_test.go
new file mode 100644
index 0000000..aaa6f30
--- /dev/null
+++ b/defaults_and_params_test.go
@@ -0,0 +1,15 @@
+package gostatsd
+
+import (
+	"testing"
+
+	"github.com/spf13/pflag"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAddFlags(t *testing.T) {
+	require.NotPanics(t, func() {
+		fs := &pflag.FlagSet{}
+		AddFlags(fs)
+	})
+}
diff --git a/pkg/statsd/aggregator_test.go b/pkg/statsd/aggregator_test.go
index 23784e0..359a867 100644
--- a/pkg/statsd/aggregator_test.go
+++ b/pkg/statsd/aggregator_test.go
@@ -17,6 +17,9 @@ func newFakeAggregator() *MetricAggregator {
 	return NewMetricAggregator(
 		[]float64{90},
 		5*time.Minute,
+		5*time.Minute,
+		5*time.Minute,
+		5*time.Minute,
 		gostatsd.TimerSubtypes{},
 		math.MaxUint32,
 	)
@@ -315,7 +318,10 @@ func TestReset(t *testing.T) {
 	pastNano := gostatsd.Nanotime(now.Add(-30 * time.Second).UnixNano())
 
 	actual = newFakeAggregator()
-	actual.expiryInterval = 10 * time.Second
+	actual.expiryIntervalCounter = 10 * time.Second
+	actual.expiryIntervalGauge = 5 * time.Minute
+	actual.expiryIntervalSet = 5 * time.Minute
+	actual.expiryIntervalTimer = 5 * time.Minute
 	actual.metricMap.Counters["some"] = map[string]gostatsd.Counter{
 		"thing":       gostatsd.NewCounter(pastNano, 50, host, nil),
 		"other:thing": gostatsd.NewCounter(pastNano, 90, host, nil),
@@ -329,7 +335,10 @@ func TestReset(t *testing.T) {
 	assrt.Equal(expected.metricMap.Counters, actual.metricMap.Counters)
 
 	actual = newFakeAggregator()
-	actual.expiryInterval = 10 * time.Second
+	actual.expiryIntervalCounter = 5 * time.Minute
+	actual.expiryIntervalGauge = 5 * time.Minute
+	actual.expiryIntervalSet = 5 * time.Minute
+	actual.expiryIntervalTimer = 10 * time.Second
 	actual.metricMap.Timers["some"] = map[string]gostatsd.Timer{
 		"thing": gostatsd.NewTimer(pastNano, []float64{50}, host, nil),
 	}
@@ -342,7 +351,10 @@ func TestReset(t *testing.T) {
 	assrt.Equal(expected.metricMap.Timers, actual.metricMap.Timers)
 
 	actual = newFakeAggregator()
-	actual.expiryInterval = 10 * time.Second
+	actual.expiryIntervalCounter = 5 * time.Minute
+	actual.expiryIntervalGauge = 10 * time.Second
+	actual.expiryIntervalSet = 5 * time.Minute
+	actual.expiryIntervalTimer = 5 * time.Minute
 	actual.metricMap.Gauges["some"] = map[string]gostatsd.Gauge{
 		"thing":       gostatsd.NewGauge(pastNano, 50, host, nil),
 		"other:thing": gostatsd.NewGauge(pastNano, 90, host, nil),
@@ -356,7 +368,10 @@ func TestReset(t *testing.T) {
 	assrt.Equal(expected.metricMap.Gauges, actual.metricMap.Gauges)
 
 	actual = newFakeAggregator()
-	actual.expiryInterval = 10 * time.Second
+	actual.expiryIntervalCounter = 5 * time.Minute
+	actual.expiryIntervalGauge = 5 * time.Minute
+	actual.expiryIntervalSet = 10 * time.Second
+	actual.expiryIntervalTimer = 5 * time.Minute
 	actual.metricMap.Sets["some"] = map[string]gostatsd.Set{
 		"thing": gostatsd.NewSet(pastNano, map[string]struct{}{"user": {}}, host, nil),
 	}
@@ -375,16 +390,18 @@ func TestIsExpired(t *testing.T) {
 
 	now := gostatsd.Nanotime(time.Now().UnixNano())
 
-	ma := &MetricAggregator{expiryInterval: 0}
-	assrt.Equal(false, ma.isExpired(now, now))
+	ma := &MetricAggregator{
+		expiryIntervalCounter: 0,
+	}
+	assrt.Equal(false, isExpired(ma.expiryIntervalCounter, now, now))
 
-	ma.expiryInterval = 10 * time.Second
+	ma.expiryIntervalCounter = 10 * time.Second
 
 	ts := gostatsd.Nanotime(time.Now().Add(-30 * time.Second).UnixNano())
-	assrt.Equal(true, ma.isExpired(now, ts))
+	assrt.Equal(true, isExpired(ma.expiryIntervalCounter, now, ts))
 
 	ts = gostatsd.Nanotime(time.Now().Add(-1 * time.Second).UnixNano())
-	assrt.Equal(false, ma.isExpired(now, ts))
+	assrt.Equal(false, isExpired(ma.expiryIntervalCounter, now, ts))
 }
 
 func TestDisabledCount(t *testing.T) {
@@ -457,6 +474,9 @@ func TestDisabledLower(t *testing.T) {
 	ma := NewMetricAggregator(
 		[]float64{-90},
 		5*time.Minute,
+		5*time.Minute,
+		5*time.Minute,
+		5*time.Minute,
 		gostatsd.TimerSubtypes{},
 		math.MaxUint32,
 	)
diff --git a/pkg/statsd/statsd_test.go b/pkg/statsd/statsd_test.go
index a28c36c..45109b9 100644
--- a/pkg/statsd/statsd_test.go
+++ b/pkg/statsd/statsd_test.go
@@ -9,12 +9,13 @@ import (
 	"time"
 
 	"github.com/ash2k/stager/wait"
-	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
-	"github.com/atlassian/gostatsd/pkg/fakesocket"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 	"golang.org/x/time/rate"
+
+	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
+	"github.com/atlassian/gostatsd/pkg/fakesocket"
 )
 
 // TestStatsdThroughput emulates statsd work using fake network socket and null backend to
@@ -41,22 +42,25 @@ func TestStatsdThroughput(t *testing.T) {
 			CacheNegativeTTL:          gostatsd.DefaultCacheNegativeTTL,
 		})
 	s := Server{
-		Backends:            []gostatsd.Backend{backend},
-		CachedInstances:     cachedInstances,
-		DefaultTags:         gostatsd.DefaultTags,
-		ExpiryInterval:      gostatsd.DefaultExpiryInterval,
-		FlushInterval:       gostatsd.DefaultFlushInterval,
-		MaxReaders:          gostatsd.DefaultMaxReaders,
-		MaxParsers:          gostatsd.DefaultMaxParsers,
-		MaxWorkers:          gostatsd.DefaultMaxWorkers,
-		MaxQueueSize:        gostatsd.DefaultMaxQueueSize,
-		EstimatedTags:       1, // Travis has limited memory
-		PercentThreshold:    gostatsd.DefaultPercentThreshold,
-		HeartbeatEnabled:    gostatsd.DefaultHeartbeatEnabled,
-		ReceiveBatchSize:    gostatsd.DefaultReceiveBatchSize,
-		MaxConcurrentEvents: 2,
-		ServerMode:          "standalone",
-		Viper:               viper.New(),
+		Backends:              []gostatsd.Backend{backend},
+		CachedInstances:       cachedInstances,
+		DefaultTags:           gostatsd.DefaultTags,
+		ExpiryIntervalCounter: gostatsd.DefaultExpiryInterval,
+		ExpiryIntervalSet:     gostatsd.DefaultExpiryInterval,
+		ExpiryIntervalGauge:   gostatsd.DefaultExpiryInterval,
+		ExpiryIntervalTimer:   gostatsd.DefaultExpiryInterval,
+		FlushInterval:         gostatsd.DefaultFlushInterval,
+		MaxReaders:            gostatsd.DefaultMaxReaders,
+		MaxParsers:            gostatsd.DefaultMaxParsers,
+		MaxWorkers:            gostatsd.DefaultMaxWorkers,
+		MaxQueueSize:          gostatsd.DefaultMaxQueueSize,
+		EstimatedTags:         1, // Travis has limited memory
+		PercentThreshold:      gostatsd.DefaultPercentThreshold,
+		HeartbeatEnabled:      gostatsd.DefaultHeartbeatEnabled,
+		ReceiveBatchSize:      gostatsd.DefaultReceiveBatchSize,
+		MaxConcurrentEvents:   2,
+		ServerMode:            "standalone",
+		Viper:                 viper.New(),
 	}
 	ctxCached, cancelCached := context.WithCancel(context.Background())
 	defer cancelCached()