Repository: honeycombio/honeycomb-kubernetes-agent
Pull request #232 (2022-02-22 18:05:50)

Implements retry logic for sending events
Adds a ring buffer with configurable size and expiration timeout to be leveraged by the libhoney response channel. As events are sent to Honeycomb, they are added to the buffer. If during response handling, the event was not successfully sent, the event will be retried for sending.  Libhoney is also now configured to block when the libhoney buffers are full.

The buffer size and timeout are configurable.  Setting the buffer size to 0, will disable the usage of the buffer, and continue operation as before this change.

The buffer will most certainly increase memory consumption, especially on startup when the agent is backfilling logs. The buffer should never be set to more than 16000 (2 x libhoney buffer of 8000). Even a buffer of 16000 can increase initial memory consumption by 150+MB.  As such a recommended default of 1000 which provides retry protection should Honeycomb service return a 500 status, with a modest memory consumption increase.


Patch:
diff --git a/config/config.go b/config/config.go
index 0e7105d..879c8fb 100644
--- a/config/config.go
+++ b/config/config.go
@@ -13,13 +13,15 @@ type Config struct {
 	APIKey  string `yaml:"apiKey"`
 	APIHost string `yaml:"apiHost"`
 	// Deprecated: use APIKey instead.
-	WriteKey         string `yaml:"writekey"`
-	Watchers         []*WatcherConfig
-	Verbosity        string
-	LegacyLogPaths   bool                   `yaml:"legacyLogPaths"`
-	SplitLogging     bool                   `yaml:"splitLogging"`
-	AdditionalFields map[string]interface{} `yaml:"additionalFields"`
-	Metrics          *MetricsConfig
+	WriteKey          string        `yaml:"writekey"`
+	RetryBufferSize   int           `yaml:"retryBufferSize"`
+	RetryBufferExpire time.Duration `yaml:"retryBufferExpire"`
+	Watchers          []*WatcherConfig
+	Verbosity         string
+	LegacyLogPaths    bool                   `yaml:"legacyLogPaths"`
+	SplitLogging      bool                   `yaml:"splitLogging"`
+	AdditionalFields  map[string]interface{} `yaml:"additionalFields"`
+	Metrics           *MetricsConfig
 }
 
 type WatcherConfig struct {
diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md
index 6f21541..d3d0aaa 100644
--- a/docs/configuration-reference.md
+++ b/docs/configuration-reference.md
@@ -323,6 +323,22 @@ additionalFields:
   owner: me@example.com
 ```
 
+### retryBufferSize
+
+You can enable retrying to send events to Honeycomb by setting this option.
+The default size is 0 which means retry is not enabled.
+The recommendation when first enabling retry is to set this buffer size to 1000 which provides some buffer for temporary Honeycomb API ingest errors with only a modest increase in memory consumption by the agent.
+The retry buffer is a ring buffer, which will contain no more than the maximum size of events, removing the oldest one when new events are added to a filled buffer.
+Note: there is no flushing of the events in the retry buffer on process shutdown.
+
+### retryBufferExpire
+
+When enabling retry, it is recommended to expire events out of the retry buffer to minimize memory impact.
+The default expiration is 0 which means events will remain in the retry buffer until forced out from new events being added.
+With a short expiration time, events that failed to send may expire before being retried.
+To expire events from the buffer, set this to the time duration that marks an event for removal from the buffer.
+This should be set with the appropriate time suffix (e.g., `10s`, `1m`, etc.).
+
 ## Sample configurations
 
 Here are some example configurations for the Honeycomb agent.
@@ -397,7 +413,7 @@ Here are some example configurations for the Honeycomb agent.
         parser: json
         dataset: frontend
     ```
-    
+
 * Process all logs except for those with a given field set to a certain value:
 
     ```
@@ -412,5 +428,5 @@ Here are some example configurations for the Honeycomb agent.
           - drop_event:
               field: level
               values:
-                - debug       
+                - debug
     ```
diff --git a/main.go b/main.go
index d476369..aee10c4 100644
--- a/main.go
+++ b/main.go
@@ -17,7 +17,7 @@ import (
 	"github.com/honeycombio/honeycomb-kubernetes-agent/transmission"
 	"github.com/honeycombio/honeycomb-kubernetes-agent/unwrappers"
 	"github.com/honeycombio/honeycomb-kubernetes-agent/version"
-	libhoney "github.com/honeycombio/libhoney-go"
+	"github.com/honeycombio/libhoney-go"
 	flag "github.com/jessevdk/go-flags"
 	"github.com/sirupsen/logrus"
 
@@ -103,7 +103,7 @@ func main() {
 		}
 	}
 
-	err = transmission.InitLibhoney(apiKey, cfg.APIHost)
+	err = transmission.InitLibhoney(apiKey, cfg.APIHost, cfg.RetryBufferSize, cfg.RetryBufferExpire)
 	if err != nil {
 		logrus.WithError(err).Fatal("Error initializing Honeycomb transmission")
 	}
@@ -182,7 +182,7 @@ func createLogTailers(config *config.Config) ([]*tailer.PathWatcher, []*podtaile
 				logrus.WithError(err).Error("Error setting up watcher")
 				continue
 			}
-			// Even though we have a static path, NewPathWatcher expects a function
+			// Even though we have a static path, NewPathWatcher expects a function,
 			// so we build one that just returns the path
 			patternFunc := func() (string, error) { return path, nil }
 			t := tailer.NewPathWatcher(patternFunc, nil, handlerFactory, stateRecorder)
@@ -224,14 +224,8 @@ func startMetricsService(config *config.MetricsConfig) error {
 		if config.Dataset == "" {
 			config.Dataset = "kubernetes-metrics"
 		}
-		builder := libhoney.NewBuilder()
-		builder.Dataset = config.Dataset
 
-		for k, v := range config.AdditionalFields {
-			builder.AddField(k, v)
-		}
-
-		svc, err := service.NewMetricsService(config, builder, kubeClient)
+		svc, err := service.NewMetricsService(config, kubeClient)
 		if err != nil {
 			return err
 		}
@@ -244,7 +238,7 @@ func startMetricsService(config *config.MetricsConfig) error {
 }
 
 func newKubeClient() (*corev1.CoreV1Client, error) {
-	// Get clientset to query API server.
+	// Get client to query API server.
 	kubeClientConfig, err := rest.InClusterConfig()
 	if err != nil {
 		return nil, err
diff --git a/service/runnable.go b/service/runnable.go
index 497d1e3..bb4ac48 100644
--- a/service/runnable.go
+++ b/service/runnable.go
@@ -7,8 +7,10 @@ import (
 	"context"
 	"time"
 
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+
 	"github.com/honeycombio/honeycomb-kubernetes-agent/metrics"
-	"github.com/honeycombio/libhoney-go"
+	"github.com/honeycombio/honeycomb-kubernetes-agent/transmission"
 	"github.com/sirupsen/logrus"
 
 	v1 "k8s.io/api/core/v1"
@@ -25,26 +27,30 @@ type runnable struct {
 	statsProvider     *kubelet.StatsProvider
 	metadataProvider  *kubelet.MetadataProvider
 	metricsProvider   *metrics.Processor
+	dataset           string
 	interval          time.Duration
 	k8sClusterName    string
 	restClient        kubelet.RestClient
-	builder           *libhoney.Builder
 	omitLabels        []metrics.OmitLabel
+	additionalFields  map[string]interface{}
 	includeNodeLabels bool
 
 	metricGroupsToCollect map[metrics.MetricGroup]bool
+	transmitter           *transmission.HoneycombTransmitter
 	apiClient             corev1.NodesGetter
 }
 
-func newRunnable(rc kubelet.RestClient, builder *libhoney.Builder, opt Options, client *corev1.CoreV1Client) *runnable {
+func newRunnable(rc kubelet.RestClient, opt Options, client *corev1.CoreV1Client) *runnable {
 	return &runnable{
+		dataset:               opt.Dataset,
 		interval:              opt.Interval,
 		k8sClusterName:        opt.ClusterName,
 		restClient:            rc,
-		builder:               builder,
 		omitLabels:            opt.OmitLabels,
+		additionalFields:      opt.AdditionalFields,
 		includeNodeLabels:     opt.IncludeNodeLabels,
 		metricGroupsToCollect: opt.MetricGroupsToCollect,
+		transmitter:           &transmission.HoneycombTransmitter{},
 		apiClient:             client,
 	}
 }
@@ -107,21 +113,14 @@ func (r *runnable) Run() error {
 	// iterate over resource metrics data
 	for _, rm := range resourceMetrics {
 
-		// create event from Resource meta
-		ev, err := r.createEventFromResource(rm.Resource)
-		if err != nil {
-			logrus.WithFields(logrus.Fields{
-				"resourceType": rm.Resource.Type,
-				"resourceName": rm.Resource.Name,
-			}).WithError(err).Error("Could not create event for resource")
-			continue
-		}
-
 		logrus.WithFields(logrus.Fields{
 			"resourceType": rm.Resource.Type,
 			"resourceName": rm.Resource.Name,
 		}).Trace("Creating event for resource")
 
+		// create event from Resource meta
+		ev := r.createEventFromResource(rm.Resource)
+
 		pre := metrics.PrefixMetrics
 
 		// loop through all metrics to add them to resource event
@@ -141,49 +140,46 @@ func (r *runnable) Run() error {
 			}
 
 			// add metric to resource event
-			ev.AddField(pre+k, val)
+			ev.Data[pre+k] = val
 		}
 
 		logrus.WithFields(logrus.Fields{
 			"resourceType": rm.Resource.Type,
 			"resourceName": rm.Resource.Name,
 			"metricCount":  len(rm.Metrics),
-		}).Debug("Event Data: ", ev.Fields())
+		}).Trace("Event Data: ", ev.Data)
 
 		// send resource event to Honeycomb
-		if err = ev.Send(); err != nil {
-			logrus.WithFields(logrus.Fields{
-				"resourceType": rm.Resource.Type,
-				"resourceName": rm.Resource.Name,
-				"metricCount":  len(rm.Metrics),
-			}).WithError(err).Error("Error sending event to honeycomb")
-		}
+		r.transmitter.Send(ev)
 	}
 
 	return nil
 }
 
-func (r *runnable) createEventFromResource(res *metrics.Resource) (*libhoney.Event, error) {
-	ev := r.builder.NewEvent()
+func (r *runnable) createEventFromResource(res *metrics.Resource) *event.Event {
 
 	// add basic attributes to each resource event
-	_ = ev.Add(map[string]string{
+	data := map[string]interface{}{
 		metrics.PrefixMetrics + metrics.MetricSourceName: res.Name,
 		metrics.PrefixMetrics + metrics.MetricSourceType: res.Type,
 		metrics.KubernetesResourceType:                   res.Type,
 		metrics.PrefixCluster + "name":                   r.k8sClusterName,
-	})
-	ev.Timestamp = res.Timestamp
+	}
+
+	// add all additional fields
+	for k, v := range r.additionalFields {
+		data[k] = v
+	}
 
 	// add all labels to event
-	if err := ev.Add(res.Labels); err != nil {
-		return nil, err
+	for k, v := range res.Labels {
+		data[k] = v
 	}
 
-	// add status attributes as fields in event
+	// add status details to event
 	if res.Status != nil {
-		if err := ev.Add(res.Status); err != nil {
-			return nil, err
+		for k, v := range res.Status {
+			data[k] = v
 		}
 
 		// Check if resource restarted
@@ -197,12 +193,18 @@ func (r *runnable) createEventFromResource(res *metrics.Resource) (*libhoney.Eve
 				delta := r.metricsProvider.GetCounterDelta(res, "restarts", m)
 				if delta > 0 {
 					// resource had a restart since last collection
-					ev.AddField(metrics.StatusRestart, true)
+					data[metrics.StatusRestart] = true
 				}
-				ev.AddField(metrics.StatusRestartDelta, delta)
+				data[metrics.StatusRestartDelta] = delta
 			}
 		}
 	}
 
-	return ev, nil
+	ev := &event.Event{
+		Data:      data,
+		Dataset:   r.dataset,
+		Timestamp: res.Timestamp,
+	}
+
+	return ev
 }
diff --git a/service/service.go b/service/service.go
index eb3c31a..bdd1258 100644
--- a/service/service.go
+++ b/service/service.go
@@ -12,7 +12,6 @@ import (
 	"github.com/honeycombio/honeycomb-kubernetes-agent/interval"
 	"github.com/honeycombio/honeycomb-kubernetes-agent/kubelet"
 	"github.com/honeycombio/honeycomb-kubernetes-agent/metrics"
-	"github.com/honeycombio/libhoney-go"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
@@ -31,19 +30,20 @@ type Service struct {
 	restClient kubelet.RestClient
 	runner     *interval.Runner
 	options    Options
-	builder    *libhoney.Builder
 	client     *corev1.CoreV1Client
 }
 
 type Options struct {
+	Dataset               string
 	Interval              time.Duration
 	ClusterName           string
 	OmitLabels            []metrics.OmitLabel
+	AdditionalFields      map[string]interface{}
 	IncludeNodeLabels     bool
 	MetricGroupsToCollect map[metrics.MetricGroup]bool
 }
 
-func NewMetricsService(cfg *config.MetricsConfig, builder *libhoney.Builder, client *corev1.CoreV1Client) (*Service, error) {
+func NewMetricsService(cfg *config.MetricsConfig, client *corev1.CoreV1Client) (*Service, error) {
 
 	if cfg.Endpoint == "" {
 		// Not all Managed K8s offerings allow the nodename to be DNS reachable, try by IP if available.
@@ -70,9 +70,11 @@ func NewMetricsService(cfg *config.MetricsConfig, builder *libhoney.Builder, cli
 	}
 
 	opt := &Options{
+		Dataset:               cfg.Dataset,
 		Interval:              cfg.Interval,
 		ClusterName:           cfg.ClusterName,
 		OmitLabels:            cfg.OmitLabels,
+		AdditionalFields:      cfg.AdditionalFields,
 		IncludeNodeLabels:     cfg.IncludeNodeLabels,
 		MetricGroupsToCollect: mg,
 	}
@@ -81,7 +83,6 @@ func NewMetricsService(cfg *config.MetricsConfig, builder *libhoney.Builder, cli
 		config:     cfg,
 		restClient: rc,
 		options:    *opt,
-		builder:    builder,
 		client:     client,
 	}, nil
 }
@@ -134,7 +135,7 @@ func (s *Service) Start() error {
 	}).Info("Creating Metrics Service Runner...")
 
 	// setup primary interval runner for metrics service
-	runnable := newRunnable(s.restClient, s.builder, s.options, s.client)
+	runnable := newRunnable(s.restClient, s.options, s.client)
 	s.runner = interval.NewRunner("k8s-stats", s.options.Interval, runnable)
 	logrus.Debug("Metrics Service Runner created")
 
diff --git a/transmission/ringbuffer.go b/transmission/ringbuffer.go
new file mode 100644
index 0000000..7ac117b
--- /dev/null
+++ b/transmission/ringbuffer.go
@@ -0,0 +1,135 @@
+package transmission
+
+import (
+	"sync"
+	"time"
+
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+	"github.com/sirupsen/logrus"
+)
+
+type BufferEvent struct {
+	*event.Event
+	expires *time.Time
+}
+
+func (be *BufferEvent) expired() bool {
+	if be.expires != nil {
+		return be.expires.Before(time.Now())
+	}
+	return true
+}
+
+type RingBuffer struct {
+	sync.RWMutex
+
+	items     map[uint64]*BufferEvent
+	itemOrder []uint64
+	size      int
+	position  int
+	ttl       time.Duration
+}
+
+func NewRingBuffer(size int, duration time.Duration) *RingBuffer {
+	logrus.WithFields(logrus.Fields{
+		"size":    size,
+		"timeout": duration,
+	}).Info("Creating retry buffer.")
+
+	r := &RingBuffer{
+		items:     make(map[uint64]*BufferEvent),
+		itemOrder: make([]uint64, size),
+		size:      size,
+		ttl:       duration,
+	}
+
+	if r.enabled() && r.ttl > 0 {
+		r.startCleanupTimer()
+	}
+
+	return r
+}
+
+func (r *RingBuffer) enabled() bool {
+	return r.size > 0
+}
+
+// Add an event to ring buffer
+func (r *RingBuffer) Add(key uint64, ev *event.Event) {
+	// fast return if disabled
+	if !r.enabled() {
+		return
+	}
+
+	expiration := time.Now().Add(r.ttl)
+	be := &BufferEvent{
+		Event:   ev,
+		expires: &expiration,
+	}
+
+	r.Lock()
+	// check if we have an existing event at this position
+	oldKey := r.itemOrder[r.position]
+	if oldKey != 0 {
+		// If the key at the current position is associated with an event
+		// in the buffer, remove it. Otherwise, this delete is a NOOP.
+		delete(r.items, oldKey)
+	}
+
+	// add event to map and order array
+	r.items[key] = be
+	r.itemOrder[r.position] = key
+
+	// increment position
+	r.position++
+	if r.position >= r.size {
+		r.position = 0
+	}
+	r.Unlock()
+}
+
+// Get an event from the buffer's map
+func (r *RingBuffer) Get(key uint64) (*event.Event, bool) {
+	// fast return if disabled
+	if !r.enabled() {
+		return nil, false
+	}
+
+	r.RLock()
+	be, ok := r.items[key]
+	r.RUnlock()
+	if ok {
+		return be.Event, true
+	} else {
+		return nil, false
+	}
+}
+
+func (r *RingBuffer) cleanup() {
+	r.Lock()
+	beforeSize := len(r.items)
+	for key, be := range r.items {
+		if be.expired() {
+			delete(r.items, key)
+		}
+	}
+	afterSize := len(r.items)
+	r.Unlock()
+
+	logrus.WithFields(logrus.Fields{
+		"beforeSize": beforeSize,
+		"afterSize":  afterSize,
+	}).Debug("Retry buffer cleanup")
+}
+
+func (r *RingBuffer) startCleanupTimer() {
+	duration := r.ttl
+	if duration < time.Second {
+		duration = time.Second
+	}
+	go (func() {
+		for range time.Tick(duration) {
+			r.cleanup()
+		}
+	})()
+}
diff --git a/transmission/transmission.go b/transmission/transmission.go
index 20b14aa..cd4fed4 100644
--- a/transmission/transmission.go
+++ b/transmission/transmission.go
@@ -2,75 +2,152 @@ package transmission
 
 import (
 	"strings"
+	"sync/atomic"
+	"time"
 
 	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
-	libhoney "github.com/honeycombio/libhoney-go"
+	"github.com/honeycombio/libhoney-go"
 	"github.com/sirupsen/logrus"
 )
 
 const truncatedLineMax = 1000
 
+var (
+	eventCounter uint64
+	ringBuffer   *RingBuffer
+)
+
 type Transmitter interface {
 	Send(*event.Event)
 }
 
-type HoneycombTransmitter struct{}
+type HoneycombTransmitter struct {
+}
 
 func (ht *HoneycombTransmitter) Send(ev *event.Event) {
+
 	libhoneyEvent := libhoney.NewEvent()
-	libhoneyEvent.Metadata = ev.RawMessage
 	libhoneyEvent.Dataset = ev.Dataset
 	if ev.SampleRate != 0 {
 		libhoneyEvent.SampleRate = ev.SampleRate
 	}
 	libhoneyEvent.Timestamp = ev.Timestamp
-	libhoneyEvent.Add(ev.Data)
-	libhoneyEvent.SendPresampled()
+
+	err := libhoneyEvent.Add(ev.Data)
+	if err != nil {
+		// event isn't a proper type that can be added
+		logrus.WithFields(logrus.Fields{
+			"error":      err.Error(),
+			"rawMessage": ev.RawMessage,
+		}).Error("Unable to create libhoney event.")
+		return
+	}
+
+	// generate auto-incrementing value to be used as buffer key
+	key := atomic.AddUint64(&eventCounter, 1)
+	if key == 0 {
+		// if the counter is rolled back to 0, add 1 more
+		key = atomic.AddUint64(&eventCounter, 1)
+	}
+	libhoneyEvent.Metadata = key
+	ringBuffer.Add(key, ev)
+
+	// send event
+	err = libhoneyEvent.SendPresampled()
+	if err != nil {
+		// error while sending
+		logrus.WithFields(logrus.Fields{
+			"error":      err.Error(),
+			"rawMessage": ev.RawMessage,
+		}).Error("Unable to send libhoney event.")
+		return
+	}
 }
 
-func InitLibhoney(writeKey string, apiHost string) error {
+func InitLibhoney(apiKey string, apiHost string, bufferSize int, bufferExpire time.Duration) error {
 	err := libhoney.Init(libhoney.Config{
-		WriteKey: writeKey,
-		APIHost:  apiHost,
+		APIKey:      apiKey,
+		APIHost:     apiHost,
+		BlockOnSend: true,
 	})
 	if err != nil {
 		return err
 	}
+
+	ringBuffer = NewRingBuffer(bufferSize, bufferExpire)
+
 	go readResponses()
 	return nil
 }
 
-// TODO: support retrying, etc.
 func readResponses() {
-	for resp := range libhoney.Responses() {
-		// If an event is too large, include the first part of the raw message
+	for resp := range libhoney.TxResponses() {
 		if resp.Err != nil && strings.HasPrefix(resp.Err.Error(), "event exceeds max event size") {
-			rawMessage, ok := resp.Metadata.(string)
-			// if this is somehow not a string, handle the error normally below
+			// error sending event due to size, try to find it in cache
+			key, ok := resp.Metadata.(uint64)
+			if ok {
+				ev, exists := ringBuffer.Get(key)
+
+				if exists {
+					rawMessage := ev.RawMessage
+
+					// include the first part of the raw message
+					msgLen := len(rawMessage)
+					truncatedMsgLen := truncatedLineMax
+					// This shouldn't happen - if the event is too large for the API, it's at least
+					// 1000 bytes, but be defensive here anyway
+					if msgLen < truncatedMsgLen {
+						truncatedMsgLen = msgLen
+					}
+
+					logrus.WithFields(logrus.Fields{
+						"error":             resp.Err,
+						"status":            resp.StatusCode,
+						"responseBody":      string(resp.Body),
+						"truncated_raw_msg": rawMessage[:truncatedMsgLen],
+						"raw_msg_len":       msgLen,
+					}).Error("Unable to send event to Honeycomb API due to size")
+
+					continue
+				}
+			}
+
+			// if we get here we couldn't find the response metadata or event in send buffer
+			logrus.WithFields(logrus.Fields{
+				"error":        resp.Err,
+				"status":       resp.StatusCode,
+				"responseBody": string(resp.Body),
+			}).Error("Unable to send event to Honeycomb API due to size. Event not found in local send buffer.")
+
+		} else if resp.Err != nil || (resp.StatusCode != 200 && resp.StatusCode != 202) {
+			// error sending event, try to find it in buffer
+			key, ok := resp.Metadata.(uint64)
 			if ok {
-				msgLen := len(rawMessage)
-				truncatedMsgLen := truncatedLineMax
-				// This shouldn't happen - if the event is too large for the API, it's at least
-				// 1000 bytes, but be defensive here anyway
-				if msgLen < truncatedMsgLen {
-					truncatedMsgLen = msgLen
+				ev, exists := ringBuffer.Get(key)
+
+				if exists {
+					// event found, we can retry it
+					logrus.WithFields(logrus.Fields{
+						"error":        resp.Err,
+						"status":       resp.StatusCode,
+						"responseBody": string(resp.Body),
+					}).Debug("Failed to send event to Honeycomb. Will retry.")
+
+					// Resend event
+					ht := &HoneycombTransmitter{}
+					ht.Send(ev)
+
+					continue
 				}
-				logrus.WithFields(logrus.Fields{
-					"error":             resp.Err,
-					"status":            resp.StatusCode,
-					"responseBody":      string(resp.Body),
-					"truncated_raw_msg": rawMessage[:truncatedMsgLen],
-					"raw_msg_len":       msgLen,
-				}).Error("unable to send event to Honeycomb API due to size")
-				continue
 			}
-		}
-		if resp.Err != nil || (resp.StatusCode != 200 && resp.StatusCode != 202) {
+
+			// if we get here we couldn't find the response metadata or event in send buffer
+			// retry isn't possible
 			logrus.WithFields(logrus.Fields{
 				"error":        resp.Err,
 				"status":       resp.StatusCode,
 				"responseBody": string(resp.Body),
-			}).Error("Failed to send event to Honeycomb")
+			}).Error("Failed to send event to Honeycomb. Unable to retry.")
 		}
 	}
 }
@@ -78,4 +155,4 @@ func readResponses() {
 // NullTransmitter does nothing
 type NullTransmitter struct{}
 
-func (nt *NullTransmitter) Send(ev *event.Event) {}
+func (nt *NullTransmitter) Send(_ *event.Event) {}


Test patch:
diff --git a/transmission/ringbuffer_test.go b/transmission/ringbuffer_test.go
new file mode 100644
index 0000000..2c6deb6
--- /dev/null
+++ b/transmission/ringbuffer_test.go
@@ -0,0 +1,114 @@
+package transmission
+
+import (
+	"testing"
+	"time"
+
+	"github.com/honeycombio/honeycomb-kubernetes-agent/event"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGet(t *testing.T) {
+	rb := NewRingBuffer(100, 0)
+
+	assert.True(t, rb.enabled(), "A buffer of size greater than 0 is enabled")
+
+	for i := uint64(0); i < 10; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	ev, ok := rb.Get(5)
+	assert.True(t, ok, "Expected event wasn't found in the buffer")
+	assert.Equal(t, uint64(5), ev.Data["item"], "Event found didn't match expected data")
+
+	ev, ok = rb.Get(8)
+	assert.True(t, ok, "Expected event wasn't found in the buffer")
+	assert.Equal(t, uint64(8), ev.Data["item"], "Event found didn't match expected data")
+}
+
+func TestGetWhenDisabled(t *testing.T) {
+	rb := NewRingBuffer(0, 0)
+
+	assert.False(t, rb.enabled(), "A buffer of size 0 is not enabled")
+
+	for i := uint64(0); i < 10; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	ev, ok := rb.Get(5)
+	assert.False(t, ok, "Retrieving an event from a disabled buffer is not ok")
+	assert.Nil(t, ev, "A disabled buffer will return nil, not an event")
+}
+
+func TestRingOverflow(t *testing.T) {
+	rb := NewRingBuffer(100, 0)
+
+	for i := uint64(0); i < 100; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	for i := uint64(100); i < 110; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	_, ok := rb.Get(5)
+	assert.False(t, ok, "Found an event on the ring that should have been evicted by the size limit")
+
+	ev, ok := rb.Get(105)
+	assert.True(t, ok, "Expected event wasn't found in the buffer")
+	assert.Equal(t, uint64(105), ev.Data["item"], "Event found didn't match expected data")
+}
+
+func TestExpire(t *testing.T) {
+	rb := NewRingBuffer(100, 1*time.Second)
+
+	// this range of events will expire
+	for i := uint64(0); i < 10; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	// sleep for more than the TTL
+	time.Sleep(2 * time.Second)
+
+	// this range of events will be alive
+	for i := uint64(10); i < 20; i++ {
+		ev := &event.Event{
+			Data: map[string]interface{}{
+				"item": i,
+			},
+		}
+		rb.Add(i, ev)
+	}
+
+	_, ok := rb.Get(5)
+	assert.False(t, ok, "Found an event on the ring that should have expired")
+
+	ev, ok := rb.Get(15)
+	assert.True(t, ok, "Expected unexpired event wasn't found")
+	assert.Equal(t, uint64(15), ev.Data["item"], "Event found didn't match expected data")
+}