Repository: atlassian/gostatsd
Pull request #329 (2020-07-06 06:17:07)

Aligned time
This should be 3 PRs, but that's just a bunch of cycling through PRs for not a lot of point.

Each commit can be assessed completely on its own, and it's the last of what I've got pending. More to come over the coming days though, I'm sure.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 62b1806..1f37f90 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -19,10 +19,10 @@ install:
 
 script:
   - make build-all
-  - make test-race
+  - make coveralls
   - make check
+  - make test-race
   - make bench-race
-  - make coveralls
 
 after_success:
   - if [ "$TRAVIS_OS_NAME" == "linux" -a ! -z "$TRAVIS_TAG" ]; then
diff --git a/CHANGELOG.md b/CHANGELOG.md
index d9c6960..27eb073 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,9 @@
+26.0.0
+------
+- Adds new options `flush-offset` and `flush-aligned` to align the flush interval to a configurable multiple.  See [README.md](README.md) for details.
+- Adds documentation for many of the server options.
+- Internal library refactoring
+
 25.0.0
 ------
 - Adds InfluxDB as a backend, see [BACKENDS.md](BACKENDS.md) for details.
diff --git a/Makefile b/Makefile
index 5536448..8c106f7 100644
--- a/Makefile
+++ b/Makefile
@@ -46,7 +46,7 @@ build-race: fmt
 build-all: pb/gostatsd.pb.go
 	go install -v ./...
 
-test-all: fmt test test-race bench bench-race check cover
+test-all: fmt cover test-race bench bench-race check
 
 fmt:
 	gofmt -w=true -s $$(find . -type f -name '*.go' -not -path "./vendor/*" -not -path "./pb/*")
diff --git a/README.md b/README.md
index 0c2595d..ecd307f 100644
--- a/README.md
+++ b/README.md
@@ -55,6 +55,52 @@ The server can currently run in two modes: `standalone` and `forwarder`.  It is
 In `standalone` mode, raw metrics are processed and aggregated as normal, and aggregated data is submitted to
 configured backends (see below)
 
+This configuration mode allows the following configuration options:
+- `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-aligned`: whether or not the flush should be aligned.  Setting this will flush at an exact time interval.  With
+  a 10 second flush-interval, if the service happens to be started at 12:47:13, then flushing will occur at 12:47:20,
+  12:47:30, etc, rather than 12:47:23, 12:47:33, etc.  This removes query time ambiguity in a multi-server environment.
+  Defaults to `false`.
+- `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`.
+- `flush-offset`: offset for flush interval when flush alignment is enabled.  For example, with an offset of 7s and an
+  interval of 10s, it will flush at 12:47:10+7 = 12:47:17, etc.
+- `ignore-host`: indicates whether or not an explicit `host` field will be added to all incoming metrics and events.
+  Defaults to `false`
+- `max-readers`: the number of UDP receivers to run.  Defaults to 8 or the number of logical cores, whichever is less.
+- `max-parsers`: the number of workers available to parse metrics.  Defaults to the number of logical cores.
+- `max-workers`: the number of aggregators to process metrics.  Defaults to the number of logical cores.
+- `max-queue-size`: the size of the buffers between parsers and workers.  Defaults to `10000`, monitored via
+  `channel.*` metric, with `dispatch_aggregator_batch` and `dispatch_aggregator_map` channels.
+- `max-concurrent-events`: the maximum number of concurrent events to be dispatching.  Defaults to `1024`, monitored
+  via `channel.*` metric, with `backend_events_sem` channel.
+- `estimated-tags`: provides a hint to the system as to how many tags are expected to be seen on any particular metric,
+  so that memory can be pre-allocated and reducing churn.  Defaults to `4`.  Note: this is only a hint, and it is safe
+  to send more.
+- `log-raw-metric`: logs raw metrics received from the network.  Defaults to `false`.
+- `metrics-addr`: the address to listen to metrics on. Defaults to `:8125`.
+- `namespace`: a namespace to prefix all metrics with.  Defaults to ''.
+- `statser-type`: configures where internal metrics are sent to.  May be `internal` which sends them to the internal
+  processing pipeline, `logging` which logs them, `null` which drops them.  Defaults to `internal`, or `null` if the
+  NewRelic backend is enabled.
+- `percent-threshold`: configures the "percentiles" sent on timers.  Space separated string.  Defaults to `90`.
+- `heartbeat-enabled`: emits a metric named `heartbeat` every flush interval, tagged by `version` and `commit`.
+  Defaults to `false`.
+- `receive-batch-size`: the number of datagrams to attempt to read.  It is more CPU efficient to read multiple, however
+  it takes extra memory.  See [Memory allocation for read buffers] section below for details.  Defaults to 50.
+- `conn-per-reader`: attempts to create a connection for every UDP receiver.  Not supported by all OS versions.
+  Defaults to `false`.
+- `bad-lines-per-minute`: the number of metrics which fail to parse to log per minute.  This is used to prevent a bad
+  client spamming malformed statsd data, while still logging some information to enable troubleshooting.  Defaults to `0`.
+- `hostname`: sets the hostname on internal metrics
+- `timer-histogram-limit`: specifies the maximum number of buckets on histograms.  See [Timer histograms] below.
+
+
 In `forwarder` mode, raw metrics are collected from a frontend, and instead of being aggregated they are sent via http
 to another gostatsd server after passing through the processing pipeline (cloud provider, static tags, filtering, etc).
 
