nsqd: buffer and spread UDP writes

Author: mreiferson

apps/nsqd/nsqd.go

@@ -123,6 +123,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.Duration("statsd-interval", opts.StatsdInterval, "duration between pushing to statsd")
 	flagSet.Bool("statsd-mem-stats", opts.StatsdMemStats, "toggle sending memory and GC stats to statsd")
 	flagSet.String("statsd-prefix", opts.StatsdPrefix, "prefix used for keys sent to statsd (%s for host replacement)")
+	flagSet.Int("statsd-udp-packet-size", opts.StatsdUDPPacketSize, "the size in bytes of statsd UDP packets")
 
 	// End to end percentile flags
 	e2eProcessingLatencyPercentiles := app.FloatArray{}

contrib/nsqd.cfg.example

@@ -84,6 +84,9 @@ statsd_interval = "60s"
 ## toggle sending memory and GC stats to statsd
 statsd_mem_stats = true
 
+## the size in bytes of statsd UDP packets
+# statsd_udp_packet_size = 508
+
 
 ## message processing time percentiles to keep track of (float)
 e2e_processing_latency_percentiles = [

internal/statsd/client.go

@@ -1,42 +1,22 @@
 package statsd
 
 import (
-	"errors"
 	"fmt"
-	"net"
-	"time"
+	"io"
 )
 
 type Client struct {
-	conn   net.Conn
-	addr   string
+	w      io.Writer
 	prefix string
 }
 
-func NewClient(addr string, prefix string) *Client {
+func NewClient(w io.Writer, prefix string) *Client {
 	return &Client{
-		addr:   addr,
+		w:      w,
 		prefix: prefix,
 	}
 }
 
-func (c *Client) String() string {
-	return c.addr
-}
-
-func (c *Client) CreateSocket() error {
-	conn, err := net.DialTimeout("udp", c.addr, time.Second)
-	if err != nil {
-		return err
-	}
-	c.conn = conn
-	return nil
-}
-
-func (c *Client) Close() error {
-	return c.conn.Close()
-}
-
 func (c *Client) Incr(stat string, count int64) error {
 	return c.send(stat, "%d|c", count)
 }
@@ -54,10 +34,7 @@ func (c *Client) Gauge(stat string, value int64) error {
 }
 
 func (c *Client) send(stat string, format string, value int64) error {
-	if c.conn == nil {
-		return errors.New("not connected")
-	}
-	format = fmt.Sprintf("%s%s:%s", c.prefix, stat, format)
-	_, err := fmt.Fprintf(c.conn, format, value)
+	format = fmt.Sprintf("%s%s:%s\n", c.prefix, stat, format)
+	_, err := fmt.Fprintf(c.w, format, value)
 	return err
 }

internal/writers/boundary_buffered_writer.go

@@ -0,0 +1,30 @@
+package writers
+
+import (
+	"bufio"
+	"io"
+)
+
+type BoundaryBufferedWriter struct {
+	bw *bufio.Writer
+}
+
+func NewBoundaryBufferedWriter(w io.Writer, size int) *BoundaryBufferedWriter {
+	return &BoundaryBufferedWriter{
+		bw: bufio.NewWriterSize(w, size),
+	}
+}
+
+func (b *BoundaryBufferedWriter) Write(p []byte) (int, error) {
+	if len(p) > b.bw.Available() {
+		err := b.bw.Flush()
+		if err != nil {
+			return 0, err
+		}
+	}
+	return b.bw.Write(p)
+}
+
+func (b *BoundaryBufferedWriter) Flush() error {
+	return b.bw.Flush()
+}

internal/writers/spread_writer.go

@@ -0,0 +1,38 @@
+package writers
+
+import (
+	"io"
+	"time"
+)
+
+type SpreadWriter struct {
+	w        io.Writer
+	interval time.Duration
+	buf      [][]byte
+}
+
+func NewSpreadWriter(w io.Writer, interval time.Duration) *SpreadWriter {
+	return &SpreadWriter{
+		w:        w,
+		interval: interval,
+		buf:      make([][]byte, 0),
+	}
+}
+
+func (s *SpreadWriter) Write(p []byte) (int, error) {
+	b := make([]byte, len(p))
+	copy(b, p)
+	s.buf = append(s.buf, b)
+	return len(p), nil
+}
+
+func (s *SpreadWriter) Flush() {
+	sleep := s.interval / time.Duration(len(s.buf))
+	for _, b := range s.buf {
+		start := time.Now()
+		s.w.Write(b)
+		latency := time.Now().Sub(start)
+		time.Sleep(sleep - latency)
+	}
+	s.buf = s.buf[:0]
+}

nsqd/options.go

@@ -58,10 +58,11 @@ type Options struct {
 	MaxOutputBufferTimeout time.Duration `flag:"max-output-buffer-timeout"`
 
 	// statsd integration
-	StatsdAddress  string        `flag:"statsd-address"`
-	StatsdPrefix   string        `flag:"statsd-prefix"`
-	StatsdInterval time.Duration `flag:"statsd-interval"`
-	StatsdMemStats bool          `flag:"statsd-mem-stats"`
+	StatsdAddress       string        `flag:"statsd-address"`
+	StatsdPrefix        string        `flag:"statsd-prefix"`
+	StatsdInterval      time.Duration `flag:"statsd-interval"`
+	StatsdMemStats      bool          `flag:"statsd-mem-stats"`
+	StatsdUDPPacketSize int           `flag:"statsd-udp-packet-size"`
 
 	// e2e message latency
 	E2EProcessingLatencyWindowTime  time.Duration `flag:"e2e-processing-latency-window-time"`
@@ -130,9 +131,10 @@ func NewOptions() *Options {
 		MaxOutputBufferSize:    64 * 1024,
 		MaxOutputBufferTimeout: 1 * time.Second,
 
-		StatsdPrefix:   "nsq.%s",
-		StatsdInterval: 60 * time.Second,
-		StatsdMemStats: true,
+		StatsdPrefix:        "nsq.%s",
+		StatsdInterval:      60 * time.Second,
+		StatsdMemStats:      true,
+		StatsdUDPPacketSize: 508,
 
 		E2EProcessingLatencyWindowTime: time.Duration(10 * time.Minute),
 

nsqd/statsd.go

@@ -3,9 +3,11 @@ package nsqd
 import (
 	"fmt"
 	"math"
+	"net"
 	"time"
 
 	"github.com/nsqio/nsq/internal/statsd"
+	"github.com/nsqio/nsq/internal/writers"
 )
 
 type Uint64Slice []uint64
@@ -25,20 +27,25 @@ func (s Uint64Slice) Less(i, j int) bool {
 func (n *NSQD) statsdLoop() {
 	var lastMemStats memStats
 	var lastStats []TopicStats
-	ticker := time.NewTicker(n.getOpts().StatsdInterval)
+	interval := n.getOpts().StatsdInterval
+	ticker := time.NewTicker(interval)
 	for {
 		select {
 		case <-n.exitChan:
 			goto exit
 		case <-ticker.C:
-			client := statsd.NewClient(n.getOpts().StatsdAddress, n.getOpts().StatsdPrefix)
-			err := client.CreateSocket()
+			addr := n.getOpts().StatsdAddress
+			prefix := n.getOpts().StatsdPrefix
+			conn, err := net.DialTimeout("udp", addr, time.Second)
 			if err != nil {
-				n.logf(LOG_ERROR, "failed to create UDP socket to statsd(%s)", client)
+				n.logf(LOG_ERROR, "failed to create UDP socket to statsd(%s)", addr)
 				continue
 			}
+			sw := writers.NewSpreadWriter(conn, interval-time.Second)
+			bw := writers.NewBoundaryBufferedWriter(sw, n.getOpts().StatsdUDPPacketSize)
+			client := statsd.NewClient(bw, prefix)
 
-			n.logf(LOG_INFO, "STATSD: pushing stats to %s", client)
+			n.logf(LOG_INFO, "STATSD: pushing stats to %s", addr)
 
 			stats := n.GetStats("", "")
 			for _, topic := range stats {
@@ -128,7 +135,9 @@ func (n *NSQD) statsdLoop() {
 				lastMemStats = ms
 			}
 
-			client.Close()
+			bw.Flush()
+			sw.Flush()
+			conn.Close()
 		}
 	}