feat(agent): output buffer persistence

accumulator.go

@@ -57,6 +57,10 @@ type Accumulator interface {
 // TrackingID uniquely identifies a tracked metric group
 type TrackingID uint64
 
+type TrackingData interface {
+	ID() TrackingID
+}
+
 // DeliveryInfo provides the results of a delivered metric group.
 type DeliveryInfo interface {
 	// ID is the TrackingID

agent/agent.go

@@ -14,6 +14,7 @@ import (
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/internal/snmp"
+	"github.com/influxdata/telegraf/metric"
 	"github.com/influxdata/telegraf/models"
 	"github.com/influxdata/telegraf/plugins/processors"
 	"github.com/influxdata/telegraf/plugins/serializers/influx"
@@ -127,6 +128,7 @@ func (a *Agent) Run(ctx context.Context) error {
 	startTime := time.Now()
 
 	log.Printf("D! [agent] Connecting outputs")
+	metric.Init()
 	next, ou, err := a.startOutputs(ctx, a.Config.Outputs)
 	if err != nil {
 		return err
@@ -871,12 +873,12 @@ func (a *Agent) runOutputs(
 		}(output)
 	}
 
-	for metric := range unit.src {
+	for m := range unit.src {
 		for i, output := range unit.outputs {
 			if i == len(a.Config.Outputs)-1 {
-				output.AddMetric(metric)
+				output.AddMetric(m)
 			} else {
-				output.AddMetric(metric.Copy())
+				output.AddMetric(m.Copy())
 			}
 		}
 	}

config/config.go

@@ -278,6 +278,9 @@ type AgentConfig struct {
 	// Number of attempts to obtain a remote configuration via a URL during
 	// startup. Set to -1 for unlimited attempts.
 	ConfigURLRetryAttempts int `toml:"config_url_retry_attempts"`
+
+	BufferStrategy  string `toml:"buffer_strategy"`
+	BufferDirectory string `toml:"buffer_directory"`
 }
 
 // InputNames returns a list of strings of the configured inputs.
@@ -1521,6 +1524,8 @@ func (c *Config) buildOutput(name string, tbl *ast.Table) (*models.OutputConfig,
 	c.getFieldString(tbl, "name_suffix", &oc.NameSuffix)
 	c.getFieldString(tbl, "name_prefix", &oc.NamePrefix)
 	c.getFieldString(tbl, "startup_error_behavior", &oc.StartupErrorBehavior)
+	c.getFieldString(tbl, "buffer_strategy", &oc.BufferStrategy)
+	c.getFieldString(tbl, "buffer_directory", &oc.BufferDirectory)
 
 	if c.hasErrs() {
 		return nil, c.firstErr()

docs/LICENSE_OF_DEPENDENCIES.md

@@ -349,6 +349,8 @@ following works:
 - github.com/tidwall/gjson [MIT License](https://github.com/tidwall/gjson/blob/master/LICENSE)
 - github.com/tidwall/match [MIT License](https://github.com/tidwall/match/blob/master/LICENSE)
 - github.com/tidwall/pretty [MIT License](https://github.com/tidwall/pretty/blob/master/LICENSE)
+- github.com/tidwall/tinylru [MIT License](https://github.com/tidwall/tinylru/blob/master/LICENSE)
+- github.com/tidwall/wal [MIT License](https://github.com/tidwall/wal/blob/master/LICENSE)
 - github.com/tinylib/msgp [MIT License](https://github.com/tinylib/msgp/blob/master/LICENSE)
 - github.com/tklauser/go-sysconf [BSD 3-Clause "New" or "Revised" License](https://github.com/tklauser/go-sysconf/blob/master/LICENSE)
 - github.com/tklauser/numcpus [Apache License 2.0](https://github.com/tklauser/numcpus/blob/master/LICENSE)

go.mod

@@ -185,6 +185,7 @@ require (
 	github.com/testcontainers/testcontainers-go/modules/kafka v0.30.0
 	github.com/thomasklein94/packer-plugin-libvirt v0.5.0
 	github.com/tidwall/gjson v1.17.0
+	github.com/tidwall/wal v1.1.7
 	github.com/tinylib/msgp v1.1.9
 	github.com/urfave/cli/v2 v2.27.1
 	github.com/vapourismo/knx-go v0.0.0-20240217175130-922a0d50c241
@@ -451,6 +452,7 @@ require (
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
+	github.com/tidwall/tinylru v1.1.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.13 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/twmb/murmur3 v1.1.7 // indirect
@@ -483,7 +485,7 @@ require (
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.24.0 // indirect
+	go.uber.org/zap v1.25.0 // indirect
 	golang.org/x/exp v0.0.0-20240529005216-23cca8864a10 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect

go.sum

@@ -2222,12 +2222,17 @@ github.com/testcontainers/testcontainers-go/modules/kafka v0.30.0 h1:lQx20102vAH
 github.com/testcontainers/testcontainers-go/modules/kafka v0.30.0/go.mod h1:n3m3SH0ivwFZbehY8fgTLADfwSPK2ZC5za4r9nYYm4Q=
 github.com/thomasklein94/packer-plugin-libvirt v0.5.0 h1:aj2HLHZZM/ClGLIwVp9rrgh+2TOU/w4EiaZHAwCpOgs=
 github.com/thomasklein94/packer-plugin-libvirt v0.5.0/go.mod h1:GwN82FQ6KxCNKtS8LNUgLbwTZs90GGhBzCmTNkrTCrY=
+github.com/tidwall/gjson v1.10.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/gjson v1.17.0 h1:/Jocvlh98kcTfpN2+JzGQWQcqrPQwDrVEMApx/M5ZwM=
 github.com/tidwall/gjson v1.17.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/tinylru v1.1.0 h1:XY6IUfzVTU9rpwdhKUF6nQdChgCdGjkMfLzbWyiau6I=
+github.com/tidwall/tinylru v1.1.0/go.mod h1:3+bX+TJ2baOLMWTnlyNWHh4QMnFyARg2TLTQ6OFbzw8=
+github.com/tidwall/wal v1.1.7 h1:emc1TRjIVsdKKSnpwGBAcsAGg0767SvUk8+ygx7Bb+4=
+github.com/tidwall/wal v1.1.7/go.mod h1:r6lR1j27W9EPalgHiB7zLJDYu3mzW5BQP5KrzBpYY/E=
 github.com/tinylib/msgp v1.1.9 h1:SHf3yoO2sGA0veCJeCBYLHuttAVFHGm2RHgNodW7wQU=
 github.com/tinylib/msgp v1.1.9/go.mod h1:BCXGB54lDD8qUEPmiG0cQQUANC4IUQyB2ItS2UDlO/k=
 github.com/tj/assert v0.0.0-20171129193455-018094318fb0/go.mod h1:mZ9/Rh9oLWpLLDRpvE+3b7gP/C2YyLFYxNmcLnPTMe0=
@@ -2392,8 +2397,8 @@ go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9E
 go.uber.org/zap v1.9.1/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.13.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM=
-go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
-go.uber.org/zap v1.24.0/go.mod h1:2kMP+WWQ8aoFoedH3T2sq6iJ2yDWpHbP0f6MQbS9Gkg=
+go.uber.org/zap v1.25.0 h1:4Hvk6GtkucQ790dqmj7l1eEnRdKm3k3ZUrUMS2d5+5c=
+go.uber.org/zap v1.25.0/go.mod h1:JIAUzQIH94IC4fOJQm7gMmBJP5k7wQfdcnYdPoEXJYk=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

metric.go

@@ -149,5 +149,6 @@ type UnwrappableMetric interface {
 type TrackingMetric interface {
 	// TrackingID returns the ID used for tracking the metric
 	TrackingID() TrackingID
+	TrackingData() TrackingData
 	UnwrappableMetric
 }

metric/deserialize.go

@@ -0,0 +1,78 @@
+package metric
+
+import (
+	"bytes"
+	"encoding/gob"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/influxdata/telegraf"
+)
+
+// storage for tracking data that can't be serialized to disk
+var (
+	// todo need some way to empty this map out when done with a tracking ID.
+	// grouped tracking metrics means that ID->Data association is not one to one,
+	// many metrics could be associated with one tracking ID so we cannot just
+	// clear this every time in FromBytes.
+	trackingStore = make(map[telegraf.TrackingID]telegraf.TrackingData)
+	mu            = sync.Mutex{}
+
+	// ErrSkipTracking indicates that tracking information could not be found after
+	// deserializing a metric from bytes. In this case we should skip the metric
+	// and continue as if it does not exist.
+	ErrSkipTracking = errors.New("metric tracking data not found")
+)
+
+type serializedMetric struct {
+	M   telegraf.Metric
+	TID telegraf.TrackingID
+}
+
+func ToBytes(m telegraf.Metric) ([]byte, error) {
+	var sm serializedMetric
+	if um, ok := m.(telegraf.UnwrappableMetric); ok {
+		sm.M = um.Unwrap()
+	} else {
+		sm.M = m
+	}
+
+	if tm, ok := m.(telegraf.TrackingMetric); ok {
+		sm.TID = tm.TrackingID()
+
+		mu.Lock()
+		trackingStore[sm.TID] = tm.TrackingData()
+		mu.Unlock()
+	}
+
+	var buf bytes.Buffer
+	encoder := gob.NewEncoder(&buf)
+	if err := encoder.Encode(&sm); err != nil {
+		return nil, fmt.Errorf("failed to encode metric to bytes: %w", err)
+	}
+	return buf.Bytes(), nil
+}
+
+func FromBytes(b []byte) (telegraf.Metric, error) {
+	buf := bytes.NewBuffer(b)
+	decoder := gob.NewDecoder(buf)
+
+	var sm *serializedMetric
+	if err := decoder.Decode(&sm); err != nil {
+		return nil, fmt.Errorf("failed to decode metric from bytes: %w", err)
+	}
+
+	m := sm.M
+	if sm.TID != 0 {
+		mu.Lock()
+		td := trackingStore[sm.TID]
+		mu.Unlock()
+
+		if td == nil {
+			return nil, ErrSkipTracking
+		}
+		m = rebuildTrackingMetric(m, td)
+	}
+	return m, nil
+}

metric/init.go

@@ -0,0 +1,7 @@
+package metric
+
+import "encoding/gob"
+
+func Init() {
+	gob.RegisterName("metric.metric", &metric{})
+}

metric/tracking.go

@@ -33,35 +33,36 @@ func newTrackingID() telegraf.TrackingID {
 }
 
 type trackingData struct {
-	id          telegraf.TrackingID
-	rc          int32
-	acceptCount int32
-	rejectCount int32
+	//nolint:revive // method is already named ID
+	Id          telegraf.TrackingID
+	Rc          int32
+	AcceptCount int32
+	RejectCount int32
 	notifyFunc  NotifyFunc
 }
 
 func (d *trackingData) incr() {
-	atomic.AddInt32(&d.rc, 1)
+	atomic.AddInt32(&d.Rc, 1)
 }
 
 func (d *trackingData) decr() int32 {
-	return atomic.AddInt32(&d.rc, -1)
+	return atomic.AddInt32(&d.Rc, -1)
 }
 
 func (d *trackingData) accept() {
-	atomic.AddInt32(&d.acceptCount, 1)
+	atomic.AddInt32(&d.AcceptCount, 1)
 }
 
 func (d *trackingData) reject() {
-	atomic.AddInt32(&d.rejectCount, 1)
+	atomic.AddInt32(&d.RejectCount, 1)
 }
 
 func (d *trackingData) notify() {
 	d.notifyFunc(
 		&deliveryInfo{
-			id:       d.id,
-			accepted: int(d.acceptCount),
-			rejected: int(d.rejectCount),
+			id:       d.Id,
+			accepted: int(d.AcceptCount),
+			rejected: int(d.RejectCount),
 		},
 	)
 }
@@ -75,26 +76,33 @@ func newTrackingMetric(metric telegraf.Metric, fn NotifyFunc) (telegraf.Metric,
 	m := &trackingMetric{
 		Metric: metric,
 		d: &trackingData{
-			id:          newTrackingID(),
-			rc:          1,
-			acceptCount: 0,
-			rejectCount: 0,
+			Id:          newTrackingID(),
+			Rc:          1,
+			AcceptCount: 0,
+			RejectCount: 0,
 			notifyFunc:  fn,
 		},
 	}
 
 	if finalizer != nil {
 		runtime.SetFinalizer(m.d, finalizer)
 	}
-	return m, m.d.id
+	return m, m.d.Id
+}
+
+func rebuildTrackingMetric(metric telegraf.Metric, td telegraf.TrackingData) telegraf.Metric {
+	return &trackingMetric{
+		Metric: metric,
+		d:      td.(*trackingData),
+	}
 }
 
 func newTrackingMetricGroup(group []telegraf.Metric, fn NotifyFunc) ([]telegraf.Metric, telegraf.TrackingID) {
 	d := &trackingData{
-		id:          newTrackingID(),
-		rc:          0,
-		acceptCount: 0,
-		rejectCount: 0,
+		Id:          newTrackingID(),
+		Rc:          0,
+		AcceptCount: 0,
+		RejectCount: 0,
 		notifyFunc:  fn,
 	}
 
@@ -114,7 +122,7 @@ func newTrackingMetricGroup(group []telegraf.Metric, fn NotifyFunc) ([]telegraf.
 		d.notify()
 	}
 
-	return group, d.id
+	return group, d.Id
 }
 
 func (m *trackingMetric) Copy() telegraf.Metric {
@@ -152,7 +160,11 @@ func (m *trackingMetric) decr() {
 
 // Unwrap allows to access the underlying metric directly e.g. for go-templates
 func (m *trackingMetric) TrackingID() telegraf.TrackingID {
-	return m.d.id
+	return m.d.ID()
+}
+
+func (m *trackingMetric) TrackingData() telegraf.TrackingData {
+	return m.d
 }
 
 // Unwrap allows to access the underlying metric directly e.g. for go-templates
@@ -173,3 +185,7 @@ func (r *deliveryInfo) ID() telegraf.TrackingID {
 func (r *deliveryInfo) Delivered() bool {
 	return r.rejected == 0
 }
+
+func (d *trackingData) ID() telegraf.TrackingID {
+	return d.Id
+}