Repository: jumptrading/influx-spout
Pull request #105 (2018-07-24 01:29:24)

New downsampler component
This pull request implements a downsampler component for influx-spout. This component reads measurements from one or more NATS subjects, averaging numeric fields over a sampling period (as set by `downsample_period`), emitting the averaged values every sampling period. For non-numeric fields, the last value seen for each sampling period is emitted. Measurements are emitted on clean time boundaries regardless of when the downsampler was started.

Downsampled lines are emitted to a NATS subject with the same name as the input subject but with a suffix as set by `downsample_suffix` (default is "-archive").

* [x] Configurable sampling interval
* [x] Configurable output subject suffix
* [x] Configuration validation
* [x] Tags are assumed to be ordered (not something we can actually be sure of, will likely implement this in the filter)
* [x] Tests for sampling batch
* [x] Correct handling of integers (using "i" suffix)
* [x] Handling unescaping and escaping of measurement lines
* [x] Integer field values may only be positive
* [x] Split downsampler output to avoid NATS message limit
* [x] Tests for downsampler
* [x] Metric counters
* [x] Include downsampler in end-to-end test
* [x] Handling of measurements which arrive slightly late (I have a plan for this)
* [x] Exclusion of timestamps which don't belong in the current batch.
* [x] Performance optimisation


mjs: Tag sorting is done in #107 
oplehto: In my testing some of the more data-intensive measurements are hitting this:  
```
publish error for measurement.test: nats: maximum payload exceeded
```

There should be a way to define the max batch size and split the measurements based on that so that we do not exceed the max NATS payload.



Patch:
diff --git a/README.md b/README.md
index 9a54539..de0ba37 100644
--- a/README.md
+++ b/README.md
@@ -32,35 +32,41 @@ The following diagram shows a typical influx-spout deployment, and the
 flow of data between the various components:
 
 ```
-                           +-----+              +------+
-                           | UDP |              | HTTP |
-                           +-----+              +------+
-                              |                     |
-                              v                     v
-                     +-----------------+   +-----------------+
-                     |                 |   |                 |
-                     |     Listener    |   |  HTTP Listener  |
-                     |                 |   |                 |
-                     +-----------------+   +-----------------+
-                              |                     |
-                              v                     v
- +----------+   +-------------------------------------------------+   +---------+
- |          |<--+                                                 |   |         |
- |  Filter  |   |                     NATS                        +-->| Monitor |
- |          +-->|                                                 |   |         |
- +----------+   +--------+----------------+----------------+------+   +----+----+
-                         |                |                |               |
-                         v                v                v               |
-                   +----------+     +----------+     +----------+          |
-                   |          |     |          |     |          |          |
-                   |  Writer  |     |  Writer  |     |  Writer  |          |
-                   |          |     |          |     |          |          |
-                   +-----+----+     +-----+----+     +-----+----+          |
-                         |                |                |               |
-                         v                v                v               V
-                   +----------+     +----------+     +----------+   +----------------+
-                   | InfluxDB |     | InfluxDB |     | InfluxDB |   | Metrics (HTTP) |
-                   +----------+     +----------+     +----------+   +----------------+
+                                 +-----+              +------+
+                                 | UDP |              | HTTP |
+                                 +--+--+              +---+--+
+                                    |                     |
+                                    v                     v
+                           +-----------------+   +-----------------+
+                           |                 |   |                 |
+                           |     Listener    |   |  HTTP Listener  |
+                           |                 |   |                 |
+                           +--------+--------+   +--------+--------+
+                                    |                     |
+                                    v                     v
+ +----------------+   +-------------------------------------------------+
+ |                |<--+                                                 |
+ |    Filter(s)   |   |                                                 |
+ |                +-->|                                                 |   +---------+
+ +----------------+   |                                                 |   |         |
+                      |                      NATS                       +-->| Monitor |
+ +----------------+   |                                                 |   |         |
+ |                |<--+                                                 |   +----+----+
+ | Downsampler(s) |   |                                                 |        |
+ |                +-->|                                                 |        |
+ +----------------+   +-------+----------------+----------------+-------+        |
+                              |                |                |                |
+                              v                v                v                |
+                        +----------+     +----------+     +----------+           |
+                        |          |     |          |     |          |           |
+                        |  Writer  |     |  Writer  |     |  Writer  |           |
+                        |          |     |          |     |          |           |
+                        +-----+----+     +-----+----+     +-----+----+           |
+                              |                |                |                |
+                              v                v                v                V
+                        +----------+     +----------+     +----------+   +----------------+
+                        | InfluxDB |     | InfluxDB |     | InfluxDB |   | Metrics (HTTP) |
+                        +----------+     +----------+     +----------+   +----------------+
 ```
 
 All the influx-spout components may be run on a single host or may be
@@ -339,6 +345,66 @@ subject = "not-web"
 Ordering of rules in the configuration is important. Only the first rule that
 matches a given measurement is applied.
 
+
+### Downsampler
+
+The downsampler reads measurements from one or more NATS subjects,
+accumlates them and emits sampled values at a configured sampling
+period. This is useful for generating long term archive feeds of
+measurements.
+
+Integer and float fields are averaged over each sampling period. For
+string and boolean fields, the latest value for each sampling interval
+is emitted.
+
+The input subjects to a downsampler are intended to be the outputs
+from the filter.
+
+The downsampler sends its output to NATS subjects that match the input
+subjects with a suffix as configured by `downsample_suffix` (default
+is `-archive`). For example, averaged measurements for an input
+subject of `monitoring` would be output by the downsampler to a
+NATS subject named `monitoring-archive`.
+
+Writers should be configured to consume the downsampler output
+subjects and write them to an InfluxDB instance for storage.
+
+```toml
+mode = "downsampler"  # Required
+
+# Address of NATS server.
+nats_address = "nats://localhost:4222"
+
+# Subjects to accumulate measurements from (these should be filter outputs).
+nats_subject = ["influx-spout"]
+
+# The period to sample measurements over. Averaged measurements will be
+# emitted for each period at this interval.
+downsample_period = "1m"
+
+# This suffix will be added to each input NATS subject name to form the
+# output subject names.
+downsample_suffix = "-archive"
+
+# The maximum size that the pending buffer for the NATS subject that the
+# downsampler is reading from may become. Measurements will be dropped if
+# this limit is reached.
+nats_max_pending_size = "200MB"
+
+# The downsampler will publish its own metrics to this NATS subject (for
+# consumption by the monitor).
+nats_subject_monitor = "influx-spout-monitor"
+
+# The downsampler will publish its internal metrics to the monitor component at
+# this interval.
+stats_interval = "3s"
+
+# The downsampler 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
+```
+
+
 ### Writer
 
 A writer is responsible for reading measurements from one or more NATS subjects,