@@ -62,6 +108,10 @@ A `forwarder` server is intended to run on-host and collect metrics, forwarding
 service.  At present the central aggregation service can only scale vertically, but horizontal scaling through
 clustering is planned.
 
+Aligned flushing is deliberately not supported in `forwarder` mode, as it would impact the central aggregation server
+due to all for forwarder nodes transmitting at once, and the expectation that many forwarding flushes will occur per
+central flush anyway.
+
 Configuring `forwarder` mode requires a configuration file, with a section named `http-transport`.  The raw version
 spoken is not configurable per server (see [HTTP.md](HTTP.md) for version guarantees).  The configuration section allows the
 following configuration options:
@@ -76,16 +126,7 @@ 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.
-- `log-raw-metric`: logs raw metrics received from the network.  Defaults to `false`.
 - `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"}'`
 - `dynamic-headers` : similar with `custom-headers`, but the header values are extracted from metric tags matching the
@@ -95,6 +136,23 @@ following configuration options:
   is empty. Example: `--dynamic-headers='["region", "service"]'`.
   This is an experimental feature and it may be removed or changed in future versions.
 
+The following settings from the previous section are also supported:
+- `expiry-*`
+- `ignore-host`
+- `max-readers`
+- `max-parsers`
+- `estimated-tags`
+- `log-raw-metric`
+- `metrics-addr`
+- `namespace`
+- `statser-type`
+- `heartbeat-enabled`
+- `receive-batch-size`
+- `conn-per-reader`
+- `bad-lines-per-minute`
+- `hostname`
+- `log-raw-metric`
+
 
 Metric expiry and persistence
 -----------------------------
