cmd/statreceiver: lua-scriptable stat receiver (#636)

* cmd/statreceiver: lua-scriptable stat receiver Change-Id: I3ce0fe3f1ef4b1f4f27eed90bac0e91cfecf22d7 * some updates Change-Id: I7c3485adcda1278fce01ae077b4761b3ddb9fb7a * more comments Change-Id: I0bb22993cd934c3d40fc1da80d07e49e686b80dd * linter fixes Change-Id: Ied014304ecb9aadcf00a6b66ad28f856a428d150 * catch errors Change-Id: I6e1920f1fd941e66199b30bc427285c19769fc70 * review feedback Change-Id: I9d4051851eab18970c5f5ddcf4ff265508e541d3 * errorgroup improvements Change-Id: I4699dda3022f0485fbb50c9dafe692d3921734ff * too tricky the previous thing was better for memory with lots of errors at a time but https://play.golang.org/p/RweTMRjoSCt is too much of a foot gun Change-Id: I23f0b3d77dd4288fcc20b3756a7110359576bf44
2018-12-11 11:24:31 -07:00 · 2018-12-11 11:24:31 -07:00 · 362f447d9f
commit 362f447d9f
parent 9e1ec97b31
20 changed files with 1154 additions and 13 deletions
--- a/cmd/statreceiver/README.md
+++ b/cmd/statreceiver/README.md
@ -0,0 +1,25 @@
+# statreceiver
+
+This package implements a Lua-scriptable pipeline processor for
+[zeebo/admission](https://github.com/zeebo/admission) telemetry packets (like
+[monkit](https://github.com/spacemonkeygo/monkit/) or something).
+
+There are a number of types of objects involved in making this work:
+
+ * *Sources* - A source is a source of packets. Each packet is a byte slice that,
+   when parsed, consists of application and instance identification information
+   (such as the application name and perhaps the MAC address or some other id
+   of the computer running the application), and a list of named floating point
+   values. There are currently two types of sources, a UDP source and a file
+   source. A UDP source appends the current time as the timestamp to all
+   packets, whereas a file source should have a prior timestamp to attach to
+   each packet.
+ * *Packet Destinations* - A packet destination is something that can handle
+   a packet with a timestamp. This is either a packet parser, a UDP packet
+   destination for forwarding to another process, or a file destination that
+   will serialize all packets and timestamps for later replay.
+ * *Metric Destinations* - Once a packet has been parsed, the contained metrics
+   can get sent to a metric destination, such as a time series database, a
+   relational database, stdout, a metric filterer, etc.
+
+Please see example.lua for a good example of using this pipeline.
--- a/cmd/statreceiver/common.go
+++ b/cmd/statreceiver/common.go
@ -0,0 +1,58 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"io"
+	"log"
+	"sync/atomic"
+	"time"
+)
+
+type closerfunc func() error
+
+func (f closerfunc) Close() error { return f() }
+
+// Deliver kicks off a goroutine that reads packets from s and delivers them
+// to p. To stop delivery, call Close on the return value then close the source.
+func Deliver(s Source, p PacketDest) io.Closer {
+	done := new(uint32)
+	go func() {
+		for {
+			data, ts, err := s.Next()
+			if atomic.LoadUint32(done) == 1 {
+				return
+			}
+			if err != nil {
+				log.Printf("failed getting packet: %v", err)
+				continue
+			}
+			err = p.Packet(data, ts)
+			if err != nil {
+				log.Printf("failed delivering packet: %v", err)
+				continue
+			}
+		}
+	}()
+	return closerfunc(func() error {
+		atomic.StoreUint32(done, 1)
+		return nil
+	})
+}
+
+// Source reads incoming packets
+type Source interface {
+	Next() (data []byte, ts time.Time, err error)
+}
+
+// PacketDest handles packets
+type PacketDest interface {
+	Packet(data []byte, ts time.Time) error
+}
+
+// MetricDest handles metrics
+type MetricDest interface {
+	Metric(application, instance string, key []byte, val float64, ts time.Time) (
+		err error)
+}
--- a/cmd/statreceiver/copy.go
+++ b/cmd/statreceiver/copy.go
@ -0,0 +1,173 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"storj.io/storj/pkg/utils"
+)
+
+// PacketCopier sends the same packet to multiple destinations
+type PacketCopier struct {
+	d []PacketDest
+}
+
+// NewPacketCopier creates a packet copier that sends the same packets to
+// the provided different destinations
+func NewPacketCopier(d ...PacketDest) *PacketCopier {
+	return &PacketCopier{d: d}
+}
+
+// Packet implements the PacketDest interface
+func (p *PacketCopier) Packet(data []byte, ts time.Time) (ferr error) {
+	var errs utils.ErrorGroup
+	for _, d := range p.d {
+		errs.Add(d.Packet(data, ts))
+	}
+	return errs.Finish()
+}
+
+// MetricCopier sends the same metric to multiple destinations
+type MetricCopier struct {
+	d []MetricDest
+}
+
+// NewMetricCopier creates a metric copier that sends the same metrics to
+// the provided different destinations
+func NewMetricCopier(d ...MetricDest) *MetricCopier {
+	return &MetricCopier{d: d}
+}
+
+// Metric implements the MetricDest interface
+func (m *MetricCopier) Metric(application, instance string,
+	key []byte, val float64, ts time.Time) (ferr error) {
+	var errs utils.ErrorGroup
+	for _, d := range m.d {
+		errs.Add(d.Metric(application, instance, key, val, ts))
+	}
+	return errs.Finish()
+}
+
+// Packet represents a single packet
+type Packet struct {
+	Data []byte
+	TS   time.Time
+}
+
+// PacketBuffer is a packet buffer. It has a given buffer size and allows
+// packets to buffer in memory to deal with potentially variable processing
+// speeds. PacketBuffers drop packets if the buffer is full.
+type PacketBuffer struct {
+	ch chan Packet
+}
+
+// NewPacketBuffer makes a packet buffer with a buffer size of bufsize
+func NewPacketBuffer(p PacketDest, bufsize int) *PacketBuffer {
+	ch := make(chan Packet, bufsize)
+	go func() {
+		for pkt := range ch {
+			err := p.Packet(pkt.Data, pkt.TS)
+			if err != nil {
+				log.Printf("failed delivering buffered packet: %v", err)
+			}
+		}
+	}()
+	return &PacketBuffer{ch: ch}
+}
+
+// Packet implements the PacketDest interface
+func (p *PacketBuffer) Packet(data []byte, ts time.Time) error {
+	select {
+	case p.ch <- Packet{Data: data, TS: ts}:
+		return nil
+	default:
+		return fmt.Errorf("packet buffer overrun")
+	}
+}
+
+// Metric represents a single metric
+type Metric struct {
+	Application string
+	Instance    string
+	Key         []byte
+	Val         float64
+	TS          time.Time
+}
+
+// MetricBuffer is a metric buffer. It has a given buffer size and allows
+// metrics to buffer in memory to deal with potentially variable processing
+// speeds. MetricBuffers drop metrics if the buffer is full.
+type MetricBuffer struct {
+	ch chan Metric
+}
+
+// NewMetricBuffer makes a metric buffer with a buffer size of bufsize
+func NewMetricBuffer(p MetricDest, bufsize int) *MetricBuffer {
+	ch := make(chan Metric, bufsize)
+	go func() {
+		for pkt := range ch {
+			err := p.Metric(pkt.Application, pkt.Instance, pkt.Key, pkt.Val, pkt.TS)
+			if err != nil {
+				log.Printf("failed delivering buffered metric: %v", err)
+			}
+		}
+	}()
+	return &MetricBuffer{ch: ch}
+}
+
+// Metric implements the MetricDest interface
+func (p *MetricBuffer) Metric(application, instance string, key []byte,
+	val float64, ts time.Time) error {
+	select {
+	case p.ch <- Metric{
+		Application: application,
+		Instance:    instance,
+		Key:         key,
+		Val:         val,
+		TS:          ts}:
+		return nil
+	default:
+		return fmt.Errorf("metric buffer overrun")
+	}
+}
+
+// PacketBufPrep prepares a packet destination for a packet buffer.
+// By default, packet memory is reused, which would cause data race conditions
+// when a buffer is also used. PacketBufPrep copies the memory to make sure
+// there are no data races
+type PacketBufPrep struct {
+	d PacketDest
+}
+
+// NewPacketBufPrep creates a PacketBufPrep
+func NewPacketBufPrep(d PacketDest) *PacketBufPrep {
+	return &PacketBufPrep{d: d}
+}
+
+// Packet implements the PacketDest interface
+func (p *PacketBufPrep) Packet(data []byte, ts time.Time) error {
+	return p.d.Packet(append([]byte(nil), data...), ts)
+}
+
+// MetricBufPrep prepares a metric destination for a metric buffer.
+// By default, metric key memory is reused, which would cause data race
+// conditions when a buffer is also used. MetricBufPrep copies the memory to
+// make sure there are no data races
+type MetricBufPrep struct {
+	d MetricDest
+}
+
+// NewMetricBufPrep creates a MetricBufPrep
+func NewMetricBufPrep(d MetricDest) *MetricBufPrep {
+	return &MetricBufPrep{d: d}
+}
+
+// Metric implements the MetricDest interface
+func (p *MetricBufPrep) Metric(application, instance string, key []byte,
+	val float64, ts time.Time) error {
+	return p.d.Metric(application, instance, append([]byte(nil), key...), val, ts)
+}
--- a/cmd/statreceiver/db.go
+++ b/cmd/statreceiver/db.go
@ -0,0 +1,62 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"sync"
+	"time"
+
+	_ "github.com/lib/pq"
+	_ "github.com/mattn/go-sqlite3"
+)
+
+var (
+	sqlupsert = map[string]string{
+		"sqlite3": "INSERT INTO metrics (metric, instance, val, timestamp) " +
+			"VALUES (?, ?, ?, ?) ON CONFLICT(metric, instance) DO UPDATE SET " +
+			"val=excluded.val, timestamp=excluded.timestamp;",
+		"postgres": "INSERT INTO metrics (metric, instance, val, timestamp) " +
+			"VALUES ($1, $2, $3, $4) ON CONFLICT(metric, instance) DO UPDATE SET " +
+			"val=EXCLUDED.val, timestamp=EXCLUDED.timestamp;",
+	}
+)
+
+// DBDest is a database metric destination. It stores the latest value given
+// a metric key and application per instance.
+type DBDest struct {
+	mtx             sync.Mutex
+	driver, address string
+	db              *sql.DB
+}
+
+// NewDBDest creates a DBDest
+func NewDBDest(driver, address string) *DBDest {
+	if _, found := sqlupsert[driver]; !found {
+		panic(fmt.Sprintf("driver %s not supported", driver))
+	}
+	return &DBDest{
+		driver:  driver,
+		address: address,
+	}
+}
+
+// Metric implements the MetricDest interface
+func (db *DBDest) Metric(application, instance string, key []byte, val float64,
+	ts time.Time) error {
+	db.mtx.Lock()
+	if db.db == nil {
+		conn, err := sql.Open(db.driver, db.address)
+		if err != nil {
+			db.mtx.Unlock()
+			return err
+		}
+		db.db = conn
+	}
+	db.mtx.Unlock()
+	_, err := db.db.Exec(sqlupsert[db.driver], application+"."+string(key),
+		instance, val, ts.Unix())
+	return err
+}
--- a/cmd/statreceiver/example.lua
+++ b/cmd/statreceiver/example.lua
@ -0,0 +1,56 @@
+-- possible sources:
+--  * udpin(address)
+--  * filein(path)
+-- multiple sources can be handled in the same run (including multiple sources
+-- of the same type) by calling deliver more than once.
+source = udpin("localhost:9000")
+
+-- multiple metric destination types
+--  * graphite(address) goes to tcp with the graphite wire protocol
+--  * print() goes to stdout
+--  * db("sqlite3", path) goes to sqlite
+--  * db("postgres", connstring) goes to postgres
+graphite_out = graphite("localhost:5555")
+db_out = mcopy(
+  db("sqlite3", "db.db"),
+  db("postgres", "user=dbuser dbname=dbname"))
+
+metric_handlers = mcopy(
+  -- send all satellite data to graphite
+  appfilter("satellite-prod",
+    graphite("localhost:5555")),
+  -- send specific storagenode data to the db
+  appfilter("storagenode-prod",
+    keyfilter(
+      "env\\.process\\." ..
+        "|hw\\.disk\\..*Used" ..
+        "|hw\\.disk\\..*Avail" ..
+        "|hw\\.network\\.stats\\..*\\.(tx|rx)_bytes\\.(deriv|val)",
+      db_out)),
+  -- just print uplink stuff
+  appfilter("uplink-prod",
+    print()))
+
+-- create a metric parser.
+metric_parser =
+  parse(  -- parse takes one or two arguments. the first argument is
+          -- a metric handler, the remaining one is a per-packet application or
+          -- instance filter. each filter is a regex. all packets must
+          -- match all packet filters.
+    sanitize(metric_handlers), -- sanitize converts weird chars to underscores
+    packetfilter("storagenode-prod|satellite-prod|uplink-prod", ""))
+
+-- pcopy forks data to multiple outputs
+-- output types include parse, fileout, and udpout
+destination = pcopy(
+  fileout("dump.out"),
+  metric_parser,
+
+  -- useful local debugging
+  udpout("localhost:9001"),
+
+  -- rothko
+  udpout("localhost:9002"))
+
+-- tie the source to the destination
+deliver(source, destination)
--- a/cmd/statreceiver/file.go
+++ b/cmd/statreceiver/file.go
@ -0,0 +1,75 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"bufio"
+	"encoding/gob"
+	"io"
+	"os"
+	"sync"
+	"time"
+)
+
+// FileSource reads packets from a file
+type FileSource struct {
+	mtx     sync.Mutex
+	path    string
+	decoder *gob.Decoder
+}
+
+// NewFileSource creates a FileSource
+func NewFileSource(path string) *FileSource {
+	return &FileSource{path: path}
+}
+
+// Next implements the Source interface
+func (f *FileSource) Next() ([]byte, time.Time, error) {
+	f.mtx.Lock()
+	defer f.mtx.Unlock()
+	if f.decoder == nil {
+		fh, err := os.Open(f.path)
+		if err != nil {
+			return nil, time.Time{}, err
+		}
+		f.decoder = gob.NewDecoder(bufio.NewReader(fh))
+	}
+
+	var p Packet
+	err := f.decoder.Decode(&p)
+	if err != nil {
+		return nil, time.Time{}, err
+	}
+	return p.Data, p.TS, nil
+}
+
+// FileDest sends packets to a file for later processing. FileDest preserves
+// the timestamps.
+type FileDest struct {
+	mtx     sync.Mutex
+	path    string
+	fh      io.Closer
+	encoder *gob.Encoder
+}
+
+// NewFileDest creates a FileDest
+func NewFileDest(path string) *FileDest {
+	return &FileDest{path: path}
+}
+
+// Packet implements PacketDest
+func (f *FileDest) Packet(data []byte, ts time.Time) error {
+	f.mtx.Lock()
+	defer f.mtx.Unlock()
+
+	if f.encoder == nil {
+		fh, err := os.Create(f.path)
+		if err != nil {
+			return err
+		}
+		f.fh = fh
+		f.encoder = gob.NewEncoder(bufio.NewWriter(fh))
+	}
+	return f.encoder.Encode(Packet{Data: data, TS: ts})
+}
--- a/cmd/statreceiver/filter.go
+++ b/cmd/statreceiver/filter.go
@ -0,0 +1,107 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"regexp"
+	"time"
+)
+
+// PacketFilter is used during Packet parsing to determine if the Packet should
+// continue to be parsed.
+type PacketFilter struct {
+	appRegex, instRegex *regexp.Regexp
+}
+
+// NewPacketFilter creates a PacketFilter. It takes an application regular
+// expression and an instance regular expression. If the regular expression
+// is matched, the packet will be parsed.
+func NewPacketFilter(applicationRegex, instanceRegex string) *PacketFilter {
+	return &PacketFilter{
+		appRegex:  regexp.MustCompile(applicationRegex),
+		instRegex: regexp.MustCompile(instanceRegex),
+	}
+}
+
+// Filter returns true if the application and instance match the filter.
+func (a *PacketFilter) Filter(application, instance string) bool {
+	return a.appRegex.MatchString(application) && a.instRegex.MatchString(instance)
+}
+
+// KeyFilter is a MetricDest that only passes along metrics that pass the key
+// filter
+type KeyFilter struct {
+	re *regexp.Regexp
+	m  MetricDest
+}
+
+// NewKeyFilter creates a KeyFilter. regex is the regular expression that must
+// match, and m is the MetricDest to send matching metrics to.
+func NewKeyFilter(regex string, m MetricDest) *KeyFilter {
+	return &KeyFilter{
+		re: regexp.MustCompile(regex),
+		m:  m,
+	}
+}
+
+// Metric implements MetricDest
+func (k *KeyFilter) Metric(application, instance string,
+	key []byte, val float64, ts time.Time) error {
+	if k.re.Match(key) {
+		return k.m.Metric(application, instance, key, val, ts)
+	}
+	return nil
+}
+
+// ApplicationFilter is a MetricDest that only passes along metrics that pass
+// the application filter
+type ApplicationFilter struct {
+	re *regexp.Regexp
+	m  MetricDest
+}
+
+// NewApplicationFilter creates an ApplicationFilter. regex is the regular
+// expression that must match, and m is the MetricDest to send matching metrics
+// to.
+func NewApplicationFilter(regex string, m MetricDest) *ApplicationFilter {
+	return &ApplicationFilter{
+		re: regexp.MustCompile(regex),
+		m:  m,
+	}
+}
+
+// Metric implements MetricDest
+func (k *ApplicationFilter) Metric(application, instance string,
+	key []byte, val float64, ts time.Time) error {
+	if k.re.MatchString(application) {
+		return k.m.Metric(application, instance, key, val, ts)
+	}
+	return nil
+}
+
+// InstanceFilter is a MetricDest that only passes along metrics that pass
+// the instance filter
+type InstanceFilter struct {
+	re *regexp.Regexp
+	m  MetricDest
+}
+
+// NewInstanceFilter creates an InstanceFilter. regex is the regular
+// expression that must match, and m is the MetricDest to send matching metrics
+// to.
+func NewInstanceFilter(regex string, m MetricDest) *InstanceFilter {
+	return &InstanceFilter{
+		re: regexp.MustCompile(regex),
+		m:  m,
+	}
+}
+
+// Metric implements MetricDest
+func (k *InstanceFilter) Metric(application, instance string,
+	key []byte, val float64, ts time.Time) error {
+	if k.re.MatchString(instance) {
+		return k.m.Metric(application, instance, key, val, ts)
+	}
+	return nil
+}
--- a/cmd/statreceiver/graphite.go
+++ b/cmd/statreceiver/graphite.go
@ -0,0 +1,81 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"bufio"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"time"
+)
+
+// GraphiteDest is a MetricDest that sends data with the Graphite TCP wire
+// protocol
+type GraphiteDest struct {
+	mtx     sync.Mutex
+	address string
+	conn    net.Conn
+	buf     *bufio.Writer
+	stopped bool
+}
+
+// NewGraphiteDest creates a GraphiteDest with TCP address address. Because
+// this function is called in a Lua pipeline domain-specific language, the DSL
+// wants a graphite destination to be flushing every few seconds, so this
+// constructor will start that process. Use Close to stop it.
+func NewGraphiteDest(address string) *GraphiteDest {
+	rv := &GraphiteDest{address: address}
+	go rv.flush()
+	return rv
+}
+
+// Metric implements MetricDest
+func (d *GraphiteDest) Metric(application, instance string,
+	key []byte, val float64, ts time.Time) error {
+
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
+
+	if d.conn == nil {
+		conn, err := net.Dial("tcp", d.address)
+		if err != nil {
+			return err
+		}
+		d.conn = conn
+		d.buf = bufio.NewWriter(conn)
+	}
+
+	_, err := fmt.Fprintf(d.buf, "%s.%s.%s %v %d\n", application, string(key),
+		instance, val, ts.Unix())
+	return err
+}
+
+// Close stops the flushing goroutine
+func (d *GraphiteDest) Close() error {
+	d.mtx.Lock()
+	d.stopped = true
+	d.mtx.Unlock()
+	return nil
+}
+
+func (d *GraphiteDest) flush() {
+	for {
+		time.Sleep(5 * time.Second)
+		d.mtx.Lock()
+		if d.stopped {
+			d.mtx.Unlock()
+			return
+		}
+		var err error
+		if d.buf != nil {
+			err = d.buf.Flush()
+		}
+		d.mtx.Unlock()
+		if err != nil {
+			log.Printf("failed flushing: %v", err)
+		}
+	}
+}
--- a/cmd/statreceiver/main.go
+++ b/cmd/statreceiver/main.go
@ -0,0 +1,88 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+
+	"github.com/spf13/cobra"
+
+	"storj.io/storj/pkg/cfgstruct"
+	"storj.io/storj/pkg/luacfg"
+	"storj.io/storj/pkg/process"
+	"storj.io/storj/pkg/utils"
+)
+
+// Config is the set of configuration values we care about
+var Config struct {
+	Input string `default:"" help:"path to configuration file"`
+}
+
+const defaultConfDir = "$HOME/.storj/statreceiver"
+
+func main() {
+	cmd := &cobra.Command{
+		Use:   os.Args[0],
+		Short: "stat receiving",
+		RunE:  Main,
+	}
+	cfgstruct.Bind(cmd.Flags(), &Config, cfgstruct.ConfDir(defaultConfDir))
+	cmd.Flags().String("config", filepath.Join(defaultConfDir, "config.yaml"),
+		"path to configuration")
+	process.Exec(cmd)
+}
+
+// Main is the real main method
+func Main(cmd *cobra.Command, args []string) error {
+	var input io.Reader
+	switch Config.Input {
+	case "":
+		return fmt.Errorf("--input path to script is required")
+	case "stdin":
+		input = os.Stdin
+	default:
+		fh, err := os.Open(Config.Input)
+		if err != nil {
+			return err
+		}
+		defer utils.LogClose(fh)
+		input = fh
+	}
+
+	s := luacfg.NewScope()
+	err := utils.CombineErrors(
+		s.RegisterVal("deliver", Deliver),
+		s.RegisterVal("filein", NewFileSource),
+		s.RegisterVal("fileout", NewFileDest),
+		s.RegisterVal("udpin", NewUDPSource),
+		s.RegisterVal("udpout", NewUDPDest),
+		s.RegisterVal("parse", NewParser),
+		s.RegisterVal("print", NewPrinter),
+		s.RegisterVal("pcopy", NewPacketCopier),
+		s.RegisterVal("mcopy", NewMetricCopier),
+		s.RegisterVal("pbuf", NewPacketBuffer),
+		s.RegisterVal("mbuf", NewMetricBuffer),
+		s.RegisterVal("packetfilter", NewPacketFilter),
+		s.RegisterVal("appfilter", NewApplicationFilter),
+		s.RegisterVal("instfilter", NewInstanceFilter),
+		s.RegisterVal("keyfilter", NewKeyFilter),
+		s.RegisterVal("sanitize", NewSanitizer),
+		s.RegisterVal("graphite", NewGraphiteDest),
+		s.RegisterVal("db", NewDBDest),
+		s.RegisterVal("pbufprep", NewPacketBufPrep),
+		s.RegisterVal("mbufprep", NewMetricBufPrep),
+	)
+	if err != nil {
+		return err
+	}
+	err = s.Run(input)
+	if err != nil {
+		return err
+	}
+
+	select {}
+}
--- a/cmd/statreceiver/parser.go
+++ b/cmd/statreceiver/parser.go
@ -0,0 +1,71 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"log"
+	"sync"
+	"time"
+
+	"github.com/zeebo/admission/admproto"
+)
+
+const (
+	kb = 1024
+)
+
+// Parser is a PacketDest that sends data to a MetricDest
+type Parser struct {
+	d       MetricDest
+	f       []*PacketFilter
+	scratch sync.Pool
+}
+
+// NewParser creates a Parser. It sends metrics to d, provided they pass all
+// of the provided PacketFilters
+func NewParser(d MetricDest, f ...*PacketFilter) *Parser {
+	return &Parser{
+		d: d, f: f,
+		scratch: sync.Pool{
+			New: func() interface{} {
+				var x [10 * kb]byte
+				return &x
+			},
+		}}
+}
+
+// Packet implements PacketDest
+func (p *Parser) Packet(data []byte, ts time.Time) (err error) {
+	data, err = admproto.CheckChecksum(data)
+	if err != nil {
+		return err
+	}
+	scratch := p.scratch.Get().(*[10 * kb]byte)
+	defer p.scratch.Put(scratch)
+	r := admproto.NewReaderWith((*scratch)[:])
+	data, appb, instb, err := r.Begin(data)
+	if err != nil {
+		return err
+	}
+	app, inst := string(appb), string(instb)
+	for _, f := range p.f {
+		if !f.Filter(app, inst) {
+			return nil
+		}
+	}
+	var key []byte
+	var value float64
+	for len(data) > 0 {
+		data, key, value, err = r.Next(data)
+		if err != nil {
+			return err
+		}
+		err = p.d.Metric(app, inst, key, value, ts)
+		if err != nil {
+			log.Printf("failed to write metric: %v", err)
+			continue
+		}
+	}
+	return nil
+}
--- a/cmd/statreceiver/print.go
+++ b/cmd/statreceiver/print.go
@ -0,0 +1,29 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+// Printer is a MetricDest that writes to stdout
+type Printer struct {
+	mtx sync.Mutex
+}
+
+// NewPrinter creates a Printer
+func NewPrinter() *Printer {
+	return &Printer{}
+}
+
+// Metric implements MetricDest
+func (p *Printer) Metric(application, instance string, key []byte, val float64,
+	ts time.Time) error {
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	_, err := fmt.Println(application, instance, string(key), val, ts.Unix())
+	return err
+}
--- a/cmd/statreceiver/sanitizer.go
+++ b/cmd/statreceiver/sanitizer.go
@ -0,0 +1,48 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"bytes"
+	"strings"
+	"time"
+	"unicode"
+)
+
+// Sanitizer is a MetricDest that replaces nonalphanumeric characters with
+// underscores.
+type Sanitizer struct {
+	m MetricDest
+}
+
+// NewSanitizer creates a Sanitizer that sends sanitized metrics to m.
+func NewSanitizer(m MetricDest) *Sanitizer { return &Sanitizer{m: m} }
+
+// Metric implements MetricDest
+func (s *Sanitizer) Metric(application, instance string, key []byte,
+	val float64, ts time.Time) error {
+	return s.m.Metric(sanitize(application), sanitize(instance), sanitizeb(key),
+		val, ts)
+}
+
+func sanitize(val string) string {
+	return strings.Replace(strings.Map(safechar, val), "..", ".", -1)
+}
+
+func sanitizeb(val []byte) []byte {
+	return bytes.Replace(bytes.Map(safechar, val), []byte(".."), []byte("."), -1)
+}
+
+func safechar(r rune) rune {
+	if unicode.IsLetter(r) || unicode.IsNumber(r) {
+		return r
+	}
+	switch r {
+	case '/':
+		return '.'
+	case '.', '-':
+		return r
+	}
+	return '_'
+}
--- a/cmd/statreceiver/schema.sql
+++ b/cmd/statreceiver/schema.sql
@ -0,0 +1,9 @@
+create table if not exists metrics (
+  metric text,
+  instance text,
+  val real,
+  timestamp integer,
+  primary key (metric, instance)
+);
+
+create unique index metrics_index ON metrics(metric, instance);
--- a/cmd/statreceiver/udp.go
+++ b/cmd/statreceiver/udp.go
@ -0,0 +1,112 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"time"
+)
+
+// UDPSource is a packet source
+type UDPSource struct {
+	mtx     sync.Mutex
+	address string
+	conn    *net.UDPConn
+	buf     [1024 * 10]byte
+	closed  bool
+}
+
+// NewUDPSource creates a UDPSource that listens on address
+func NewUDPSource(address string) *UDPSource {
+	return &UDPSource{address: address}
+}
+
+// Next implements the Source interface
+func (s *UDPSource) Next() ([]byte, time.Time, error) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if s.closed {
+		return nil, time.Time{}, fmt.Errorf("udp source closed")
+	}
+	if s.conn == nil {
+		addr, err := net.ResolveUDPAddr("udp", s.address)
+		if err != nil {
+			return nil, time.Time{}, err
+		}
+		conn, err := net.ListenUDP("udp", addr)
+		if err != nil {
+			return nil, time.Time{}, err
+		}
+		s.conn = conn
+	}
+
+	n, _, err := s.conn.ReadFrom(s.buf[:])
+	if err != nil {
+		return nil, time.Time{}, err
+	}
+	return s.buf[:n], time.Now(), nil
+}
+
+// Close closes the source
+func (s *UDPSource) Close() error {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	s.closed = true
+	if s.conn != nil {
+		return s.conn.Close()
+	}
+	return nil
+}
+
+// UDPDest is a packet destination. IMPORTANT: It throws away timestamps.
+type UDPDest struct {
+	mtx     sync.Mutex
+	address string
+	addr    *net.UDPAddr
+	conn    *net.UDPConn
+	closed  bool
+}
+
+// NewUDPDest creates a UDPDest that sends incoming packets to address.
+func NewUDPDest(address string) *UDPDest {
+	return &UDPDest{address: address}
+}
+
+// Packet implements PacketDest
+func (d *UDPDest) Packet(data []byte, ts time.Time) error {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
+	if d.closed {
+		return fmt.Errorf("closed destination")
+	}
+
+	if d.conn == nil {
+		addr, err := net.ResolveUDPAddr("udp", d.address)
+		if err != nil {
+			return err
+		}
+		conn, err := net.ListenUDP("udp", &net.UDPAddr{Port: 0})
+		if err != nil {
+			return err
+		}
+		d.addr = addr
+		d.conn = conn
+	}
+
+	_, err := d.conn.WriteTo(data, d.addr)
+	return err
+}
+
+// Close closes the destination
+func (d *UDPDest) Close() error {
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
+	d.closed = true
+	if d.conn != nil {
+		return d.conn.Close()
+	}
+	return nil
+}
--- a/examples/luacfg/main.go
+++ b/examples/luacfg/main.go
@ -0,0 +1,20 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"storj.io/storj/pkg/luacfg"
+)
+
+func main() {
+	s := luacfg.NewScope()
+	s.RegisterVal("print", fmt.Println)
+	err := s.Run(os.Stdin)
+	if err != nil {
+		panic(err)
+	}
+}
--- a/go.mod
+++ b/go.mod
@ -21,6 +21,7 @@ exclude gopkg.in/olivere/elastic.v5 v5.0.72 // buggy import, see https://github.

 require (
 	github.com/BurntSushi/toml v0.3.1 // indirect
+	github.com/Shopify/go-lua v0.0.0-20181106184032-48449c60c0a9
 	github.com/Shopify/toxiproxy v2.1.3+incompatible // indirect
 	github.com/StackExchange/wmi v0.0.0-20180725035823-b12b22c5341f // indirect
 	github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6 // indirect
@ -54,6 +55,7 @@ require (
 	github.com/howeyc/gopass v0.0.0-20170109162249-bf9dde6d0d2c // indirect
 	github.com/inconshreveable/go-update v0.0.0-20160112193335-8152e7eb6ccf // indirect
 	github.com/jbenet/go-base58 v0.0.0-20150317085156-6237cf65f3a6
+	github.com/jtolds/go-luar v0.0.0-20170419063437-0786921db8c0
 	github.com/jtolds/monkit-hw v0.0.0-20180827162413-5a254051f35d
 	github.com/klauspost/cpuid v0.0.0-20180405133222-e7e905edc00e // indirect
 	github.com/klauspost/reedsolomon v0.0.0-20180704173009-925cb01d6510 // indirect
@ -63,7 +65,7 @@ require (
 	github.com/mailru/easyjson v0.0.0-20180823135443-60711f1a8329 // indirect
 	github.com/mattn/go-isatty v0.0.4 // indirect
 	github.com/mattn/go-runewidth v0.0.3 // indirect
-	github.com/mattn/go-sqlite3 v1.9.0
+	github.com/mattn/go-sqlite3 v1.10.0
 	github.com/minio/cli v1.3.0
 	github.com/minio/dsync v0.0.0-20180124070302-439a0961af70 // indirect
 	github.com/minio/highwayhash v0.0.0-20180501080913-85fc8a2dacad // indirect
--- a/go.sum
+++ b/go.sum
@ -7,6 +7,8 @@ github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Microsoft/go-winio v0.4.11 h1:zoIOcVf0xPN1tnMVbTtEdI+P8OofVk3NObnwOQ6nK2Q=
 github.com/Microsoft/go-winio v0.4.11/go.mod h1:VhR8bwka0BXejwEJY73c50VrPtXAaKcyvVC4A4RozmA=
+github.com/Shopify/go-lua v0.0.0-20181106184032-48449c60c0a9 h1:+2M9NEk3+xSg0+bWzt1kxsL6EtoEg7sgtT11CZjGwq8=
+github.com/Shopify/go-lua v0.0.0-20181106184032-48449c60c0a9/go.mod h1:lvS2IGWEGk+KQkRrCXuWlcsHO5BitT0HyhnP51rh3gA=
 github.com/Shopify/toxiproxy v2.1.3+incompatible h1:awiJqUYH4q4OmoBiRccJykjd7B+w0loJi2keSna4X/M=
 github.com/Shopify/toxiproxy v2.1.3+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
 github.com/Sirupsen/logrus v1.0.6 h1:HCAGQRk48dRVPA5Y+Yh0qdCSTzPOyU1tBJ7Q9YzotII=
@ -163,6 +165,8 @@ github.com/jmespath/go-jmespath v0.0.0-20160202185014-0b12d6b521d8/go.mod h1:Nht
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af/go.mod h1:Nht3zPeWKUH0NzdCt2Blrr5ys8VGpn0CEB0cQHVjt7k=
 github.com/jtolds/gls v4.2.1+incompatible h1:fSuqC+Gmlu6l/ZYAoZzx2pyucC8Xza35fpRVWLVmUEE=
 github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
+github.com/jtolds/go-luar v0.0.0-20170419063437-0786921db8c0 h1:UyVaeqfY1fLPMt1iUTaWsxUNxYAzZVyK+7G+a3sRfhk=
+github.com/jtolds/go-luar v0.0.0-20170419063437-0786921db8c0/go.mod h1:OtVLEpPHGJkn8jgGrHlYELCA3uXLU0YSfNN0faeDM2M=
 github.com/jtolds/monkit-hw v0.0.0-20180827162413-5a254051f35d h1:+TWXq24sHXNI6DVobckzgJCbCvb4bYjDZuKRR+d/Nuw=
 github.com/jtolds/monkit-hw v0.0.0-20180827162413-5a254051f35d/go.mod h1:1liUiYZzx8ZoPTfZrBE2q8DFRDPafts0GGjslOtcZcw=
 github.com/kisielk/gotool v1.0.0 h1:AV2c/EiW3KqPNT9ZKl07ehoAGi4C5/01Cfbblndcapg=
@ -197,6 +201,8 @@ github.com/mattn/go-runewidth v0.0.3 h1:a+kO+98RDGEfo6asOGMmpodZq4FNtnGP54yps8Bz
 github.com/mattn/go-runewidth v0.0.3/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
 github.com/mattn/go-sqlite3 v1.9.0 h1:pDRiWfl+++eC2FEFRy6jXmQlvp4Yh3z1MJKg4UeYM/4=
 github.com/mattn/go-sqlite3 v1.9.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
+github.com/mattn/go-sqlite3 v1.10.0 h1:jbhqpg7tQe4SupckyijYiy0mJJ/pRyHvXf7JdWK860o=
+github.com/mattn/go-sqlite3 v1.10.0/go.mod h1:FPy6KqzDD04eiIsT53CuJW3U88zkxoIYsOqkbpncsNc=
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/minio/cli v1.3.0 h1:vB0iUpmyaH54+1jJJj62Aa0qFF3xO3i0J3IcKiM6bHM=
--- a/pkg/luacfg/cfg.go
+++ b/pkg/luacfg/cfg.go
@ -0,0 +1,85 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package luacfg
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"sync"
+
+	lua "github.com/Shopify/go-lua"
+	luar "github.com/jtolds/go-luar"
+)
+
+// Scope represents a collection of values registered in a Lua namespace.
+type Scope struct {
+	mtx           sync.Mutex
+	registrations map[string]func(*lua.State) error
+}
+
+// NewScope creates an empty Scope.
+func NewScope() *Scope {
+	return &Scope{
+		registrations: map[string]func(*lua.State) error{},
+	}
+}
+
+// RegisterType allows you to add a Lua function that creates new
+// values of the given type to the scope.
+func (s *Scope) RegisterType(name string, example interface{}) error {
+	return s.register(name, example, luar.PushType)
+}
+
+// RegisterVal adds the Go value 'value', including Go functions, to the Lua
+// scope.
+func (s *Scope) RegisterVal(name string, value interface{}) error {
+	return s.register(name, value, luar.PushValue)
+}
+
+func (s *Scope) register(name string, val interface{},
+	pusher func(l *lua.State, val interface{}) error) error {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	if _, exists := s.registrations[name]; exists {
+		return fmt.Errorf("Registration %#v already exists", name)
+	}
+	s.registrations[name] = func(l *lua.State) error {
+		err := pusher(l, val)
+		if err != nil {
+			return err
+		}
+		l.SetGlobal(name)
+		return nil
+	}
+	return nil
+}
+
+// Run runs the Lua source represented by in
+func (s *Scope) Run(in io.Reader) error {
+	l := lua.NewState()
+	luar.SetOptions(l, luar.Options{AllowUnexportedAccess: true})
+
+	s.mtx.Lock()
+	registrations := make([]func(l *lua.State) error, 0, len(s.registrations))
+	for _, reg := range s.registrations {
+		registrations = append(registrations, reg)
+	}
+	s.mtx.Unlock()
+
+	for _, reg := range registrations {
+		err := reg(l)
+		if err != nil {
+			return err
+		}
+	}
+
+	data, err := ioutil.ReadAll(in)
+	if err != nil {
+		return err
+	}
+
+	err = lua.DoString(l, string(data))
+	return err
+}
--- a/pkg/utils/utils.go
+++ b/pkg/utils/utils.go
@ -50,18 +50,9 @@ func ParseURL(s string) (*url.URL, error) {

 // CombineErrors combines multiple errors to a single error
 func CombineErrors(errs ...error) error {
-	var errlist combinedError
-	for _, err := range errs {
-		if err != nil {
-			errlist = append(errlist, err)
-		}
-	}
-	if len(errlist) == 0 {
-		return nil
-	} else if len(errlist) == 1 {
-		return errlist[0]
-	}
-	return errlist
+	var errlist ErrorGroup
+	errlist.Add(errs...)
+	return errlist.Finish()
 }

 type combinedError []error
@ -88,6 +79,31 @@ func (errs combinedError) Error() string {
 	return ""
 }

+// ErrorGroup contains a set of non-nil errors
+type ErrorGroup []error
+
+// Add adds an error to the ErrorGroup if it is non-nil
+func (e *ErrorGroup) Add(errs ...error) {
+	for _, err := range errs {
+		if err != nil {
+			*e = append(*e, err)
+		}
+	}
+}
+
+// Finish returns nil if there were no non-nil errors, the first error if there
+// was only one non-nil error, or the result of CombineErrors if there was more
+// than one non-nil error.
+func (e ErrorGroup) Finish() error {
+	if len(e) == 0 {
+		return nil
+	}
+	if len(e) == 1 {
+		return e[0]
+	}
+	return combinedError(e)
+}
+
 // CollectErrors returns first error from channel and all errors that happen within duration
 func CollectErrors(errch chan error, duration time.Duration) error {
 	errch = discardNil(errch)
--- a/pkg/utils/utils_test.go
+++ b/pkg/utils/utils_test.go
@ -40,3 +40,21 @@ func TestCollecMultipleError(t *testing.T) {
 	assert.Error(t, err)
 	assert.Equal(t, err.Error(), "error1\nerror2\nerror3")
 }
+
+func TestErrorGroup(t *testing.T) {
+	var errlist utils.ErrorGroup
+	errlist.Add(nil, nil, nil)
+	assert.NoError(t, errlist.Finish())
+	assert.Equal(t, len(errlist), 0)
+	e1 := errs.New("err1")
+	errlist.Add(nil, nil, e1, nil)
+	assert.Equal(t, errlist.Finish(), e1)
+	assert.Equal(t, len(errlist), 1)
+	e2, e3 := errs.New("err2"), errs.New("err3")
+	errlist.Add(e2, e3)
+	assert.Error(t, errlist.Finish())
+	assert.Equal(t, len(errlist), 3)
+	assert.Equal(t, errlist[0], e1)
+	assert.Equal(t, errlist[1], e2)
+	assert.Equal(t, errlist[2], e3)
+}