Implement telegraf metric types

And use them in the prometheus output plugin.

Still need to test the prometheus output plugin.

Also need to actually create typed metrics in the system plugins.

closes #1683

metric.go

@@ -6,6 +6,17 @@ import (
 	"github.com/influxdata/influxdb/client/v2"
 )
 
+// ValueType is an enumeration of metric types that represent a simple value.
+type ValueType int
+
+// Possible values for the ValueType enum.
+const (
+	_ ValueType = iota
+	Counter
+	Gauge
+	Untyped
+)
+
 type Metric interface {
 	// Name returns the measurement name of the metric
 	Name() string
@@ -16,6 +27,9 @@ type Metric interface {
 	// Time return the timestamp for the metric
 	Time() time.Time
 
+	// Type returns the metric type. Can be either telegraf.Gauge or telegraf.Counter
+	Type() ValueType
+
 	// UnixNano returns the unix nano time of the metric
 	UnixNano() int64
 
@@ -35,12 +49,11 @@ type Metric interface {
 // metric is a wrapper of the influxdb client.Point struct
 type metric struct {
 	pt *client.Point
+
+	mType ValueType
 }
 
-// NewMetric returns a metric with the given timestamp. If a timestamp is not
+// NewMetric returns an untyped metric.
-// given, then data is sent to the database without a timestamp, in which case
-// the server will assign local time upon reception. NOTE: it is recommended to
-// send data with a timestamp.
 func NewMetric(
 	name string,
 	tags map[string]string,
@@ -52,7 +65,46 @@ func NewMetric(
 		return nil, err
 	}
 	return &metric{
-		pt: pt,
+		pt:    pt,
+		mType: Untyped,
+	}, nil
+}
+
+// NewGaugeMetric returns a gauge metric.
+// Gauge metrics should be used when the metric is can arbitrarily go up and
+// down. ie, temperature, memory usage, cpu usage, etc.
+func NewGaugeMetric(
+	name string,
+	tags map[string]string,
+	fields map[string]interface{},
+	t time.Time,
+) (Metric, error) {
+	pt, err := client.NewPoint(name, tags, fields, t)
+	if err != nil {
+		return nil, err
+	}
+	return &metric{
+		pt:    pt,
+		mType: Gauge,
+	}, nil
+}
+
+// NewCounterMetric returns a Counter metric.
+// Counter metrics should be used when the metric being created is an
+// always-increasing counter. ie, net bytes received, requests served, errors, etc.
+func NewCounterMetric(
+	name string,
+	tags map[string]string,
+	fields map[string]interface{},
+	t time.Time,
+) (Metric, error) {
+	pt, err := client.NewPoint(name, tags, fields, t)
+	if err != nil {
+		return nil, err
+	}
+	return &metric{
+		pt:    pt,
+		mType: Counter,
 	}, nil
 }
 
@@ -68,6 +120,10 @@ func (m *metric) Time() time.Time {
 	return m.pt.Time()
 }
 
+func (m *metric) Type() ValueType {
+	return m.mType
+}
+
 func (m *metric) UnixNano() int64 {
 	return m.pt.UnixNano()
 }

metric_test.go

@@ -23,6 +23,51 @@ func TestNewMetric(t *testing.T) {
 	m, err := NewMetric("cpu", tags, fields, now)
 	assert.NoError(t, err)
 
+	assert.Equal(t, Untyped, m.Type())
+	assert.Equal(t, tags, m.Tags())
+	assert.Equal(t, fields, m.Fields())
+	assert.Equal(t, "cpu", m.Name())
+	assert.Equal(t, now, m.Time())
+	assert.Equal(t, now.UnixNano(), m.UnixNano())
+}
+
+func TestNewGaugeMetric(t *testing.T) {
+	now := time.Now()
+
+	tags := map[string]string{
+		"host":       "localhost",
+		"datacenter": "us-east-1",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": float64(99),
+		"usage_busy": float64(1),
+	}
+	m, err := NewGaugeMetric("cpu", tags, fields, now)
+	assert.NoError(t, err)
+
+	assert.Equal(t, Gauge, m.Type())
+	assert.Equal(t, tags, m.Tags())
+	assert.Equal(t, fields, m.Fields())
+	assert.Equal(t, "cpu", m.Name())
+	assert.Equal(t, now, m.Time())
+	assert.Equal(t, now.UnixNano(), m.UnixNano())
+}
+
+func TestNewCounterMetric(t *testing.T) {
+	now := time.Now()
+
+	tags := map[string]string{
+		"host":       "localhost",
+		"datacenter": "us-east-1",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": float64(99),
+		"usage_busy": float64(1),
+	}
+	m, err := NewCounterMetric("cpu", tags, fields, now)
+	assert.NoError(t, err)
+
+	assert.Equal(t, Counter, m.Type())
 	assert.Equal(t, tags, m.Tags())
 	assert.Equal(t, fields, m.Fields())
 	assert.Equal(t, "cpu", m.Name())

plugins/outputs/prometheus_client/prometheus_client.go

@@ -102,6 +102,7 @@ func (p *PrometheusClient) Write(metrics []telegraf.Metric) error {
 		key := point.Name()
 		key = invalidNameCharRE.ReplaceAllString(key, "_")
 
+		// convert tags into prometheus labels
 		var labels []string
 		l := prometheus.Labels{}
 		for k, v := range point.Tags() {
@@ -113,6 +114,17 @@ func (p *PrometheusClient) Write(metrics []telegraf.Metric) error {
 			l[k] = v
 		}
 
+		// Get a type if it's available, defaulting to Untyped
+		var mType prometheus.ValueType
+		switch point.Type() {
+		case telegraf.Counter:
+			mType = prometheus.CounterValue
+		case telegraf.Gauge:
+			mType = prometheus.GaugeValue
+		default:
+			mType = prometheus.UntypedValue
+		}
+
 		for n, val := range point.Fields() {
 			// Ignore string and bool fields.
 			switch val.(type) {
@@ -134,11 +146,13 @@ func (p *PrometheusClient) Write(metrics []telegraf.Metric) error {
 			desc := prometheus.NewDesc(mname, "Telegraf collected metric", nil, l)
 			var metric prometheus.Metric
 			var err error
+
+			// switch for field type
 			switch val := val.(type) {
 			case int64:
-				metric, err = prometheus.NewConstMetric(desc, prometheus.UntypedValue, float64(val))
+				metric, err = prometheus.NewConstMetric(desc, mType, float64(val))
 			case float64:
-				metric, err = prometheus.NewConstMetric(desc, prometheus.UntypedValue, val)
+				metric, err = prometheus.NewConstMetric(desc, mType, val)
 			default:
 				continue
 			}