diff --git a/cmd/gostatsd/main.go b/cmd/gostatsd/main.go
index f114fbe..594edec 100644
--- a/cmd/gostatsd/main.go
+++ b/cmd/gostatsd/main.go
@@ -19,13 +19,13 @@ import (
 	"golang.org/x/time/rate"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/backends"
 	"github.com/atlassian/gostatsd/pkg/cachedinstances"
 	"github.com/atlassian/gostatsd/pkg/cachedinstances/cloudprovider"
 	"github.com/atlassian/gostatsd/pkg/cloudproviders"
 	"github.com/atlassian/gostatsd/pkg/statsd"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
@@ -160,6 +160,8 @@ func constructServer(v *viper.Viper) (*statsd.Server, error) {
 		ExpiryIntervalSet:     v.GetDuration(gostatsd.ParamExpiryIntervalSet),
 		ExpiryIntervalTimer:   v.GetDuration(gostatsd.ParamExpiryIntervalTimer),
 		FlushInterval:         v.GetDuration(gostatsd.ParamFlushInterval),
+		FlushOffset:           v.GetDuration(gostatsd.ParamFlushOffset),
+		FlushAligned:          v.GetBool(gostatsd.ParamFlushAligned),
 		IgnoreHost:            v.GetBool(gostatsd.ParamIgnoreHost),
 		MaxReaders:            v.GetInt(gostatsd.ParamMaxReaders),
 		MaxParsers:            v.GetInt(gostatsd.ParamMaxParsers),
diff --git a/cover.sh b/cover.sh
index eb38af1..18ed654 100755
--- a/cover.sh
+++ b/cover.sh
@@ -1,36 +1,3 @@
 #!/usr/bin/env bash
-
-function die() {
-  echo $*
-  exit 1
-}
-
-# Initialize coverage.out
-echo "mode: count" > coverage.out
-
-# Initialize error tracking
-ERROR=""
-
-declare -a packages=('' \
-    'cmd/gostatsd' 'cmd/tester'
-    'pkg/backends' 'pkg/backends/sender'
-    'pkg/backends/datadog' 'pkg/backends/graphite' 'pkg/backends/null' \
-    'pkg/backends/statsdaemon' 'pkg/backends/stdout' 'pkg/backends/newrelic'  \
-    'pkg/backends/influxdb' \
-    'pkg/cloudproviders' 'pkg/cloudproviders/aws' 'pkg/cloudproviders/k8s' \
-    'pkg/fakesocket' 'pkg/statsd' \
-    'pkg/util' 'pkg/stats' 'pkg/cluster/nodes' 'pkg/web');
-
-# Test each package and append coverage profile info to coverage.out
-for pkg in "${packages[@]}"
-do
-    go test -v -covermode=count -coverprofile=coverage_tmp.out "github.com/atlassian/gostatsd/$pkg" || ERROR="Error testing $pkg"
-    tail -n +2 coverage_tmp.out >> coverage.out 2> /dev/null ||:
-done
-
-rm -f coverage_tmp.out
-
-if [ ! -z "$ERROR" ]
-then
-    die "Encountered error, last error was: $ERROR"
-fi
+set -euo pipefail
+go test -covermode=count -coverprofile=coverage.out ./...
diff --git a/defaults_and_params.go b/defaults_and_params.go
index a29cacb..3324710 100644
--- a/defaults_and_params.go
+++ b/defaults_and_params.go
@@ -53,6 +53,10 @@ const (
 	DefaultExpiryInterval = 5 * time.Minute
 	// DefaultFlushInterval is the default metrics flush interval.
 	DefaultFlushInterval = 1 * time.Second
+	// DefaultFlushOffset is the default metrics flush interval offset when alignment is enabled
+	DefaultFlushOffset = 0
+	// DefaultFlushOffset is the default for whether metric flushing should be aligned
+	DefaultFlushAligned = false
 	// DefaultIgnoreHost is the default value for whether the source should be used as the host
 	DefaultIgnoreHost = false
 	// DefaultMetricsAddr is the default address on which to listen for metrics.
@@ -118,6 +122,10 @@ const (
 	ParamExpiryIntervalTimer = "expiry-interval-timer"
 	// ParamFlushInterval is the name of parameter with metrics flush interval.
 	ParamFlushInterval = "flush-interval"
+	// ParamFlushInterval is the name of parameter with metrics flush interval alignment.
+	ParamFlushOffset = "flush-offset"
+	// ParamFlushInterval is the name of parameter with metrics flush interval alignment enable state.
+	ParamFlushAligned = "flush-aligned"
 	// ParamIgnoreHost is the name of parameter indicating if the source should be used as the host
 	ParamIgnoreHost = "ignore-host"
 	// ParamMaxReaders is the name of parameter with number of socket readers.
@@ -175,6 +183,8 @@ func AddFlags(fs *pflag.FlagSet) {
 	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.Duration(ParamFlushOffset, DefaultFlushOffset, "Flush offset to use when flush alignment is enabled")
+	fs.Bool(ParamFlushAligned, DefaultFlushAligned, "Enable aligned flush interval")
 	fs.Bool(ParamIgnoreHost, DefaultIgnoreHost, "Ignore the source for populating the hostname field of metrics")
 	fs.Int(ParamMaxReaders, DefaultMaxReaders, "Maximum number of socket readers")
 	fs.Int(ParamMaxParsers, DefaultMaxParsers, "Maximum number of workers to parse datagrams into metrics")
diff --git a/internal/fixtures/clock.go b/internal/fixtures/clock.go
index f831070..fa70023 100644
--- a/internal/fixtures/clock.go
+++ b/internal/fixtures/clock.go
@@ -8,7 +8,8 @@ import (
 )
 
 // NewAdvancingClock attaches a virtual clock to a context which advances
-// at full speed (not wall speed), and a cancel function to stop it.
+// at full speed (not wall speed), and a cancel function to stop it.  The
+// clock also stops if the context is canceled.
 func NewAdvancingClock(ctx context.Context) (context.Context, func()) {
 	clck := clock.NewMock(time.Unix(1, 0))
 	ctx = clock.Context(ctx, clck)
@@ -18,12 +19,23 @@ func NewAdvancingClock(ctx context.Context) (context.Context, func()) {
 			select {
 			case <-ch:
 				return
+			case <-ctx.Done():
+				return
 			default:
 				clck.AddNext()
 			}
 		}
 	}()
 	return ctx, func() {
-		ch <- struct{}{}
+		close(ch)
+	}
+}
+
+// NextStep will advance the supplied clock.Mock until it moves, or the context.Context is canceled (which typically
+// means it timed out in wall-time).  This is useful when testing things that exist inside goroutines, when it's not
+// possible to tell when the goroutine is ready to consume mock time.
+func NextStep(ctx context.Context, clck *clock.Mock) {
+	for _, d := clck.AddNext(); d == 0 && ctx.Err() == nil; _, d = clck.AddNext() {
+		time.Sleep(time.Nanosecond) // Allows the system to actually idle, runtime.Gosched() does not.
 	}
 }
diff --git a/internal/util/aligned_ticker.go b/internal/util/aligned_ticker.go
new file mode 100644
index 0000000..2396fea
--- /dev/null
+++ b/internal/util/aligned_ticker.go
@@ -0,0 +1,99 @@
+package util
+
+import (
+	"context"
+	"time"
+
+	"github.com/tilinna/clock"
+)
+
+// AlignedTicker is a wrapper around time.Ticker will align the time on a specific interval.  The first tick will
+// always occur within interval, the caller should explicitly discard this on creation, or not.
+// Instead of firing at:
+// [T+1*interval, T+2*interval, T+3*interval, ...]
+//
+// It will fire at:
+// r = roundup(T, interval)+offset
+// [r+1*interval, r+interval*2, r+3*interval, ...]
+//
+// The time.Time sent to the channel is guaranteed to be r+offset+n*interval, rather than the actual time of firing.
+type AlignedTicker struct {
+	C          <-chan time.Time
+	chInternal chan time.Time
+	chStop     chan struct{}
+	interval   time.Duration
+	offset     time.Duration
+}
+
+func NewAlignedTicker(interval, offset time.Duration) *AlignedTicker {
+	return NewAlignedTickerWithContext(context.Background(), interval, offset)
+}
+
+func NewAlignedTickerWithContext(ctx context.Context, interval, offset time.Duration) *AlignedTicker {
+	ch := make(chan time.Time, 1)
+	at := &AlignedTicker{
+		C:          ch,
+		chInternal: ch,
+		chStop:     make(chan struct{}),
+		interval:   interval,
+		offset:     offset,
+	}
+	go at.start(ctx)
+	return at
+}
+
+func roundup(t time.Time, i time.Duration) time.Time {
+	return t.Truncate(i).Add(i)
+}
+
+func (at *AlignedTicker) start(ctx context.Context) {
+	clck := clock.FromContext(ctx)
+	now := clck.Now()
+	initialWait := roundup(now.Add(-at.offset), at.interval).Add(at.offset).Sub(now)
+	tmr := clck.NewTimer(initialWait)
+	var tckr *clock.Ticker
+
+	// Phase 1: wait for the next interval
+	select {
+	case now := <-tmr.C:
+		// Start the repeating timer as soon as possible
+		tckr = clck.NewTicker(at.interval)
+		defer tckr.Stop()
+		if !at.sendTick(now) {
+			tmr.Stop()
+			return
+		}
+	case <-at.chStop:
+		tmr.Stop()
+		return
+	}
+
+	// Phase 2: process the ticker
+	for {
+		select {
+		case now := <-tckr.C:
+			if !at.sendTick(now) {
+				return
+			}
+		case <-at.chStop:
+			tckr.Stop()
+			return
+		}
+	}
+}
+
+func (at *AlignedTicker) sendTick(t time.Time) bool {
+	rounded := t.Add(-at.offset).Truncate(at.interval).Add(at.offset)
+	select {
+	case at.chInternal <- rounded:
+		return true
+	case <-at.chStop:
+		return false
+	default:
+		return true
+	}
+}
+
+func (at *AlignedTicker) Stop() {
+	close(at.chStop)
+}
diff --git a/pkg/util/io.go b/internal/util/io.go
similarity index 100%
rename from pkg/util/io.go
rename to internal/util/io.go
diff --git a/pkg/util/io_test.go b/internal/util/io_test.go
similarity index 100%
rename from pkg/util/io_test.go
rename to internal/util/io_test.go
diff --git a/pkg/util/viper.go b/internal/util/viper.go
similarity index 100%
rename from pkg/util/viper.go
rename to internal/util/viper.go
diff --git a/pkg/backends/cloudwatch/cloudwatch.go b/pkg/backends/cloudwatch/cloudwatch.go
index e040b35..3041785 100644
--- a/pkg/backends/cloudwatch/cloudwatch.go
+++ b/pkg/backends/cloudwatch/cloudwatch.go
@@ -15,8 +15,8 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 // Maximum number of dimensions per metric
diff --git a/pkg/backends/datadog/datadog.go b/pkg/backends/datadog/datadog.go
index e05465e..4b47bee 100644
--- a/pkg/backends/datadog/datadog.go
+++ b/pkg/backends/datadog/datadog.go
@@ -23,9 +23,9 @@ import (
 	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/backends/graphite/graphite.go b/pkg/backends/graphite/graphite.go
index 34a2720..544e9da 100644
--- a/pkg/backends/graphite/graphite.go
+++ b/pkg/backends/graphite/graphite.go
@@ -16,9 +16,9 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/backends/sender"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/backends/influxdb/influxdb.go b/pkg/backends/influxdb/influxdb.go
index 5e3db47..efb4989 100644
--- a/pkg/backends/influxdb/influxdb.go
+++ b/pkg/backends/influxdb/influxdb.go
@@ -22,9 +22,9 @@ import (
 	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/backends/newrelic/newrelic.go b/pkg/backends/newrelic/newrelic.go
index 9047343..adcd28c 100644
--- a/pkg/backends/newrelic/newrelic.go
+++ b/pkg/backends/newrelic/newrelic.go
@@ -22,9 +22,9 @@ import (
 	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/backends/statsdaemon/statsdaemon.go b/pkg/backends/statsdaemon/statsdaemon.go
index c02cea9..303dcff 100644
--- a/pkg/backends/statsdaemon/statsdaemon.go
+++ b/pkg/backends/statsdaemon/statsdaemon.go
@@ -15,9 +15,9 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/backends/sender"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/cachedinstances/k8s/k8s.go b/pkg/cachedinstances/k8s/k8s.go
index 8d2ca23..3e21700 100644
--- a/pkg/cachedinstances/k8s/k8s.go
+++ b/pkg/cachedinstances/k8s/k8s.go
@@ -19,7 +19,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 
 	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/util"
+	"github.com/atlassian/gostatsd/internal/util"
 )
 
 var (
diff --git a/pkg/cloudproviders/aws/aws.go b/pkg/cloudproviders/aws/aws.go
index fb1784a..dd65886 100644
--- a/pkg/cloudproviders/aws/aws.go
+++ b/pkg/cloudproviders/aws/aws.go
@@ -20,8 +20,8 @@ import (
 	"golang.org/x/net/http2"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/stats"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/statsd/flusher.go b/pkg/statsd/flusher.go
index 42dd6e6..1a0abc6 100644
--- a/pkg/statsd/flusher.go
+++ b/pkg/statsd/flusher.go
@@ -8,8 +8,10 @@ import (
 	"time"
 
 	"github.com/sirupsen/logrus"
+	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pkg/stats"
 )
 
@@ -22,32 +24,47 @@ type MetricFlusher struct {
 	lastFlushError int64 // Time of the last flush error. Unix timestamp in nsec.
 
 	flushInterval      time.Duration // How often to flush metrics to the sender
+	flushOffset        time.Duration // Offset for when to flush if alignment is enabled
+	flushAligned       bool          // Indicate if flush is aligned to the interval or not
 	aggregateProcesser AggregateProcesser
 	backends           []gostatsd.Backend
 }
 
 // NewMetricFlusher creates a new MetricFlusher with provided configuration.
-func NewMetricFlusher(flushInterval time.Duration, aggregateProcesser AggregateProcesser, backends []gostatsd.Backend) *MetricFlusher {
+func NewMetricFlusher(flushInterval, flushOffset time.Duration, aligned bool, aggregateProcesser AggregateProcesser, backends []gostatsd.Backend) *MetricFlusher {
 	return &MetricFlusher{
 		flushInterval:      flushInterval,
+		flushOffset:        flushOffset,
+		flushAligned:       aligned,
 		aggregateProcesser: aggregateProcesser,
 		backends:           backends,
 	}
 }
 
+func (f *MetricFlusher) makeTicker(ctx context.Context) (<-chan time.Time, func()) {
+	if f.flushAligned {
+		flushTicker := util.NewAlignedTickerWithContext(ctx, f.flushInterval, f.flushOffset)
+		return flushTicker.C, flushTicker.Stop
+	} else {
+		clck := clock.FromContext(ctx)
+		flushTicker := clck.NewTicker(f.flushInterval)
+		return flushTicker.C, flushTicker.Stop
+	}
+}
+
 // Run runs the MetricFlusher.
 func (f *MetricFlusher) Run(ctx context.Context) {
 	statser := stats.FromContext(ctx)
 
-	flushTicker := time.NewTicker(f.flushInterval)
-	defer flushTicker.Stop()
+	ch, stop := f.makeTicker(ctx)
+	defer stop()
 
 	lastFlush := time.Now()
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case thisFlush := <-flushTicker.C: // Time to flush to the backends
+		case thisFlush := <-ch: // Time to flush to the backends
 			flushDelta := thisFlush.Sub(lastFlush)
 			if f.aggregateProcesser != AggregateProcesser(nil) {
 				f.flushData(ctx, flushDelta, statser)
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index 0707977..aed497b 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -21,10 +21,10 @@ import (
 	"github.com/tilinna/clock"
 
 	"github.com/atlassian/gostatsd"
+	"github.com/atlassian/gostatsd/internal/util"
 	"github.com/atlassian/gostatsd/pb"
 	"github.com/atlassian/gostatsd/pkg/stats"
 	"github.com/atlassian/gostatsd/pkg/transport"
-	"github.com/atlassian/gostatsd/pkg/util"
 )
 
 const (
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index 2f30be3..af7d550 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -33,6 +33,8 @@ type Server struct {
 	ExpiryIntervalSet         time.Duration
 	ExpiryIntervalTimer       time.Duration
 	FlushInterval             time.Duration
+	FlushOffset               time.Duration
+	FlushAligned              bool
 	MaxReaders                int
 	MaxParsers                int
 	MaxWorkers                int
@@ -106,7 +108,7 @@ func (s *Server) createStandaloneSink() (gostatsd.PipelineHandler, []gostatsd.Ru
 	runnables = append(runnables, backendHandler.Run, backendHandler.RunMetricsContext)
 
 	// Create the Flusher
-	flusher := NewMetricFlusher(s.FlushInterval, backendHandler, s.Backends)
+	flusher := NewMetricFlusher(s.FlushInterval, s.FlushOffset, s.FlushAligned, backendHandler, s.Backends)
 	runnables = append(runnables, flusher.Run)
 
 	return backendHandler, runnables, nil
@@ -123,7 +125,7 @@ func (s *Server) createForwarderSink(logger logrus.FieldLogger) (gostatsd.Pipeli
 	}
 
 	// Create a Flusher, this is primarily for all the periodic metrics which are emitted.
-	flusher := NewMetricFlusher(s.FlushInterval, nil, s.Backends)
+	flusher := NewMetricFlusher(s.FlushInterval, 0, false, nil, s.Backends)
 
 	return forwarderHandler, []gostatsd.Runnable{forwarderHandler.Run, forwarderHandler.RunMetricsContext, flusher.Run}, nil
 }
diff --git a/pkg/web/httpservers.go b/pkg/web/httpservers.go
index 20e3dcc..6ac7098 100644
--- a/pkg/web/httpservers.go
+++ b/pkg/web/httpservers.go
@@ -15,7 +15,7 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/atlassian/gostatsd"
-	"github.com/atlassian/gostatsd/pkg/util"
+	"github.com/atlassian/gostatsd/internal/util"
 )
 
 type httpServer struct {


Test patch:
diff --git a/internal/util/aligned_ticker_test.go b/internal/util/aligned_ticker_test.go
new file mode 100644
index 0000000..63844e2
--- /dev/null
+++ b/internal/util/aligned_ticker_test.go
@@ -0,0 +1,91 @@
+package util
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tilinna/clock"
+
+	"github.com/atlassian/gostatsd/internal/fixtures"
+)
+
+const ms = int64(time.Millisecond)
+
+func checkTime(t *testing.T, ctx context.Context, ch <-chan time.Time, expected time.Time) {
+	select {
+	case <-ctx.Done():
+		require.FailNow(t, "timed out")
+	case now := <-ch:
+		require.Equal(t, expected.UnixNano(), now.UnixNano())
+	}
+}
+
+func TestAlignedTickerSimple(t *testing.T) {
+	clck := clock.NewMock(time.Unix(1, 0))
+	ctx, cancel := context.WithTimeout(clock.Context(context.Background(), clck), 100*time.Millisecond)
+	defer cancel()
+	tckr := NewAlignedTickerWithContext(ctx, 1000*time.Millisecond, 0*time.Millisecond)
+
+	// First update will go from 1s -> 2s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(2, 0))
+
+	// Second update will go from 2s -> 3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(3, 0))
+
+	tckr.Stop()
+}
+
+func TestAlignedTickerInitialRound(t *testing.T) {
+	clck := clock.NewMock(time.Unix(1, int64(500*time.Millisecond)))
+	ctx, cancel := context.WithTimeout(clock.Context(context.Background(), clck), 100*time.Millisecond)
+	defer cancel()
+	tckr := NewAlignedTickerWithContext(ctx, 1000*time.Millisecond, 0*time.Millisecond)
+
+	// First update will go from 1.5s to 2s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(2, 0))
+
+	// Second update will go from 2s to 3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(3, 0))
+
+	tckr.Stop()
+}
+
+func TestAlignedTickerOffset(t *testing.T) {
+	clck := clock.NewMock(time.Unix(1, 300*ms))
+	ctx, cancel := context.WithTimeout(clock.Context(context.Background(), clck), 100*time.Millisecond)
+	defer cancel()
+	tckr := NewAlignedTickerWithContext(ctx, 1000*time.Millisecond, 300*time.Millisecond)
+
+	// First update will go from 1.3s to 2.3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(2, 300*ms))
+
+	// Second update will go from 2.3s to 3.3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(3, 300*ms))
+
+	tckr.Stop()
+}
+
+func TestAlignedTickerInitialRoundOffset(t *testing.T) {
+	clck := clock.NewMock(time.Unix(1, 0))
+	ctx, cancel := context.WithTimeout(clock.Context(context.Background(), clck), 100*time.Millisecond)
+	defer cancel()
+	tckr := NewAlignedTickerWithContext(ctx, 1000*time.Millisecond, 300*time.Millisecond)
+
+	// First update will go from 1s to 1.3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(1, 300*ms))
+
+	// Second update will go from 1.3s to 2.3s
+	fixtures.NextStep(ctx, clck)
+	checkTime(t, ctx, tckr.C, time.Unix(2, 300*ms))
+
+	tckr.Stop()
+}
diff --git a/pkg/statsd/flusher_test.go b/pkg/statsd/flusher_test.go
index 122df7e..55d0fe0 100644
--- a/pkg/statsd/flusher_test.go
+++ b/pkg/statsd/flusher_test.go
@@ -17,7 +17,7 @@ func TestFlusherHandleSendResultNoErrors(t *testing.T) {
 		errs := errs
 		t.Run(strconv.Itoa(pos), func(t *testing.T) {
 			t.Parallel()
-			fl := NewMetricFlusher(0, nil, nil)
+			fl := NewMetricFlusher(0, 0, false, nil, nil)
 			fl.handleSendResult(errs)
 
 			if fl.lastFlush == 0 || fl.lastFlushError != 0 {
@@ -39,7 +39,7 @@ func TestFlusherHandleSendResultError(t *testing.T) {
 		errs := errs
 		t.Run(strconv.Itoa(pos), func(t *testing.T) {
 			t.Parallel()
-			fl := NewMetricFlusher(0, nil, nil)
+			fl := NewMetricFlusher(0, 0, false, nil, nil)
 			fl.handleSendResult(errs)
 
 			if fl.lastFlushError == 0 || fl.lastFlush != 0 {
diff --git a/pkg/statsd/statsd_test.go b/pkg/statsd/statsd_test.go
index 45bca0b..222a8b9 100644
--- a/pkg/statsd/statsd_test.go
+++ b/pkg/statsd/statsd_test.go
@@ -53,6 +53,8 @@ func TestStatsdThroughput(t *testing.T) {
 		ExpiryIntervalGauge:   gostatsd.DefaultExpiryInterval,
 		ExpiryIntervalTimer:   gostatsd.DefaultExpiryInterval,
 		FlushInterval:         gostatsd.DefaultFlushInterval,
+		FlushOffset:           0,
+		FlushAligned:          false,
 		MaxReaders:            gostatsd.DefaultMaxReaders,
 		MaxParsers:            gostatsd.DefaultMaxParsers,
 		MaxWorkers:            gostatsd.DefaultMaxWorkers,