diff --git a/cmd/influx-spout/main.go b/cmd/influx-spout/main.go
index 94e7f68..04fc1a5 100644
--- a/cmd/influx-spout/main.go
+++ b/cmd/influx-spout/main.go
@@ -25,6 +25,7 @@ import (
 	_ "net/http/pprof"
 
 	"github.com/jumptrading/influx-spout/config"
+	"github.com/jumptrading/influx-spout/downsampler"
 	"github.com/jumptrading/influx-spout/filter"
 	"github.com/jumptrading/influx-spout/listener"
 	"github.com/jumptrading/influx-spout/monitor"
@@ -101,6 +102,8 @@ func runComponent(configFile string) (out stoppable, err error) {
 		out, err = listener.StartHTTPListener(c)
 	case "filter":
 		out, err = filter.StartFilter(c)
+	case "downsampler":
+		out, err = downsampler.StartDownsampler(c)
 	case "writer":
 		if c.Workers == 0 {
 			// this seems to be an okay default from our testing experience:
diff --git a/config/config.go b/config/config.go
index 269f578..8ab038c 100644
--- a/config/config.go
+++ b/config/config.go
@@ -56,6 +56,8 @@ type Config struct {
 	ProbePort           int               `toml:"probe_port"`
 	PprofPort           int               `toml:"pprof_port"`
 	Debug               bool              `toml:"debug"`
+	DownsamplePeriod    Duration          `toml:"downsample_period"`
+	DownsampleSuffix    string            `toml:"downsample_suffix"`
 }
 
 // Rule contains the configuration for a single filter rule.
@@ -82,6 +84,7 @@ func newDefaultConfig() *Config {
 		ReadBufferSize:      4 * datasize.MB,
 		NATSMaxPendingSize:  200 * datasize.MB,
 		MaxTimeDelta:        Duration{10 * time.Minute},
+		DownsampleSuffix:    "-archive",
 		StatsInterval:       Duration{3 * time.Second},
 		ProbePort:           0,
 		PprofPort:           0,
@@ -99,11 +102,17 @@ func (c *Config) Validate() error {
 		return errors.New("pprof_port out of range")
 	}
 
+	if c.Mode != "downsampler" && c.DownsamplePeriod.Duration != time.Duration(0) {
+		return errors.New("downsample_period is only for downsampler mode")
+	}
+
 	switch c.Mode {
 	case "listener", "listener_http":
 		return c.validateListener()
 	case "filter":
 		return c.validateFilter()
+	case "downsampler":
+		return c.validateDownsampler()
 	case "writer":
 		return c.validateWriter()
 	case "monitor":
@@ -157,6 +166,16 @@ func (c *Config) validateFilter() error {
 	return nil
 }
 
+func (c *Config) validateDownsampler() error {
+	if c.DownsamplePeriod.Duration == time.Duration(0) {
+		return errors.New("downsample_period must be set")
+	}
+	if c.DownsampleSuffix == "" {
+		return errors.New("downsample_suffix must be set")
+	}
+	return nil
+}
+
 func (c *Config) validateWriter() error {
 	if len(c.NATSSubject) < 1 {
 		return errors.New("writer needs at least one NATS subject")
@@ -219,6 +238,10 @@ func NewConfigFromFile(fileName string) (*Config, error) {
 		}
 	}
 
+	if conf.Mode == "downsampler" && conf.DownsamplePeriod.IsZero() {
+		conf.DownsamplePeriod.Duration = time.Minute
+	}
+
 	if err := conf.Validate(); err != nil {
 		return nil, err
 	}
diff --git a/config/duration.go b/config/duration.go
index 9791fee..c7efdd1 100644
--- a/config/duration.go
+++ b/config/duration.go
@@ -29,3 +29,8 @@ func (d *Duration) UnmarshalText(text []byte) error {
 	d.Duration, err = time.ParseDuration(string(text))
 	return err
 }
+
+// IsZero returns true if the duration is 0.
+func (d *Duration) IsZero() bool {
+	return d.Duration == time.Duration(0)
+}
diff --git a/downsampler/assigner.go b/downsampler/assigner.go
new file mode 100644
index 0000000..29b2e83
--- /dev/null
+++ b/downsampler/assigner.go
@@ -0,0 +1,201 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package downsampler
+
+import (
+	"bytes"
+	"fmt"
+	"time"
+
+	"github.com/jumptrading/influx-spout/influx"
+)
+
+// bucket defines the interface for some type which collects lines for
+// some time slot.
+type bucket interface {
+	AddLine([]byte) []error
+	Bytes() []byte
+	EndTime() time.Time
+}
+
+// bucketFactory defines the type of a function which creates new
+// buckets.
+type bucketFactory func(time.Time) bucket
+
+// newAssigner creates a assigner using the sampling period
+// specified. The returned assigner creates buckets using the provided
+// factory function and obtains the current time using the provided
+// clock.
+func newAssigner(period time.Duration, newBucket bucketFactory, clock clock) *assigner {
+	a := &assigner{
+		period:    period,
+		clock:     clock,
+		newBucket: newBucket,
+	}
+	tc := a.nextT(clock.Now())
+	tp := a.prevT(tc)
+	tn := a.nextT(tc)
+	a.buckets[prev] = newBucket(tp)
+	a.buckets[curr] = newBucket(tc)
+	a.buckets[next] = newBucket(tn)
+	a.tmin = tp.Add(-a.period)
+	return a
+}
+
+// These consts define the indexes for the buckets available to the
+// assigner at any given point in time.
+const (
+	prev = 0
+	curr = 1
+	next = 2
+)
+
+/* assigner takes batches of measurement lines, extracts the timestamp
+from each line and assigns each line to one of 3 buckets (previous,
+current or next). Multiple buckets are used in order to account for
+clock skew between the receiving host and the hosts that generated the
+measurements.
+
+Each bucket aggregates measurements for a specific time range and has
+a width of the downsampler sampling period.
+
+When the clock reaches the end time of the "current" bucket, the
+"previous" bucket is emitted and discarded. The "current" bucket then
+becomes the "previous" bucket, "next becomes "current" and a new
+"next" bucket is created.
+
+Measurements which fall outside of the range of the buckets being
+tracked by the assigner are discarded (errors are reported).
+*/
+type assigner struct {
+	period    time.Duration
+	clock     clock
+	newBucket bucketFactory
+	buckets   [3]bucket
+	tmin      time.Time
+}
+
+// Update takes a byte slice of measurement lines and assigns them to
+// an internal bucket based on timestamp. Any lines which contain
+// timestamps which fall outside of the range of the assigner's
+// buckets will result in an error. Data errors for any lines will
+// also result in errors being returned.
+func (a *assigner) Update(lines []byte) (errs []error) {
+	for _, line := range bytes.Split(lines, []byte{'\n'}) {
+		if len(line) < 1 {
+			continue
+		}
+
+		ts, line := a.extractTimestamp(line)
+		b, err := a.findBucket(ts)
+		if err != nil {
+			errs = append(errs, err)
+		} else {
+			lineErrs := b.AddLine(line)
+			errs = append(errs, lineErrs...)
+		}
+	}
+	return errs
+}
+
+// Bytes returns the aggregated lines for the oldest bucket, if it is
+// ready to be returned. Nil is returned otherwise.
+//
+// When data is returned, the oldest bucket is discarded and the other
+// buckets are cycled down.
+func (a *assigner) Bytes() []byte {
+	if a.clock.Now().Before(a.buckets[curr].EndTime()) {
+		return nil
+	}
+
+	out := a.buckets[prev].Bytes()
+	a.buckets[prev] = a.buckets[curr]
+	a.buckets[curr] = a.buckets[next]
+	a.buckets[next] = a.newBucket(a.nextT(a.buckets[curr].EndTime()))
+	a.tmin = a.buckets[prev].EndTime().Add(-a.period)
+	return out
+}
+
+// UntilNext returns the time until the next bucket will be released
+// for output.
+func (a *assigner) UntilNext() time.Duration {
+	d := a.buckets[curr].EndTime().Sub(a.clock.Now())
+	if d < time.Duration(0) {
+		return time.Duration(0)
+	}
+	return d
+}
+
+func (a *assigner) prevT(t time.Time) time.Time {
+	return t.Add(-a.period).Truncate(a.period)
+}
+
+func (a *assigner) nextT(t time.Time) time.Time {
+	return t.Add(a.period).Truncate(a.period)
+}
+
+func (a *assigner) findBucket(ts time.Time) (bucket, error) {
+	if ts.Before(a.tmin) {
+		return nil, newTimestampError("too old", ts)
+	}
+	for _, b := range a.buckets {
+		if ts.Before(b.EndTime()) {
+			return b, nil
+		}
+	}
+	return nil, newTimestampError("too new", ts)
+}
+
+// extractTimestamp returns the timestamp from the end of the line and
+// returns the line with the timestamp removed. The current time is
+// used if no timestamp was present.
+func (a *assigner) extractTimestamp(line []byte) (time.Time, []byte) {
+	nanos, offset := influx.ExtractTimestamp(line)
+	if offset < 0 {
+		return a.clock.Now(), line
+	}
+	return time.Unix(0, nanos), line[:offset-1]
+}
+
+// clock define an interface for retrieving the current time.
+type clock interface {
+	Now() time.Time
+}
+
+// realClock conforms to the clock interface and reports the actual
+// time.
+type realClock struct{}
+
+func (c *realClock) Now() time.Time {
+	return time.Now()
+}
+
+func isTimestampError(err error) bool {
+	_, ok := err.(*timestampError)
+	return ok
+}
+
+func newTimestampError(reason string, t time.Time) *timestampError {
+	return &timestampError{reason: reason, t: t}
+}
+
+type timestampError struct {
+	reason string
+	t      time.Time
+}
+
+func (e *timestampError) Error() string {
+	return fmt.Sprintf("timestamp %s: %s", e.reason, e.t)
+}
diff --git a/downsampler/avgbucket.go b/downsampler/avgbucket.go
new file mode 100644
index 0000000..61b6515
--- /dev/null
+++ b/downsampler/avgbucket.go
@@ -0,0 +1,261 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package downsampler
+
+import (
+	"errors"
+	"fmt"
+	"sort"
+	"strconv"
+	"time"
+
+	"github.com/jumptrading/influx-spout/convert"
+	"github.com/jumptrading/influx-spout/influx"
+)
+
+func newAvgBucket(ts time.Time) bucket {
+	return &avgBucket{
+		lines: make(map[string]*fieldPairs),
+		ts:    ts,
+	}
+}
+
+type avgBucket struct {
+	lines map[string]*fieldPairs
+	ts    time.Time
+}
+
+func (b *avgBucket) EndTime() time.Time {
+	return b.ts
+}
+
+func (b *avgBucket) AddLine(line []byte) (errs []error) {
+	var keyBytes []byte
+	keyBytes, line = influx.Token(line, []byte(" ")) // key = measurement + tags
+	key := string(keyBytes)
+
+	// Assumption: tags are already ordered (filter does this).
+	fields, found := b.lines[key]
+	if !found {
+		fields = newFieldPairs()
+	}
+	if updateErrs := fields.update(line); len(updateErrs) > 0 {
+		for _, err := range updateErrs {
+			errs = append(errs, fmt.Errorf("error parsing [%s]: %v", line, err))
+		}
+		return
+	}
+	b.lines[key] = fields
+	return
+}
+
+func (b *avgBucket) Bytes() []byte {
+	tsBytes := strconv.AppendInt(nil, b.ts.UnixNano(), 10)
+	var out []byte
+
+	for key, fields := range b.lines {
+		out = append(out, key...)
+		out = append(out, ' ')
+		fieldBytes := fields.bytes()
+		if len(fieldBytes) > 0 {
+			out = append(out, fieldBytes...)
+			out = append(out, ' ')
+		}
+		out = append(out, tsBytes...)
+		out = append(out, '\n')
+	}
+	return out
+}
+
+func newFieldPairs() *fieldPairs {
+	return &fieldPairs{
+		fields: make(map[string]fieldValue),
+	}
+}
+
+type fieldPairs struct {
+	fields map[string]fieldValue
+}
+
+func (fp *fieldPairs) count() int {
+	return len(fp.fields)
+}
+
+func (fp *fieldPairs) update(raw []byte) (errs []error) {
+	for {
+		if len(raw) == 0 {
+			return
+		}
+		raw = raw[1:] // remove leading comma or space
+		var nameBytes []byte
+		nameBytes, raw = influx.Token(raw, []byte("="))
+		if len(raw) == 0 || raw[0] != '=' {
+			errs = append(errs, errors.New("invalid field"))
+			return
+		}
+		name := string(nameBytes)
+
+		raw = raw[1:]
+		if len(raw) == 0 {
+			errs = append(errs, errors.New("missing field value"))
+			return
+		}
+
+		var rawValue []byte
+		var err error
+		if raw[0] == '"' {
+			// String field
+			rawValue, raw, err = influx.QuotedString(raw)
+			if err != nil {
+				errs = append(errs, err)
+				return
+			}
+			value, exists := fp.fields[name]
+			if !exists {
+				fp.fields[name] = newStringValue(rawValue)
+			} else {
+				if err := value.update(rawValue); err != nil {
+					errs = append(errs, err) // Non-fatal, so keep going.
+				}
+			}
+		} else {
+			// Other field
+			rawValue, raw = influx.Token(raw, []byte{','})
+			value, exists := fp.fields[name]
+			if !exists {
+				fp.fields[name] = newFieldValue(rawValue)
+			} else {
+				if err := value.update(rawValue); err != nil {
+					errs = append(errs, err) // Non-fatal, so keep going.
+				}
+			}
+		}
+	}
+}
+
+func (fp *fieldPairs) bytes() []byte {
+	if len(fp.fields) == 0 {
+		return nil
+	}
+
+	fieldNames := make([]string, 0, len(fp.fields))
+	for name := range fp.fields {
+		fieldNames = append(fieldNames, name)
+	}
+	sort.Strings(fieldNames)
+
+	var out []byte
+	lastI := len(fieldNames) - 1
+	for i, name := range fieldNames {
+		value := fp.fields[name]
+		out = append(out, name...)
+		out = append(out, '=')
+		out = append(out, value.bytes()...)
+		if i != lastI {
+			out = append(out, ',')
+		}
+	}
+	return out
+}
+
+type fieldValue interface {
+	update([]byte) error
+	bytes() []byte
+}
+
+func newFieldValue(v []byte) fieldValue {
+	if len(v) >= 2 && v[len(v)-1] == 'i' {
+		if iv, err := convert.ToInt(v[:len(v)-1]); err == nil {
+			return &intFieldValue{
+				average: iv,
+				count:   1,
+			}
+		}
+	}
+	if fv, err := strconv.ParseFloat(string(v), 64); err == nil {
+		return &floatFieldValue{
+			average: fv,
+			count:   1,
+		}
+	}
+	return &rawFieldValue{v}
+}
+
+func newStringValue(v []byte) fieldValue {
+	return &stringFieldValue{v}
+}
+
+type intFieldValue struct {
+	average int64
+	count   int64
+}
+
+func (v *intFieldValue) update(b []byte) error {
+	if len(b) < 2 || b[len(b)-1] != 'i' {
+		return fmt.Errorf("wrong type for int: %s", b)
+	}
+	bi, err := convert.ToInt(b[:len(b)-1])
+	if err != nil {
+		return fmt.Errorf("wrong type for int: %s", b)
+	}
+
+	// Update incremental average.
+	v.count++
+	v.average = v.average + ((bi - v.average) / v.count)
+	return nil
+}
+
+func (v *intFieldValue) bytes() []byte {
+	return append(strconv.AppendInt(nil, v.average, 10), 'i')
+}
+
+type floatFieldValue struct {
+	average float64
+	count   int64
+}
+
+func (v *floatFieldValue) update(b []byte) error {
+	bf, err := strconv.ParseFloat(string(b), 64)
+	if err != nil {
+		return fmt.Errorf("wrong type for float: %s", b)
+	}
+
+	// Update incremental average.
+	v.count++
+	v.average = v.average + ((bf - v.average) / float64(v.count))
+	return nil
+}
+
+func (v *floatFieldValue) bytes() []byte {
+	return []byte(strconv.AppendFloat(nil, v.average, 'g', -1, 64))
+}
+
+type stringFieldValue struct{ b []byte }
+
+func (v *stringFieldValue) update(b []byte) error {
+	v.b = b
+	return nil
+}
+func (v *stringFieldValue) bytes() []byte {
+	return influx.EscapeQuotedString(v.b)
+}
+
+type rawFieldValue struct{ b []byte }
+
+func (v *rawFieldValue) update(b []byte) error {
+	v.b = b
+	return nil
+}
+func (v *rawFieldValue) bytes() []byte { return v.b }
diff --git a/downsampler/batch_splitter.go b/downsampler/batch_splitter.go
new file mode 100644
index 0000000..a5303e9
--- /dev/null
+++ b/downsampler/batch_splitter.go
@@ -0,0 +1,73 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package downsampler
+
+// newBatchSplitter returns a batchSplitter which will efficiently
+// split BUF into chunks no larger than CHUNKSIZE with splits occuring
+// on newline boundaries.
+//
+// Use repeated calls to Next to generate the splits.
+//
+// Under no circumstances will chunks of more than chunkSize be
+// returned. If a line is larger than chunkSize, it will be broken up.
+func newBatchSplitter(buf []byte, chunkSize int) *batchSplitter {
+	return &batchSplitter{
+		buf:       buf,
+		chunkSize: chunkSize,
+	}
+}
+
+type batchSplitter struct {
+	buf       []byte
+	chunkSize int
+	out       []byte
+}
+
+// Next scans for the next chunk, returning true if there was another
+// chunk to return. Use Chunk() to retrieve the bytes for the chunk.
+func (s *batchSplitter) Next() bool {
+	if len(s.buf) == 0 {
+		s.out = nil
+		return false
+	}
+	if len(s.buf) <= s.chunkSize {
+		s.out = s.buf
+		s.buf = nil
+		return true
+	}
+
+	// Look backwards from the maximum size for a newline.
+	for i := s.chunkSize; i >= 0; i-- {
+		if s.buf[i] == '\n' {
+			i++
+			s.out = s.buf[:i]
+			s.buf = s.buf[i:]
+			return true
+		}
+	}
+
+	// No newline found, which means that a line was bigger than
+	// chunkSize. This is bad (but highly unlikely in practice). Split
+	// up the line at chunkSize.
+	s.out = s.buf[:s.chunkSize]
+	s.buf = s.buf[s.chunkSize:]
+	return true
+}
+
+// Chunk returns the chunk found by the last call to Next(). The
+// returned slice is only useful until the next call to Next().
+func (s *batchSplitter) Chunk() []byte {
+	return s.out
+}
diff --git a/downsampler/downsampler.go b/downsampler/downsampler.go
new file mode 100644
index 0000000..3a6548d
--- /dev/null
+++ b/downsampler/downsampler.go
@@ -0,0 +1,227 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package downsampler
+
+import (
+	"fmt"
+	"log"
+	"sync"
+	"time"
+
+	"github.com/nats-io/go-nats"
+
+	"github.com/jumptrading/influx-spout/config"
+	"github.com/jumptrading/influx-spout/probes"
+	"github.com/jumptrading/influx-spout/stats"
+)
+
+const maxNATSMsgSize = 1024 * 1024
+
+const (
+	statReceived          = "received"
+	statSent              = "sent"
+	statInvalidLines      = "invalid_lines"
+	statInvalidTimestamps = "invalid_timestamps"
+	statFailedNATSPublish = "failed_nats_publish"
+	statNATSDropped       = "nats_dropped"
+)
+
+// Downsampler consumes lines from one or more NATS subjects, and
+// applies downsampling to the InfluxDB measurements received from
+// them.
+type Downsampler struct {
+	c                *config.Config
+	nc               *nats.Conn
+	wg               sync.WaitGroup
+	probes           probes.Probes
+	stop             chan struct{}
+	stats            *stats.Stats
+	statsNATSDropped *stats.AnonStats
+	subscriptions    []*nats.Subscription
+}
+
+// StartDownsampler creates and configures a Downsampler.
+func StartDownsampler(c *config.Config) (_ *Downsampler, err error) {
+	ds := &Downsampler{
+		c:      c,
+		probes: probes.Listen(c.ProbePort),
+		stop:   make(chan struct{}),
+		stats: stats.New(
+			statReceived,
+			statSent,
+			statInvalidLines,
+			statInvalidTimestamps,
+			statFailedNATSPublish),
+		statsNATSDropped: stats.NewAnon(len(c.NATSSubject)),
+	}
+	defer func() {
+		if err != nil {
+			ds.Stop()
+		}
+	}()
+
+	ds.nc, err = nats.Connect(c.NATSAddress, nats.MaxReconnects(-1))
+	if err != nil {
+		return nil, fmt.Errorf("NATS Error: can't connect: %v", err)
+	}
+
+	maxPendingSize := int(c.NATSMaxPendingSize.Bytes())
+	for _, subject := range c.NATSSubject {
+		inputCh := make(chan []byte, 1024)
+		ds.wg.Add(1)
+		go ds.worker(subject, inputCh)
+		subscription, err := ds.nc.Subscribe(subject, func(msg *nats.Msg) {
+			inputCh <- msg.Data
+		})
+		if err != nil {
+			return nil, fmt.Errorf("NATS: subscription for %q failed: %v", subject, err)
+		}
+		if err := subscription.SetPendingLimits(-1, maxPendingSize); err != nil {
+			return nil, fmt.Errorf("NATS: failed to set pending limits: %v", err)
+		}
+		ds.subscriptions = append(ds.subscriptions, subscription)
+	}
+
+	// Subscriptions don't seem to be reliable without flushing after
+	// subscribing.
+	if err := ds.nc.Flush(); err != nil {
+		return nil, fmt.Errorf("NATS flush error: %v", err)
+	}
+
+	ds.wg.Add(1)
+	go ds.startStatistician()
+
+	log.Printf("downsampler subscribed to %v at %s", c.NATSSubject, c.NATSAddress)
+	log.Printf("downsampler period: %s", c.DownsamplePeriod.Duration)
+	log.Printf("downsampler output suffix: %s", c.DownsampleSuffix)
+	log.Printf("maximum NATS subject size: %s", c.NATSMaxPendingSize)
+
+	ds.probes.SetReady(true)
+
+	return ds, nil
+}
+
+// Stop aborts all goroutines belonging to the Downsampler and closes its
+// connection to NATS. It will be block until all Downsampler goroutines
+// have stopped.
+func (ds *Downsampler) Stop() {
+	// Indicate that the downsampler is shutting down.
+	ds.probes.SetReady(false)
+	ds.probes.SetAlive(false)
+
+	// Signal for goroutines to stop and wait for them.
+	close(ds.stop)
+	ds.wg.Wait()
+
+	// Clean up NATS subscriptions and close NATS connection.
+	for _, subscription := range ds.subscriptions {
+		subscription.Unsubscribe()
+	}
+	if ds.nc != nil {
+		ds.nc.Close()
+	}
+
+	// Shut down probes listener.
+	ds.probes.Close()
+}
+
+func (ds *Downsampler) worker(subject string, inputCh <-chan []byte) {
+	defer ds.wg.Done()
+
+	outSubject := subject + ds.c.DownsampleSuffix
+	assigner := newAssigner(ds.c.DownsamplePeriod.Duration, newAvgBucket, new(realClock))
+	for {
+		select {
+		case lines := <-inputCh:
+			ds.stats.Inc(statReceived)
+			errs := assigner.Update(lines)
+			for _, err := range errs {
+				if ds.c.Debug {
+					log.Println(err)
+				}
+				if isTimestampError(err) {
+					ds.stats.Inc(statInvalidTimestamps)
+				} else {
+					ds.stats.Inc(statInvalidLines)
+				}
+			}
+		case <-time.After(assigner.UntilNext()):
+		case <-ds.stop:
+			return
+		}
+
+		buf := assigner.Bytes()
+		if len(buf) > 0 {
+			if ds.c.Debug {
+				log.Printf("publishing to %s (%d bytes)", outSubject, len(buf))
+			}
+
+			splitter := newBatchSplitter(buf, maxNATSMsgSize)
+			for splitter.Next() {
+				if err := ds.nc.Publish(outSubject, splitter.Chunk()); err != nil {
+					log.Printf("publish error for %s: %v", outSubject, err)
+					ds.stats.Inc(statFailedNATSPublish)
+				}
+			}
+			ds.stats.Inc(statSent)
+		}
+	}
+}
+
+// startStatistician defines a goroutine that is responsible for
+// regularly sending the downsamplers's statistics to the monitoring
+// backend.
+func (ds *Downsampler) startStatistician() {
+	defer ds.wg.Done()
+
+	labels := stats.NewLabels("downsampler", ds.c.Name)
+	for {
+		now := time.Now()
+		snap := ds.stats.Snapshot()
+		natsDrops := ds.updateNATSDropped()
+
+		// Publish general stats.
+		lines := stats.SnapshotToPrometheus(snap, now, labels)
+		ds.nc.Publish(ds.c.NATSSubjectMonitor, lines)
+
+		// Publish the per subject nats_dropped metric.
+		for i, subject := range ds.c.NATSSubject {
+			ds.nc.Publish(ds.c.NATSSubjectMonitor, stats.CounterToPrometheus(
+				statNATSDropped,
+				int(natsDrops[i]),
+				now,
+				labels.With("subject", subject),
+			))
+		}
+
+		select {
+		case <-time.After(ds.c.StatsInterval.Duration):
+		case <-ds.stop:
+			return
+		}
+	}
+}
+
+func (ds *Downsampler) updateNATSDropped() []uint64 {
+	for i, sub := range ds.subscriptions {
+		dropped, err := sub.Dropped()
+		if err != nil {
+			log.Printf("NATS: failed to read drops for %q subscription: %v", sub.Subject, err)
+		} else {
+			ds.statsNATSDropped.Max(i, uint64(dropped))
+		}
+	}
+	return ds.statsNATSDropped.Snapshot()
+}
diff --git a/prometheus/labels.go b/prometheus/labels.go
index a73c1b1..94ee161 100644
--- a/prometheus/labels.go
+++ b/prometheus/labels.go
@@ -26,6 +26,15 @@ type Label struct {
 	Value []byte
 }
 
+// NewLabel provides a convenient ways of constructing a Label from
+// strings.
+func NewLabel(name, value string) Label {
+	return Label{
+		Name:  []byte(name),
+		Value: []byte(value),
+	}
+}
+
 // Labels contains the set of Label instances for a metric.
 type Labels []Label
 


Test patch:
diff --git a/cmd/influx-spout/e2e_large_test.go b/cmd/influx-spout/e2e_large_test.go
index 9063523..5adcbf1 100644
--- a/cmd/influx-spout/e2e_large_test.go
+++ b/cmd/influx-spout/e2e_large_test.go
@@ -38,24 +38,15 @@ import (
 )
 
 const (
-	natsPort    = 44600
-	influxdPort = 44601
-
-	listenerPort      = 44610
-	listenerProbePort = 44611
-
-	httpListenerPort      = 44620
-	httpListenerProbePort = 44621
-
-	filterProbePort = 44631
-
-	writerProbePort = 44641
-
-	monitorPort      = 44650
-	monitorProbePort = 44651
-
-	influxDBName = "test"
-	sendCount    = 10
+	natsPort         = 44600
+	influxdPort      = 44601
+	listenerPort     = 44602
+	httpListenerPort = 44603
+	monitorPort      = 44604
+
+	dbName        = "test"
+	archiveDBName = "test-archive"
+	sendCount     = 10
 )
 
 func TestEndToEnd(t *testing.T) {
@@ -70,29 +61,29 @@ func TestEndToEnd(t *testing.T) {
 	defer influxd.Stop()
 
 	// Use a fake filesystem (for config files).
-	fs := afero.NewMemMapFs()
-	config.Fs = fs
+	config.Fs = afero.NewMemMapFs()
 
 	// Start spout components.
-	listener := startListener(t, fs)
+	listener := startListener(t)
 	defer listener.Stop()
-	spouttest.AssertReadyProbe(t, listenerProbePort)
 
-	httpListener := startHTTPListener(t, fs)
+	httpListener := startHTTPListener(t)
 	defer httpListener.Stop()
-	spouttest.AssertReadyProbe(t, httpListenerProbePort)
 
-	filter := startFilter(t, fs)
+	filter := startFilter(t)
 	defer filter.Stop()
-	spouttest.AssertReadyProbe(t, filterProbePort)
 
-	writer := startWriter(t, fs)
+	downsampler := startDownsampler(t)
+	defer downsampler.Stop()
+
+	writer := startWriter(t)
 	defer writer.Stop()
-	spouttest.AssertReadyProbe(t, writerProbePort)
 
-	monitor := startMonitor(t, fs)
+	archiveWriter := startArchiveWriter(t)
+	defer archiveWriter.Stop()
+
+	monitor := startMonitor(t)
 	defer monitor.Stop()
-	spouttest.AssertReadyProbe(t, monitorProbePort)
 
 	// Connect to the listener.
 	addr := net.JoinHostPort("localhost", strconv.Itoa(listenerPort))
@@ -116,49 +107,40 @@ func TestEndToEnd(t *testing.T) {
 		require.NoError(t, err)
 	}
 
-	// Check "database".
-	maxWaitTime := time.Now().Add(spouttest.LongWait)
-	for {
-		lines := influxd.Lines()
-		recvCount := len(lines[influxDBName])
-		if recvCount == sendCount {
-			// Expected number of lines received...
-			// Now check they are correct.
-			for _, line := range lines[influxDBName] {
-				if !strings.HasPrefix(line, cpuLine) {
-					t.Fatalf("unexpected line received: %s", line)
-				}
-			}
-
-			// No writes to other databases are expected.
-			assert.Len(t, lines, 1)
-
-			break // Success
-		}
-		if time.Now().After(maxWaitTime) {
-			t.Fatalf("failed to see expected database records. Saw %d records.", recvCount)
-		}
-		time.Sleep(250 * time.Millisecond)
-	}
+	// Check "databases".
+	checkDatabase(t, influxd, dbName, sendCount, isCPULine)
+	checkDatabase(t, influxd, archiveDBName, 1, isLikeCPULine)
+	assert.Equal(t, 2, influxd.DatabaseCount()) // primary + archive
 
 	// Check metrics published by monitor component.
 	expectedMetrics := regexp.MustCompile(`
+failed_nats_publish{component="downsampler",host="h",name="downsampler"} 0
 failed_nats_publish{component="filter",host="h",name="filter"} 0
 failed_nats_publish{component="listener",host="h",name="listener"} 0
 failed_writes{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 0
+failed_writes{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test-archive",influxdb_port="44601",name="archive-writer"} 0
+invalid_lines{component="downsampler",host="h",name="downsampler"} 0
+invalid_timestamps{component="downsampler",host="h",name="downsampler"} 0
 invalid_time{component="filter",host="h",name="filter"} 0
 max_pending{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} \d+
+max_pending{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test-archive",influxdb_port="44601",name="archive-writer"} \d+
+nats_dropped{component="downsampler",host="h",name="downsampler",subject="system"} 0
 nats_dropped{component="filter",host="h",name="filter"} 0
 nats_dropped{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer",subject="system"} 0
+nats_dropped{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test-archive",influxdb_port="44601",name="archive-writer",subject="system-archive"} 0
 passed{component="filter",host="h",name="filter"} 10
 processed{component="filter",host="h",name="filter"} 20
 read_errors{component="listener",host="h",name="listener"} 0
+received{component="downsampler",host="h",name="downsampler"} 2
 received{component="listener",host="h",name="listener"} 5
 received{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 2
+received{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test-archive",influxdb_port="44601",name="archive-writer"} 1
 rejected{component="filter",host="h",name="filter"} 10
+sent{component="downsampler",host="h",name="downsampler"} 1
 sent{component="listener",host="h",name="listener"} 1
 triggered{component="filter",host="h",name="filter",rule="system"} 10
 write_requests{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test",influxdb_port="44601",name="writer"} 2
+write_requests{component="writer",host="h",influxdb_address="localhost",influxdb_dbname="test-archive",influxdb_port="44601",name="archive-writer"} 1
 $`[1:])
 	var lines string
 	for try := 0; try < 20; try++ {
@@ -178,7 +160,8 @@ $`[1:])
 	t.Fatalf("Failed to see expected metrics. Last saw:\n%s", lines)
 }
 
-const cpuLine = "cpu,cls=server,env=prod user=13.33,usage_system=0.16,usage_idle=86.53"
+const cpuLineHeader = "cpu,cls=server,env=prod "
+const cpuLine = cpuLineHeader + "user=13.33,usage_system=0.16,usage_idle=86.53"
 
 func makeTestLines() *bytes.Buffer {
 	now := time.Now().UnixNano()
@@ -193,75 +176,137 @@ foo,env=dev bar=99 %d
 	return out
 }
 
-func startListener(t *testing.T, fs afero.Fs) stoppable {
-	return startComponent(t, fs, "listener", fmt.Sprintf(`
+func startListener(t *testing.T) stoppable {
+	return startComponent(t, "listener", fmt.Sprintf(`
 mode = "listener"
 port = %d
 nats_address = "nats://localhost:%d"
 batch_max_count = 5
 debug = true
 nats_subject_monitor = "monitor"
-probe_port = %d
-`, listenerPort, natsPort, listenerProbePort))
+`, listenerPort, natsPort))
 }
 
-func startHTTPListener(t *testing.T, fs afero.Fs) stoppable {
-	return startComponent(t, fs, "listener", fmt.Sprintf(`
+func startHTTPListener(t *testing.T) stoppable {
+	return startComponent(t, "listener", fmt.Sprintf(`
 mode = "listener_http"
 port = %d
 nats_address = "nats://localhost:%d"
 batch_max_count = 5
 debug = true
 nats_subject_monitor = "monitor"
-probe_port = %d
-`, httpListenerPort, natsPort, httpListenerProbePort))
+`, httpListenerPort, natsPort))
 }
 
-func startFilter(t *testing.T, fs afero.Fs) stoppable {
-	return startComponent(t, fs, "filter", fmt.Sprintf(`
+func startFilter(t *testing.T) stoppable {
+	return startComponent(t, "filter", fmt.Sprintf(`
 mode = "filter"
 nats_address = "nats://localhost:%d"
 debug = true
 nats_subject_monitor = "monitor"
-probe_port = %d
 
 [[rule]]
 type = "basic"
 match = "cpu"
 subject = "system"
-`, natsPort, filterProbePort))
+`, natsPort))
 }
 
-func startWriter(t *testing.T, fs afero.Fs) stoppable {
-	return startComponent(t, fs, "writer", fmt.Sprintf(`
-mode = "writer"
+func startDownsampler(t *testing.T) stoppable {
+	return startComponent(t, "downsampler", fmt.Sprintf(`
+mode = "downsampler"
 nats_address = "nats://localhost:%d"
+debug = true
+nats_subject_monitor = "monitor"
+
 nats_subject = ["system"]
+downsample_period = "3s"
+`, natsPort))
+}
+
+func startWriter(t *testing.T) stoppable {
+	return baseStartWriter(t, "writer", "system", dbName)
+}
+
+func startArchiveWriter(t *testing.T) stoppable {
+	return baseStartWriter(t, "archive-writer", "system-archive", archiveDBName)
+}
+
+func baseStartWriter(t *testing.T, name, subject, dbName string) stoppable {
+	return startComponent(t, name, fmt.Sprintf(`
+mode = "writer"
+name = "%s"
+nats_address = "nats://localhost:%d"
+nats_subject = ["%s"]
 influxdb_port = %d
 influxdb_dbname = "%s"
 batch_max_count = 1
 workers = 4
 debug = true
 nats_subject_monitor = "monitor"
-probe_port = %d
-`, natsPort, influxdPort, influxDBName, writerProbePort))
+`, name, natsPort, subject, influxdPort, dbName))
 }
 
-func startMonitor(t *testing.T, fs afero.Fs) stoppable {
-	return startComponent(t, fs, "monitor", fmt.Sprintf(`
+func startMonitor(t *testing.T) stoppable {
+	return startComponent(t, "monitor", fmt.Sprintf(`
 mode = "monitor"
 nats_address = "nats://localhost:%d"
 nats_subject_monitor = "monitor"
 port = %d
-probe_port = %d
-`, natsPort, monitorPort, monitorProbePort))
+`, natsPort, monitorPort))
 }
 
-func startComponent(t *testing.T, fs afero.Fs, name, config string) stoppable {
+func startComponent(t *testing.T, name, configText string) stoppable {
+	probePort := getProbePort()
+	configText = fmt.Sprintf("probe_port = %d\n%s", probePort, configText)
+
 	configFilename := name + ".toml"
-	err := afero.WriteFile(fs, configFilename, []byte(config), 0600)
+	err := afero.WriteFile(config.Fs, configFilename, []byte(configText), 0600)
 	require.NoError(t, err)
 	s, err := runComponent(configFilename)
 	require.NoError(t, err)
+
+	if !spouttest.CheckReadyProbe(probePort) {
+		s.Stop()
+		t.Fatalf("startup probe for %s failed", name)
+	}
+
 	return s
 }
+
+func checkDatabase(t *testing.T, db *spouttest.FakeInfluxDB, dbName string, expectedCount int, checkLine func(string) bool) {
+	maxWaitTime := time.Now().Add(spouttest.LongWait)
+	for {
+		lines := db.Lines(dbName)
+		recvCount := len(lines)
+		if recvCount == expectedCount {
+			// Expected number of lines received. Now check they are correct.
+			for _, line := range lines {
+				if !checkLine(line) {
+					t.Fatalf("unexpected line received: %s", line)
+				}
+			}
+			break // Success
+		}
+		if time.Now().After(maxWaitTime) {
+			t.Fatalf("failed to see expected database records. Saw %d records.", recvCount)
+		}
+		time.Sleep(250 * time.Millisecond)
+	}
+}
+
+func isCPULine(line string) bool {
+	return strings.HasPrefix(line, cpuLine)
+}
+
+func isLikeCPULine(line string) bool {
+	return strings.HasPrefix(line, cpuLineHeader)
+}
+
+var nextProbePort = 44620
+
+func getProbePort() int {
+	out := nextProbePort
+	nextProbePort++
+	return out
+}
diff --git a/config/config_small_test.go b/config/config_small_test.go
index 48bd849..8fdb0e0 100644
--- a/config/config_small_test.go
+++ b/config/config_small_test.go
@@ -109,6 +109,8 @@ func TestAllDefaults(t *testing.T) {
 	assert.Equal(t, "writer", conf.Mode)
 	assert.Equal(t, 8, conf.Workers)
 	assert.Equal(t, 30*time.Second, conf.WriteTimeout.Duration)
+	assert.Equal(t, time.Duration(0), conf.DownsamplePeriod.Duration)
+	assert.Equal(t, "-archive", conf.DownsampleSuffix)
 	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)
@@ -149,6 +151,12 @@ func TestDefaultHTTPListenerBatchSize(t *testing.T) {
 	assert.Equal(t, 1*datasize.MB, conf.BatchMaxSize)
 }
 
+func TestDefaultDownsamplePeriod(t *testing.T) {
+	conf, err := parseConfig(`mode = "downsampler"`)
+	require.NoError(t, err)
+	assert.Equal(t, time.Minute, conf.DownsamplePeriod.Duration)
+}
+
 func TestNoMode(t *testing.T) {
 	_, err := parseConfig("")
 	assert.EqualError(t, err, "mode not specified in config")
@@ -297,80 +305,81 @@ func TestValidateProblems(t *testing.T) {
 	Fs = afero.NewMemMapFs()
 
 	tests := []struct {
-		config string
 		err    string
+		config string
 	}{{
+		"invalid mode: frog",
 		`
 mode = "frog"
 `,
-		"invalid mode: frog",
 	}, {
+		"listener port out of range",
 		`
 mode = "listener"
 port = -1
 `,
-		"listener port out of range",
 	}, {
+		"listener port out of range",
 		`
 mode = "listener"
 port = 65536
 `,
-		"listener port out of range",
 	}, {
+		"listener port out of range",
 		`
 mode = "listener_http"
 port = -1
 `,
-		"listener port out of range",
 	}, {
+		"listener port out of range",
 		`
 mode = "listener_http"
 port = 65536
 `,
-		"listener port out of range",
 	}, {
+		"probe_port out of range",
 		`
 mode = "filter"
 probe_port = 65536
 `,
-		"probe_port out of range",
 	}, {
+		"probe_port out of range",
 		`
 mode = "writer"
 probe_port = -1
 `,
-		"probe_port out of range",
 	}, {
+		"pprof_port out of range",
 		`
 mode = "filter"
 pprof_port = 65536
 `,
-		"pprof_port out of range",
 	}, {
+		"pprof_port out of range",
 		`
 mode = "writer"
 pprof_port = -1
 `,
-		"pprof_port out of range",
 	}, {
+		"listener should only use one NATS subject",
 		`
 mode = "listener"
 nats_subject = ["one", "two"]
 `,
-		"listener should only use one NATS subject",
 	}, {
+		"listener should only use one NATS subject",
 		`
 mode = "listener_http"
 nats_subject = ["one", "two"]
 `,
-		"listener should only use one NATS subject",
 	}, {
+		"filter should only use one NATS subject",
 		`
 mode = "filter"
 nats_subject = ["one", "two"]
 `,
-		"filter should only use one NATS subject",
 	}, {
+		`rule missing "type"`,
 		`
 mode = "filter"
 
@@ -378,8 +387,8 @@ mode = "filter"
 match = "needle"
 subject = "foo"
 `,
-		`rule missing "type"`,
 	}, {
+		`rule missing "match"`,
 		`
 mode = "filter"
 
@@ -387,8 +396,8 @@ mode = "filter"
 type = "basic"
 subject = "foo"
 `,
-		`rule missing "match"`,
 	}, {
+		`rule missing "subject"`,
 		`
 mode = "filter"
 
@@ -396,8 +405,8 @@ mode = "filter"
 type = "basic"
 match = "needle"
 `,
-		`rule missing "subject"`,
 	}, {
+		`basic rule cannot use "tags"`,
 		`
 mode = "filter"
 
@@ -407,8 +416,8 @@ match = "something"
 subject = "out"
 tags = [["abc", "def"]]
 `,
-		`basic rule cannot use "tags"`,
 	}, {
+		`tags rule cannot use "match"`,
 		`
 mode = "filter"
 
@@ -417,8 +426,8 @@ type = "tags"
 subject = "out"
 match = "something"
 `,
-		`tags rule cannot use "match"`,
 	}, {
+		`tags rule missing "tags"`,
 		`
 mode = "filter"
 
@@ -426,8 +435,8 @@ mode = "filter"
 type = "tags"
 subject = "out"
 `,
-		`tags rule missing "tags"`,
 	}, {
+		`each tags item must contain 2 values`,
 		`
 mode = "filter"
 
@@ -436,42 +445,42 @@ type = "tags"
 tags = [["abc", "def", "zzz"]]
 subject = "out"
 `,
-		`each tags item must contain 2 values`,
 	}, {
+		"writer needs at least one NATS subject",
 		`
 mode = "writer"
 nats_subject = []
 `,
-		"writer needs at least one NATS subject",
 	}, {
+		"influxdb_port out of range",
 		`
 mode = "writer"
 influxdb_port = 65536
 `,
-		"influxdb_port out of range",
 	}, {
+		"influxdb_port out of range",
 		`
 mode = "writer"
 influxdb_port = 0
 `,
-		"influxdb_port out of range",
 	}, {
+		`rule missing "type"`,
 		`
 mode = "writer"
 
 [[rule]]
 match = "needle"
 `,
-		`rule missing "type"`,
 	}, {
+		`rule missing "match"`,
 		`
 mode = "writer"
 
 [[rule]]
 type = "basic"
 `,
-		`rule missing "match"`,
 	}, {
+		`writer rules shouldn't have subject ("matches")`,
 		`
 mode = "writer"
 
@@ -480,7 +489,27 @@ type = "basic"
 match = "needle"
 subject = "matches"
 `,
-		`writer rules shouldn't have subject ("matches")`,
+	}, {
+		`downsample_period is only for downsampler mode`,
+		`
+mode = "filter"
+downsample_period = "30s"
+`,
+	}, {
+		`downsample_suffix must be set`,
+		`
+mode = "downsampler"
+downsample_period = "30s"
+downsample_suffix = ""
+`,
+	}, {
+		`downsample_suffix must be set`,
+		`
+mode = "downsampler"
+
+downsample_period = "30s"
+downsample_suffix = ""
+`,
 	},
 	}
 
diff --git a/downsampler/assigner_small_test.go b/downsampler/assigner_small_test.go
new file mode 100644
index 0000000..01c1ef2
--- /dev/null
+++ b/downsampler/assigner_small_test.go
@@ -0,0 +1,349 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build small
+
+package downsampler
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"math/rand"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAssignerEmpty(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+	require.Nil(t, a.Bytes())
+
+	// Advance time and see that nothing is ever emitted.
+	for i := 0; i < 10; i++ {
+		clock.Advance(time.Minute)
+		require.Nil(t, a.Bytes())
+	}
+}
+
+func TestTooOld(t *testing.T) {
+	clock := newFakeClock() // clock starts right on bucket boundary (Tp)
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	ts := clock.now.Add(-61 * time.Second)
+	errs := a.Update(makeLine("x", ts))
+	assertSingleError(t, errs, "timestamp too old: "+fmt.Sprint(ts))
+}
+
+func TestPrevSlot(t *testing.T) {
+	clock := newFakeClock() // clock starts right on bucket boundary (Tp)
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// Put some values in the "prev" slot.
+	require.Nil(t, a.Update(makeLine("a", clock.now.Add(-30*time.Second))))
+	require.Nil(t, a.Update(makeLine("b", clock.now.Add(-time.Minute))))
+	require.Nil(t, a.Update(makeLine("c", clock.now.Add(-time.Second))))
+
+	// Ensure nothing being emitted yet (time hasn't advanced).
+	assert.Nil(t, a.Bytes())
+
+	// Advance time to just before the bucket boundary. Nothing should
+	// be emitted.
+	clock.Advance(59 * time.Second)
+	assert.Nil(t, a.Bytes())
+
+	// Now advance to the next boundary and see that the correct data
+	// was emitted.
+	clock.Advance(time.Second)
+	assert.Equal(t, "a|b|c", string(a.Bytes()))
+
+	// Try again, and now nothing should be emitted.
+	assert.Nil(t, a.Bytes())
+}
+
+func TestCurrSlot(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// Put some lines in the "curr" slot.
+	require.Nil(t, a.Update(makeLine("a", clock.now)))
+	require.Nil(t, a.Update(makeLine("b", clock.now.Add(30*time.Second))))
+	require.Nil(t, a.Update(makeLine("c", clock.now.Add(59*time.Second))))
+
+	// Ensure nothing being emitted yet (time hasn't advanced).
+	assert.Nil(t, a.Bytes())
+
+	// Advance time to Tc. This should cause the "prev" slot to be
+	// emitted (but it should be empty).
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+
+	// Advance to just before the next boundary.
+	clock.Advance(59 * time.Second)
+	assert.Nil(t, a.Bytes())
+
+	// Now advance to the next boundary which should cause the lines
+	// to be emitted.
+	clock.Advance(time.Second)
+	assert.Equal(t, "a|b|c", string(a.Bytes()))
+
+	// Try again, and now nothing should be emitted.
+	assert.Nil(t, a.Bytes())
+}
+
+func TestNextSlot(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// Put some lines in the "next" slot.
+	tc := clock.now.Add(time.Minute)
+	require.Nil(t, a.Update(makeLine("a", tc)))
+	require.Nil(t, a.Update(makeLine("b", tc.Add(30*time.Second))))
+	require.Nil(t, a.Update(makeLine("c", tc.Add(59*time.Second))))
+
+	// Ensure nothing being emitted yet (time hasn't advanced).
+	assert.Nil(t, a.Bytes())
+
+	// Advance time to Tc. This should cause the "prev" slot to be
+	// emitted (but it should be empty).
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+
+	// Advance to next again. This should cause the original "curr"
+	// slot to be emitted (but it should be empty).
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+
+	// Advance to just before the next boundary.
+	clock.Advance(59 * time.Second)
+	assert.Nil(t, a.Bytes())
+
+	// Now advance to the next boundary which should cause the lines
+	// to be emitted.
+	clock.Advance(time.Second)
+	assert.Equal(t, "a|b|c", string(a.Bytes()))
+
+	// Try again, and now nothing should be emitted.
+	assert.Nil(t, a.Bytes())
+}
+
+func TestTooNew(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	ts := clock.now.Add(2 * time.Minute)
+	errs := a.Update(makeLine("x", ts))
+	assertSingleError(t, errs, "timestamp too new: "+fmt.Sprint(ts))
+}
+
+func TestAllSlots(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// Put one line in each slot.
+	require.Nil(t, a.Update(makeLine("a", clock.now.Add(-time.Second))))
+	require.Nil(t, a.Update(makeLine("b", clock.now)))
+	require.Nil(t, a.Update(makeLine("c", clock.now.Add(time.Minute))))
+
+	// Ensure nothing being emitted yet (time hasn't advanced).
+	assert.Nil(t, a.Bytes())
+
+	// Advance a period. First slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "a", string(a.Bytes()))
+
+	// Advance a period. Second slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "b", string(a.Bytes()))
+
+	// Advance a period. Third slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "c", string(a.Bytes()))
+
+	// Advance again. Nothing left to emit.
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+}
+
+func TestMultipleInputLines(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// Put two lines in each slot but with a single Update call.
+	var lines []byte
+	lines = append(lines, makeLine("a0", clock.now.Add(-time.Second))...)
+	lines = append(lines, makeLine("a1", clock.now.Add(-20*time.Second))...)
+	lines = append(lines, makeLine("b0", clock.now)...)
+	lines = append(lines, makeLine("b1", clock.now.Add(10*time.Second))...)
+	lines = append(lines, makeLine("c0", clock.now.Add(60*time.Second))...)
+	lines = append(lines, makeLine("c1", clock.now.Add(90*time.Second))...)
+	require.Nil(t, a.Update(lines))
+
+	// Advance a period. First slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "a0|a1", string(a.Bytes()))
+
+	// Advance a period. Second slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "b0|b1", string(a.Bytes()))
+
+	// Advance a period. Third slot should be emitted.
+	clock.Advance(time.Minute)
+	assert.Equal(t, "c0|c1", string(a.Bytes()))
+
+	// Advance again. Nothing left to emit.
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+}
+
+func TestNoTimestamp(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	// The current time is assumed if there's no timestamp on the line.
+	require.Nil(t, a.Update([]byte{'x'}))
+
+	clock.Advance(time.Minute)
+	assert.Nil(t, a.Bytes())
+
+	clock.Advance(time.Minute)
+	assert.Equal(t, "x", string(a.Bytes()))
+}
+
+func TestBucketErrors(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFailBucket, clock)
+
+	errs := a.Update(makeLine("x", clock.now))
+	assertSingleError(t, errs, "boom")
+
+	errs = a.Update(makeLine("y", clock.now))
+	assertSingleError(t, errs, "boom")
+}
+
+func TestUntilNext(t *testing.T) {
+	clock := newFakeClock()
+	a := newAssigner(time.Minute, newFakeBucket, clock)
+
+	assert.Equal(t, time.Minute, a.UntilNext())
+
+	clock.Advance(30 * time.Second)
+	assert.Equal(t, 30*time.Second, a.UntilNext())
+
+	clock.Advance(29 * time.Second)
+	assert.Equal(t, time.Second, a.UntilNext())
+
+	clock.Advance(time.Second)
+	assert.Equal(t, time.Duration(0), a.UntilNext())
+	a.Bytes()
+	assert.Equal(t, time.Minute, a.UntilNext())
+
+	clock.Advance(time.Second)
+	assert.Equal(t, 59*time.Second, a.UntilNext())
+}
+
+func BenchmarkAssigner(b *testing.B) {
+	a := newAssigner(time.Second, newFakeBucket, new(realClock))
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		offset := time.Duration(rand.Intn(2000)-1000) * time.Millisecond
+		_ = a.Update(makeLine("a", time.Now().Add(offset)))
+		_ = a.Bytes()
+	}
+}
+
+func newFakeBucket(t time.Time) bucket {
+	return &fakeBucket{endTime: t}
+}
+
+// fakeBucket keeps the first line of each input to Update. Its Bytes
+// method returns those words joined by the pipe symbol.
+type fakeBucket struct {
+	endTime time.Time
+	seen    [][]byte
+}
+
+func (b *fakeBucket) EndTime() time.Time {
+	return b.endTime
+}
+
+func (b *fakeBucket) AddLine(line []byte) []error {
+	parts := bytes.SplitN(line, []byte{'\n'}, 2)
+	b.seen = append(b.seen, parts[0])
+	return nil
+}
+
+func (b *fakeBucket) Bytes() []byte {
+	if len(b.seen) == 0 {
+		return nil
+	}
+	return bytes.Join(b.seen, []byte{'|'})
+}
+
+func newFailBucket(t time.Time) bucket {
+	return &failBucket{endTime: t}
+}
+
+type failBucket struct {
+	endTime time.Time
+}
+
+func (b *failBucket) EndTime() time.Time {
+	return b.endTime
+}
+
+func (b *failBucket) AddLine(line []byte) []error {
+	return []error{errors.New("boom")}
+}
+
+func (b *failBucket) Bytes() []byte {
+	return nil
+}
+
+func newFakeClock() *fakeClock {
+	return &fakeClock{
+		now: time.Date(2018, 8, 28, 14, 11, 0, 0, time.Local),
+	}
+}
+
+// fakeClock conforms to the clock interface, providing a clock that
+// can be manually advanced.
+type fakeClock struct {
+	now time.Time
+}
+
+func (c *fakeClock) Now() time.Time {
+	return c.now
+}
+
+func (c *fakeClock) Advance(d time.Duration) {
+	c.now = c.now.Add(d)
+}
+
+func makeLine(text string, t time.Time) []byte {
+	out := []byte(text)
+	out = append(out, ' ')
+	out = append(out, timestamp(t)...)
+	out = append(out, '\n')
+	return out
+}
+
+func timestamp(t time.Time) []byte {
+	return []byte(strconv.FormatInt(t.UnixNano(), 10))
+}
diff --git a/downsampler/avgbucket_small_test.go b/downsampler/avgbucket_small_test.go
new file mode 100644
index 0000000..f523623
--- /dev/null
+++ b/downsampler/avgbucket_small_test.go
@@ -0,0 +1,266 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build small
+
+package downsampler
+
+import (
+	"bytes"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var now = time.Now()
+var nowNano = strconv.FormatInt(now.UnixNano(), 10)
+
+func TestFloatField(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=1.0`)))
+	assertBytes(t, "foo,host=nyc01 bar=1", b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=2.1`)))
+	assertBytes(t, "foo,host=nyc01 bar=1.55", b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=3.2`)))
+	assertBytes(t, "foo,host=nyc01 bar=2.1", b.Bytes())
+}
+
+func TestScientificFloatField(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 sci=-1.24e+78`)))
+	assertBytes(t, "foo,host=nyc01 sci=-1.24e+78", b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 sci=-1.30e+78`)))
+	assertBytes(t, "foo,host=nyc01 sci=-1.27e+78", b.Bytes())
+}
+
+func TestFloatTypeChange(t *testing.T) {
+	b := newAvgBucket(now)
+
+	// Send a float field.
+	require.Nil(t, b.AddLine([]byte(`foo,dc=nyc x=1.2`)))
+	assertBytes(t, `foo,dc=nyc x=1.2`, b.Bytes())
+
+	// Now send it as a string, change should be ignored.
+	errs := b.AddLine([]byte(`foo,dc=nyc x="foo"`))
+	assertSingleError(t, errs, `error parsing [ x="foo"]: wrong type for float: foo`)
+	assertBytes(t, `foo,dc=nyc x=1.2`, b.Bytes())
+}
+
+func TestIntField(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=1i`)))
+	assertBytes(t, "foo,host=nyc01 bar=1i", b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=3i`)))
+	assertBytes(t, "foo,host=nyc01 bar=2i", b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=3i`)))
+	assertBytes(t, "foo,host=nyc01 bar=2i", b.Bytes())
+}
+
+func TestIntTypeChange(t *testing.T) {
+	b := newAvgBucket(now)
+
+	// Send an integer field.
+	require.Nil(t, b.AddLine([]byte(`foo,dc=nyc x=42i`)))
+	assertBytes(t, `foo,dc=nyc x=42i`, b.Bytes())
+
+	// Now send it as a string, change should be ignored.
+	errs := b.AddLine([]byte(`foo,dc=nyc x="foo"`))
+	assertSingleError(t, errs, `error parsing [ x="foo"]: wrong type for int: foo`)
+	assertBytes(t, `foo,dc=nyc x=42i`, b.Bytes())
+}
+
+func TestStringField(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar="alpha"`)))
+	assertBytes(t, `foo,host=nyc01 bar="alpha"`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar="beta"`)))
+	assertBytes(t, `foo,host=nyc01 bar="beta"`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar="delta omega"`)))
+	assertBytes(t, `foo,host=nyc01 bar="delta omega"`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar="delta \"omega\""`)))
+	assertBytes(t, `foo,host=nyc01 bar="delta \"omega\""`, b.Bytes())
+}
+
+func TestUnterminatedStringField(t *testing.T) {
+	b := newAvgBucket(now)
+	errs := b.AddLine([]byte(`foo,host=nyc01 bar="delta omega`))
+
+	assertSingleError(t, errs, `error parsing [ bar="delta omega]: missing trailing double quote`)
+	assert.Len(t, b.Bytes(), 0) // line should be rejected
+}
+
+func TestOtherFieldValues(t *testing.T) {
+	// Field values which aren't floats, ints or strings are just
+	// passed through. This covers bools but also incorrectly
+	// represented values (best efforts).
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=false`)))
+	assertBytes(t, `foo,host=nyc01 bar=false`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=T`)))
+	assertBytes(t, `foo,host=nyc01 bar=T`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 bar=abcdef`)))
+	assertBytes(t, `foo,host=nyc01 bar=abcdef`, b.Bytes())
+}
+
+func TestMultipleFields(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 b=t,c=123i`)))
+	assertBytes(t, `foo,host=nyc01 b=t,c=123i`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 c=321i,b=f`)))
+	assertBytes(t, `foo,host=nyc01 b=f,c=222i`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,host=nyc01 c=444i,b=f,a=0.4`)))
+	assertBytes(t, `foo,host=nyc01 a=0.4,b=f,c=296i`, b.Bytes())
+}
+
+func TestMultipleTags(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo,dc=nyc,host=nyc01 b=t`)))
+	assertBytes(t, `foo,dc=nyc,host=nyc01 b=t`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo,dc=nyc,host=nyc01 b=f`)))
+	assertBytes(t, `foo,dc=nyc,host=nyc01 b=f`, b.Bytes())
+}
+
+func TestNoTags(t *testing.T) {
+	b := newAvgBucket(now)
+
+	require.Nil(t, b.AddLine([]byte(`foo b=t`)))
+	assertBytes(t, `foo b=t`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`foo b=false`)))
+	assertBytes(t, `foo b=false`, b.Bytes())
+}
+
+func TestMeasurementOnly(t *testing.T) {
+	b := newAvgBucket(now)
+
+	// Degenerate case, but still needs testing.
+	require.Nil(t, b.AddLine([]byte(`foo`)))
+	assertBytes(t, `foo`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`bar`)))
+	assertBytes(t, joinLines("foo", "bar"), b.Bytes())
+}
+
+func TestMeasurementAndTagsOnly(t *testing.T) {
+	b := newAvgBucket(now)
+
+	// Degenerate case, but still needs testing.
+	require.Nil(t, b.AddLine([]byte(`foo,dc=nyc`)))
+	assertBytes(t, `foo,dc=nyc`, b.Bytes())
+
+	require.Nil(t, b.AddLine([]byte(`bar,dc=sfo`)))
+	assertBytes(t, joinLines(
+		"bar,dc=sfo",
+		"foo,dc=nyc",
+	), b.Bytes())
+}
+
+func TestInvalidField(t *testing.T) {
+	b := newAvgBucket(now)
+	errs := b.AddLine([]byte(`foo x`))
+
+	assertSingleError(t, errs, `error parsing [ x]: invalid field`)
+	assertBytes(t, "", b.Bytes())
+}
+
+func TestMissingFieldValue(t *testing.T) {
+	b := newAvgBucket(now)
+	errs := b.AddLine([]byte(`foo x=`))
+
+	assertSingleError(t, errs, `error parsing [ x=]: missing field value`)
+	assertBytes(t, "", b.Bytes())
+}
+
+func TestMultipleErrors(t *testing.T) {
+	b := newAvgBucket(now)
+
+	errs := b.AddLine([]byte(`foo x=1i,y=2`))
+	require.Nil(t, errs)
+
+	errs = b.AddLine([]byte(`foo x=1,y=2i`))
+
+	require.Len(t, errs, 2)
+	assert.EqualError(t, errs[0], `error parsing [ x=1,y=2i]: wrong type for int: 1`)
+	assert.EqualError(t, errs[1], `error parsing [ x=1,y=2i]: wrong type for float: 2i`)
+
+	assertBytes(t, `foo x=1i,y=2`, b.Bytes())
+}
+
+func BenchmarkAvgBucket(b *testing.B) {
+	bucket := newAvgBucket(now)
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		bucket.AddLine([]byte(`X,host=nyc01 foo=11i bar=1.1 qaz="short"`))
+		bucket.AddLine([]byte(`X,host=nyc01 foo=22i bar=2.2 qaz="long"`))
+		_ = bucket.Bytes()
+	}
+}
+
+func assertBytes(t *testing.T, expected string, actual []byte) {
+	var expectedWithTs []string
+	for _, line := range strings.Split(expected, "\n") {
+		if len(line) > 0 {
+			expectedWithTs = append(expectedWithTs, appendNow(line))
+		}
+	}
+
+	var actualStr []string
+	for _, line := range bytes.Split(actual, []byte{'\n'}) {
+		if len(line) > 0 {
+			actualStr = append(actualStr, string(line))
+		}
+	}
+
+	assert.ElementsMatch(t, expectedWithTs, actualStr)
+}
+
+func appendNow(in string) string {
+	return in + " " + nowNano
+}
+
+func joinLines(in ...string) string {
+	return strings.Join(in, "\n")
+}
+
+func assertSingleError(t *testing.T, errs []error, expected string) {
+	assert.Len(t, errs, 1)
+	if len(errs) > 0 {
+		assert.EqualError(t, errs[0], expected)
+	}
+}
diff --git a/downsampler/batch_splitter_test.go b/downsampler/batch_splitter_test.go
new file mode 100644
index 0000000..8761ab2
--- /dev/null
+++ b/downsampler/batch_splitter_test.go
@@ -0,0 +1,139 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build small
+
+package downsampler
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEmpty(t *testing.T) {
+	b := []byte("")
+	splitter := newBatchSplitter(b, 100)
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestNoSplit(t *testing.T) {
+	b := []byte("abcdefghij\n")
+	splitter := newBatchSplitter(b, 100)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, b, splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestNoSplitWithLines(t *testing.T) {
+	b := []byte("abcd\nefg\nhij")
+	splitter := newBatchSplitter(b, 100)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, b, splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestNoSplitExact(t *testing.T) {
+	b := []byte("abcd\nefg\nhij")
+	splitter := newBatchSplitter(b, len(b))
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, b, splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestChunks(t *testing.T) {
+	b := []byte("1111\n2222\n333\n")
+	splitter := newBatchSplitter(b, 6)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("1111\n"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("2222\n"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("333\n"), splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestChunksExact(t *testing.T) {
+	b := []byte("1111\n2222\n3333\n")
+	splitter := newBatchSplitter(b, 5)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("1111\n"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("2222\n"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("3333\n"), splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestLineTooLong(t *testing.T) {
+	b := []byte("01234567")
+	splitter := newBatchSplitter(b, 3)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("012"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("345"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("67"), splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
+
+func TestMultiLineTooLong(t *testing.T) {
+	b := []byte("0123456\n88\n99")
+	splitter := newBatchSplitter(b, 3)
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("012"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("345"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("6\n"), splitter.Chunk())
+
+	// Test recovery.
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("88\n"), splitter.Chunk())
+
+	assert.True(t, splitter.Next())
+	assert.Equal(t, []byte("99"), splitter.Chunk())
+
+	assert.False(t, splitter.Next())
+	assert.Nil(t, splitter.Chunk())
+}
diff --git a/downsampler/downsampler_medium_test.go b/downsampler/downsampler_medium_test.go
new file mode 100644
index 0000000..2b3f2ae
--- /dev/null
+++ b/downsampler/downsampler_medium_test.go
@@ -0,0 +1,207 @@
+// Copyright 2018 Jump Trading
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build medium
+
+package downsampler
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/c2h5oh/datasize"
+	"github.com/nats-io/go-nats"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/jumptrading/influx-spout/config"
+	"github.com/jumptrading/influx-spout/influx"
+	"github.com/jumptrading/influx-spout/prometheus"
+	"github.com/jumptrading/influx-spout/spouttest"
+	"github.com/jumptrading/influx-spout/stats"
+)
+
+const natsPort = 44500
+const probePort = 44501
+
+func testConfig() *config.Config {
+	return &config.Config{
+		Mode:               "downsampler",
+		Name:               "mpc-60",
+		NATSAddress:        fmt.Sprintf("nats://127.0.0.1:%d", natsPort),
+		NATSSubject:        []string{"subject0", "subject1"},
+		NATSSubjectMonitor: "downsampler-test-monitor",
+		DownsamplePeriod:   config.Duration{Duration: 1 * time.Second},
+		DownsampleSuffix:   "-arch",
+		ProbePort:          probePort,
+		NATSMaxPendingSize: 32 * datasize.MB,
+		Debug:              true,
+
+		// Make the statistician report more often during tests. This
+		// makes the tests run faster.
+		StatsInterval: config.Duration{Duration: 250 * time.Millisecond},
+	}
+}
+
+func TestDownsampler(t *testing.T) {
+	stats.SetHostname("h")
+
+	gnatsd := spouttest.RunGnatsd(natsPort)
+	defer gnatsd.Shutdown()
+
+	conf := testConfig()
+
+	ds := startDownsampler(t, conf)
+	defer ds.Stop()
+
+	nc, err := nats.Connect(conf.NATSAddress)
+	require.NoError(t, err)
+	defer nc.Close()
+
+	// Subscribe to outputs
+	output0 := subscribeOutput(t, nc, "subject0-arch")
+	output1 := subscribeOutput(t, nc, "subject1-arch")
+	monitorCh := subscribeOutput(t, nc, conf.NATSSubjectMonitor)
+
+	// Send lines to each subject to be archived.
+	err = nc.Publish("subject0", []byte(`
+string,host=host0 value="foo bar"
+nums,host=host0 x=111i,y=1.5
+string,host=host0 value="bar foo"
+bool,host=0 b=true
+string,host=host0 value="foo bar"
+bool,host=1 b=false
+nums,host=host0 x=333i,y=2.5
+bool,host=0 b=false
+bool,host=1 b=true
+too_old_timestamp 123
+`[1:]))
+	require.NoError(t, err)
+
+	err = nc.Publish("subject1", []byte(`
+string,host=host0 value="one"
+int,host=host0 xyz=111i
+string,host=host0 value="two"
+string,host=host0 value="three"
+int,host=host0 xyz=333i
+too_new_timestamp 9223372036854775807
+invalid x
+`[1:]))
+	require.NoError(t, err)
+
+	// Check archived lines.
+	assertDownsamplerRecv(t, output0, "subject0-arch", `
+string,host=host0 value="foo bar"
+nums,host=host0 x=222i,y=2
+bool,host=0 b=false
+bool,host=1 b=true
+`)
+	assertDownsamplerRecv(t, output1, "subject1-arch", `
+string,host=host0 value="three"
+int,host=host0 xyz=222i
+`)
+
+	// Now wait a little and ensure that nothing else is emitted.
+	time.Sleep(time.Second)
+
+	spouttest.AssertNoMore(t, output0)
+	spouttest.AssertNoMore(t, output1)
+
+	// // Check the monitor output.
+
+	baseLabels := prometheus.Labels{
+		prometheus.NewLabel("component", "downsampler"),
+		prometheus.NewLabel("host", "h"),
+		prometheus.NewLabel("name", "mpc-60"),
+	}
+	baseLabelStr := string(baseLabels.ToBytes())
+	extendLabels := func(name, value string) string {
+		out := append(baseLabels, prometheus.NewLabel(name, value))
+		return string(out.ToBytes())
+	}
+	spouttest.AssertMonitor(t, monitorCh, []string{
+		`received` + baseLabelStr + ` 2`,
+		`sent` + baseLabelStr + ` 2`,
+		`invalid_lines` + baseLabelStr + ` 1`,
+		`invalid_timestamps` + baseLabelStr + ` 2`,
+		`failed_nats_publish` + baseLabelStr + ` 0`,
+		`nats_dropped` + extendLabels("subject", "subject0") + ` 0`,
+		`nats_dropped` + extendLabels("subject", "subject1") + ` 0`,
+	})
+}
+
+func startDownsampler(t *testing.T, conf *config.Config) *Downsampler {
+	ds, err := StartDownsampler(conf)
+	require.NoError(t, err)
+	if !spouttest.CheckReadyProbe(conf.ProbePort) {
+		ds.Stop()
+		t.Fatal("downsampler not ready")
+	}
+	return ds
+}
+
+func subscribeOutput(t *testing.T, nc *nats.Conn, chanName string) chan string {
+	output := make(chan string, 10)
+	_, err := nc.Subscribe(chanName, func(msg *nats.Msg) {
+		output <- string(msg.Data)
+	})
+	require.NoError(t, err)
+	return output
+}
+
+func assertDownsamplerRecv(t *testing.T, ch <-chan string, label, expected string) {
+	expectedLines := splitLines(expected)
+	select {
+	case received := <-ch:
+		lines := splitLines(received)
+		lines = assertTimestamps(t, lines)
+		assert.ElementsMatch(t, expectedLines, lines)
+	case <-time.After(spouttest.LongWait):
+		t.Fatalf("timed out waiting for %s", label)
+	}
+}
+
+// assertTimestamps checks that the timestamps on the input lines are
+// recent and the same. Returns the input lines with the timestamps
+// stripped.
+func assertTimestamps(t *testing.T, lines []string) []string {
+	outLines := make([]string, 0, len(lines))
+	seenTs := make(map[int64]bool)
+	for _, line := range lines {
+		ts, offset := influx.ExtractTimestamp([]byte(line))
+		require.True(t, offset >= 0) // Must have timestamp
+		outLines = append(outLines, line[:offset-1])
+		seenTs[ts] = true
+	}
+
+	assert.Len(t, seenTs, 1, "expected one timestamp")
+	for ts := range seenTs {
+		assert.True(t, time.Duration(time.Now().UnixNano()-ts) < (10*time.Second))
+	}
+
+	return outLines
+}
+
+func splitLines(raw string) []string {
+	rawLines := strings.Split(raw, "\n")
+	out := make([]string, 0, len(rawLines))
+	for _, line := range rawLines {
+		if len(line) > 0 {
+			out = append(out, line)
+		}
+	}
+	return out
+}
diff --git a/listener/listener_medium_test.go b/listener/listener_medium_test.go
index a292e82..899b08f 100644
--- a/listener/listener_medium_test.go
+++ b/listener/listener_medium_test.go
@@ -105,7 +105,7 @@ func TestBatching(t *testing.T) {
 
 	// Should receive a single batch.
 	assertBatch(t, listenerCh, strings.Join(poetry, ""))
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 
 	assertMonitor(t, monitorCh, numLines, 1)
 }
@@ -136,7 +136,7 @@ func TestWhatComesAroundGoesAround(t *testing.T) {
 	for i := 0; i < numLines; i++ {
 		assertBatch(t, listenerCh, poetry[i])
 	}
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 
 	assertMonitor(t, monitorCh, numLines, numLines)
 }
@@ -178,7 +178,7 @@ loop:
 		}
 	}
 
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 
 	// Ensure that batch was output because batch size limit was
 	// reached, not the message count.
@@ -211,7 +211,7 @@ func TestBatchAge(t *testing.T) {
 
 	// Line should be emitted after 1 sec.
 	assertBatch(t, listenerCh, line)
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 }
 
 func TestHTTPListener(t *testing.T) {
@@ -240,7 +240,7 @@ func TestHTTPListener(t *testing.T) {
 	for i := 0; i < numLines; i++ {
 		assertBatch(t, listenerCh, poetry[i])
 	}
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 
 	assertMonitor(t, monitorCh, numLines, numLines)
 }
@@ -278,7 +278,7 @@ func TestHTTPListenerBigPOST(t *testing.T) {
 	require.NoError(t, err)
 
 	assertBatch(t, listenerCh, string(buf))
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 
 	assertMonitor(t, monitorCh, 1, 1)
 }
@@ -337,7 +337,7 @@ func TestHTTPListenerConcurrency(t *testing.T) {
 		}
 	}
 
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 }
 
 func TestHTTPListenerWithPrecision(t *testing.T) {
@@ -377,7 +377,7 @@ func TestHTTPListenerWithPrecision(t *testing.T) {
 
 	// Check for the expected output.
 	assertBatch(t, listenerCh, out.String())
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 }
 
 func TestBatchAgeHTTPListener(t *testing.T) {
@@ -403,7 +403,7 @@ func TestBatchAgeHTTPListener(t *testing.T) {
 
 	// Line should be emitted after 1 sec.
 	assertBatch(t, listenerCh, line)
-	assertNoMore(t, listenerCh)
+	spouttest.AssertNoMore(t, listenerCh)
 }
 
 func BenchmarkListenerLatency(b *testing.B) {
@@ -549,14 +549,6 @@ func assertBatch(t *testing.T, ch chan string, expected string) {
 	}
 }
 
-func assertNoMore(t *testing.T, ch chan string) {
-	select {
-	case <-ch:
-		t.Fatal("unexpectedly saw message")
-	case <-time.After(spouttest.ShortWait):
-	}
-}
-
 func assertMonitor(t *testing.T, monitorCh chan string, received, sent int) {
 	expected := []string{
 		fmt.Sprintf(`received{component="listener",host="h",name="testlistener"} %d`, received),
diff --git a/spouttest/asserts.go b/spouttest/asserts.go
index 6052a72..7810389 100644
--- a/spouttest/asserts.go
+++ b/spouttest/asserts.go
@@ -25,6 +25,16 @@ func AssertRecv(t *testing.T, ch <-chan string, label, expected string) {
 	}
 }
 
+// AssertNoMore checks that there aren't more items received from a
+// string channel in ShortWait time.
+func AssertNoMore(t *testing.T, ch <-chan string) {
+	select {
+	case <-ch:
+		t.Fatal("unexpectedly saw message")
+	case <-time.After(ShortWait):
+	}
+}
+
 func stripLeadingNL(s string) string {
 	// This allows long `expected` strings to be formatted nicely in
 	// the caller.
diff --git a/spouttest/influxd.go b/spouttest/influxd.go
index 2d29410..37e9e70 100644
--- a/spouttest/influxd.go
+++ b/spouttest/influxd.go
@@ -66,17 +66,20 @@ func (f *FakeInfluxDB) Stop() {
 	f.wg.Wait()
 }
 
-// Lines returned the lines received for each "database". The returned
-// map and slices are copies. Lines() is goroutine-safe.
-func (f *FakeInfluxDB) Lines() map[string][]string {
+// DatabaseCount returns the number of databases that have been
+// written to.
+func (f *FakeInfluxDB) DatabaseCount() int {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	return len(f.lines)
+}
 
-	out := make(map[string][]string)
-	for db, lines := range f.lines {
-		out[db] = append([]string(nil), lines...)
-	}
-	return out
+// Lines returned the lines received for a given "database".
+func (f *FakeInfluxDB) Lines(dbName string) []string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	lines := f.lines[dbName]
+	return append([]string(nil), lines...)
 }
 
 func (f *FakeInfluxDB) run() {