http listener refactor

in this commit:

- chunks out the http request body to avoid making very large
  allocations.
- establishes a limit for the maximum http request body size that the
  listener will accept.
- utilizes a pool of byte buffers to reduce GC pressure.

internal/buffer/buffer.go

@@ -1,6 +1,8 @@
 package buffer
 
 import (
+	"sync"
+
 	"github.com/influxdata/telegraf"
 )
 
@@ -11,6 +13,8 @@ type Buffer struct {
 	drops int
 	// total metrics added
 	total int
+
+	sync.Mutex
 }
 
 // NewBuffer returns a Buffer
@@ -61,11 +65,13 @@ func (b *Buffer) Add(metrics ...telegraf.Metric) {
 // the batch will be of maximum length batchSize. It can be less than batchSize,
 // if the length of Buffer is less than batchSize.
 func (b *Buffer) Batch(batchSize int) []telegraf.Metric {
+	b.Lock()
 	n := min(len(b.buf), batchSize)
 	out := make([]telegraf.Metric, n)
 	for i := 0; i < n; i++ {
 		out[i] = <-b.buf
 	}
+	b.Unlock()
 	return out
 }
 

plugins/inputs/http_listener/bufferpool.go

@@ -0,0 +1,43 @@
+package http_listener
+
+import (
+	"sync/atomic"
+)
+
+type pool struct {
+	buffers chan []byte
+	size    int
+
+	created int64
+}
+
+// NewPool returns a new pool object.
+// n is the number of buffers
+// bufSize is the size (in bytes) of each buffer
+func NewPool(n, bufSize int) *pool {
+	return &pool{
+		buffers: make(chan []byte, n),
+		size:    bufSize,
+	}
+}
+
+func (p *pool) get() []byte {
+	select {
+	case b := <-p.buffers:
+		return b
+	default:
+		atomic.AddInt64(&p.created, 1)
+		return make([]byte, p.size)
+	}
+}
+
+func (p *pool) put(b []byte) {
+	select {
+	case p.buffers <- b:
+	default:
+	}
+}
+
+func (p *pool) ncreated() int64 {
+	return atomic.LoadInt64(&p.created)
+}