Add adapter for new plugins (#333)

* add vendor dependencies and update SignalFx golib
* add adapter for additional plugins

Author: charless-splunk

Gopkg.lock

@@ -74,7 +74,7 @@
 
 [[constraint]]
   name = "github.com/signalfx/golib"
-  revision = "8321403a729d6c8fa60a3dcf2c60bf20f71f6062"
+  revision = "b1be94c3ec8d1fdeccd62735b851489f0b3798e0"
 
 [[constraint]]
   name = "github.com/signalfx/metricproxy"
@@ -142,3 +142,7 @@
 [[constraint]]
   name = "github.com/Microsoft/go-winio"
   revision = "ab35fc04b6365e8fcb18e6e9e41ea4a02b10b175"
+
+[[constraint]]
+  name = "github.com/signalfx/telegraf"
+  revision = "df5485433c763fbe76fd8bfc6ad80a81e93bb9c5"

Gopkg.toml

@@ -0,0 +1,65 @@
+package accumulator
+
+import (
+	"time"
+
+	"github.com/signalfx/golib/datapoint"
+	"github.com/signalfx/signalfx-agent/internal/monitors/telegraf/common/emitter"
+)
+
+// Accumulator is an interface used to accumulate telegraf measurements from
+// Telegraf plugins.
+type Accumulator struct {
+	emit emitter.Emitter
+}
+
+// AddFields receives a measurement with tags and a time stamp to the accumulator.
+// Measurements are passed to the Accumulator's Emitter.
+func (ac *Accumulator) AddFields(measurement string, fields map[string]interface{},
+	tags map[string]string, t ...time.Time) {
+	ac.emit.Add(measurement, fields, tags, datapoint.Gauge, "untyped", t...)
+}
+
+// AddGauge receives a measurement as a "Gauge" with tags and a time stamp to
+// the accumulator. Measurements are passed to the Accumulator's Emitter.
+func (ac *Accumulator) AddGauge(measurement string, fields map[string]interface{},
+	tags map[string]string, t ...time.Time) {
+	ac.emit.Add(measurement, fields, tags, datapoint.Gauge, "", t...)
+}
+
+// AddCounter receives a measurement as a "Counter" with tags and a time stamp
+// to the accumulator. Measurements are passed to the Accumulator's Emitter.
+func (ac *Accumulator) AddCounter(measurement string, fields map[string]interface{},
+	tags map[string]string, t ...time.Time) {
+	ac.emit.Add(measurement, fields, tags, datapoint.Counter, "", t...)
+}
+
+// AddSummary receives a measurement as a "Counter" with tags and a time stamp
+// to the accumulator. Measurements are passed to the Accumulator's Emitter.
+func (ac *Accumulator) AddSummary(measurement string, fields map[string]interface{},
+	tags map[string]string, t ...time.Time) {
+	ac.emit.Add(measurement, fields, tags, datapoint.Gauge, "summary", t...)
+}
+
+// AddHistogram receives a measurement as a "Counter" with tags and a time stamp
+// to the accumulator. Measurements are passed to the Accumulator's Emitter.
+func (ac *Accumulator) AddHistogram(measurement string, fields map[string]interface{},
+	tags map[string]string, t ...time.Time) {
+	ac.emit.Add(measurement, fields, tags, datapoint.Gauge, "histogram", t...)
+}
+
+// SetPrecision - SignalFx does not implement this
+func (ac *Accumulator) SetPrecision(precision, interval time.Duration) {
+}
+
+// AddError - log an error returned by the plugin
+func (ac *Accumulator) AddError(err error) {
+	ac.emit.AddError(err)
+}
+
+// NewAccumulator returns a pointer to an Accumulator
+func NewAccumulator(e emitter.Emitter) *Accumulator {
+	return &Accumulator{
+		emit: e,
+	}
+}

internal/monitors/telegraf/common/accumulator/accumulator.go

@@ -0,0 +1,134 @@
+package accumulator
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/signalfx/golib/datapoint"
+)
+
+type testEmitter struct {
+	measurement        string
+	fields             map[string]interface{}
+	tags               map[string]string
+	metricType         datapoint.MetricType
+	originalMetricType string
+	t                  time.Time
+	err                error
+}
+
+func (e *testEmitter) Add(measurement string, fields map[string]interface{},
+	tags map[string]string, metricType datapoint.MetricType, originalMetricType string, t ...time.Time) {
+	e.measurement = measurement
+	e.fields = fields
+	e.tags = tags
+	e.metricType = metricType
+	e.originalMetricType = originalMetricType
+	e.t = t[0]
+}
+func (e *testEmitter) IncludeEvent(string)       {}
+func (e *testEmitter) IncludeEvents([]string)    {}
+func (e *testEmitter) ExcludeDatum(string)       {}
+func (e *testEmitter) ExcludeData([]string)      {}
+func (e *testEmitter) AddTag(string, string)     {}
+func (e *testEmitter) AddTags(map[string]string) {}
+func (e *testEmitter) OmitTag(string)            {}
+func (e *testEmitter) OmitTags([]string)         {}
+func (e *testEmitter) AddError(err error) {
+	e.err = err
+}
+
+func TestAccumulator(t *testing.T) {
+	ac := NewAccumulator(&testEmitter{})
+	tests := []struct {
+		name string
+		want *testEmitter
+		fn   func(string, map[string]interface{}, map[string]string, ...time.Time)
+	}{
+		{
+			name: "AddFields()",
+			want: &testEmitter{
+				measurement:        "field_measurement",
+				fields:             map[string]interface{}{"dim1": "dimval1"},
+				tags:               map[string]string{"tag1": "tagval1"},
+				metricType:         datapoint.Gauge,
+				originalMetricType: "untyped",
+				t:                  time.Now(),
+			},
+			fn: ac.AddFields,
+		},
+		{
+			name: "AddGauge()",
+			want: &testEmitter{
+				measurement:        "gauge_measurement",
+				fields:             map[string]interface{}{"dim1": "dimval1"},
+				tags:               map[string]string{"tag1": "tagval1"},
+				metricType:         datapoint.Gauge,
+				originalMetricType: "gauge",
+				t:                  time.Now(),
+			},
+			fn: ac.AddGauge,
+		},
+		{
+			name: "AddCounter()",
+			want: &testEmitter{
+				measurement:        "counter_measurement",
+				fields:             map[string]interface{}{"dim1": "dimval1"},
+				tags:               map[string]string{"tag1": "tagval1"},
+				metricType:         datapoint.Counter,
+				originalMetricType: "counter",
+				t:                  time.Now(),
+			},
+			fn: ac.AddCounter,
+		},
+		{
+			name: "AddSummary()",
+			want: &testEmitter{
+				measurement:        "summary_measurement",
+				fields:             map[string]interface{}{"dim1": "dimval1"},
+				tags:               map[string]string{"tag1": "tagval1"},
+				metricType:         datapoint.Gauge,
+				originalMetricType: "summary",
+				t:                  time.Now(),
+			},
+			fn: ac.AddSummary,
+		},
+		{
+			name: "AddHistogram()",
+			want: &testEmitter{
+				measurement:        "histogram_measurement",
+				fields:             map[string]interface{}{"dim1": "dimval1"},
+				tags:               map[string]string{"tag1": "tagval1"},
+				metricType:         datapoint.Gauge,
+				originalMetricType: "histogram",
+				t:                  time.Now(),
+			},
+			fn: ac.AddHistogram,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ac.emit = &testEmitter{}
+			tt.fn(tt.want.measurement, tt.want.fields, tt.want.tags, tt.want.t)
+			if ac.emit.(*testEmitter).measurement != tt.want.measurement ||
+				!reflect.DeepEqual(ac.emit.(*testEmitter).fields, tt.want.fields) ||
+				!reflect.DeepEqual(ac.emit.(*testEmitter).tags, tt.want.tags) {
+				t.Errorf("Accumulator_AddFields() = %v, want %v", ac.emit, tt.want)
+			}
+		})
+	}
+	t.Run("SetPrecision()", func(t *testing.T) {
+		ac.emit = &testEmitter{}
+		ac.SetPrecision(time.Second*1, time.Second*1)
+	})
+	t.Run("AddError()", func(t *testing.T) {
+		ac.emit = &testEmitter{}
+		err := fmt.Errorf("Test Error")
+		ac.AddError(err)
+		if ac.emit.(*testEmitter).err != err {
+			t.Errorf("AddError() = %v, want %v", ac.emit.(*testEmitter).err, err)
+		}
+	})
+}