Introduce telegrafreceiver (#935)

cmd/otelcontribcol/unstable_components_enabled.go

@@ -18,8 +18,12 @@ package main
 
 import (
 	"go.opentelemetry.io/collector/component"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/telegrafreceiver"
 )
 
 func extraReceivers() []component.ReceiverFactory {
-	return []component.ReceiverFactory{}
+	return []component.ReceiverFactory{
+		telegrafreceiver.NewFactory(),
+	}
 }

go.mod

@@ -63,6 +63,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/splunkhecreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/statsdreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/syslogreceiver v0.0.0-00010101000000-000000000000
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/telegrafreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/wavefrontreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/windowsperfcountersreceiver v0.0.0-00010101000000-000000000000
 	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/zookeeperreceiver v0.0.0-00010101000000-000000000000
@@ -73,6 +74,8 @@ require (
 
 replace go.opentelemetry.io/collector => github.com/SumoLogic/opentelemetry-collector v0.21.0-sumo
 
+replace github.com/influxdata/telegraf => github.com/sumologic/telegraf v1.17.3-sumo
+
 // Replace references to modules that are in this repository with their relateive paths
 // so that we always build with current (latest) version of the source code.
 
@@ -206,6 +209,8 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafka
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor => ./processor/groupbyattrsprocessor
 
+replace github.com/open-telemetry/opentelemetry-collector-contrib/receiver/telegrafreceiver => ./receiver/telegrafreceiver
+
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/cascadingfilterprocessor => ./processor/cascadingfilterprocessor
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbytraceprocessor => ./processor/groupbytraceprocessor

receiver/telegrafreceiver/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

receiver/telegrafreceiver/README.md

@@ -0,0 +1,57 @@
+# Telegraf Receiver
+
+Telegraf receiver for ingesting metrics from various [input plugins][input_plugins]
+into otc pipeline.
+
+Supported pipeline types: metrics
+
+Use case: user configures telegraf input plugins in config for ingestion and otc
+processors and exporters for data processing and export.
+
+> :construction: This receiver is currently in **BETA** and is considered **unstable**.
+
+[input_plugins]: https://github.com/influxdata/telegraf/tree/master/plugins/inputs
+
+## Configuration
+
+The following settings are required:
+
+- `agent_config`: Telegraf config. For now it allows to provide agent and input
+  plugins configuration. One can refer to
+  [telegraf configuration docs][telegraf_config_docs] for full list of
+  configuration options.
+
+The Following settings are optional:
+
+- `separate_field` (default value is `false`): Specify whether metric field
+  should be added separately as data point label.
+
+Example:
+
+```yaml
+receivers:
+  telegraf:
+    separate_field: false
+    agent_config: |
+      [agent]
+        interval = "2s"
+        flush_interval = "3s"
+      [[inputs.mem]]
+```
+
+The full list of settings exposed for this receiver are documented in
+[config.go](./config.go).
+
+[telegraf_config_docs]: https://github.com/influxdata/telegraf/blob/master/docs/CONFIGURATION.md
+
+## Limitations
+
+With its current implementation Telegraf receiver has the following limitations:
+
+- only input plugins can be configured in telegraf agent confugration section
+  (apart from agent's configuration itself). That means that metrics go straight
+  from input plugin to the receiver for translation (into otc data model) without
+  any processing
+- ony `telegraf.Gauge` metric data is supported, which translated (loosly) into
+  `pdata.MetricDataTypeDoubleGauge` and `pdata.MetricDataTypeIntGauge` depending
+  on the underlying data type

receiver/telegrafreceiver/all.go

@@ -0,0 +1,22 @@
+// Copyright 2021, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package telegrafreceiver
+
+import (
+	// _ "github.com/influxdata/telegraf/plugins/aggregators/all"
+	_ "github.com/influxdata/telegraf/plugins/inputs/all"
+	// _ "github.com/influxdata/telegraf/plugins/outputs/all"
+	// _ "github.com/influxdata/telegraf/plugins/processors/all"
+)

receiver/telegrafreceiver/config.go

@@ -0,0 +1,34 @@
+// Copyright 2021, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package telegrafreceiver
+
+import (
+	"go.opentelemetry.io/collector/config/configmodels"
+)
+
+// Config defines configuration for the telegraf receiver.
+type Config struct {
+	configmodels.ReceiverSettings `mapstructure:",squash"`
+
+	// AgentConfig is the yaml config used as telegraf configuration.
+	// Please note that only inputs should be configured as all metrics gathered
+	// by them will be passed through to otc pipeline for processing and export.
+	AgentConfig string `mapstructure:"agent_config"`
+
+	// SeparateField controls whether the ingested metrics should have a field
+	// concatenated with metric name like e.g. metric=mem_available or maybe rather
+	// have it as a separate label like e.g. metric=mem field=available
+	SeparateField bool `mapstructure:"separate_field"`
+}

receiver/telegrafreceiver/converter.go

@@ -0,0 +1,126 @@
+// Copyright 2021, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package telegrafreceiver
+
+import (
+	"fmt"
+
+	"github.com/influxdata/telegraf"
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+const (
+	fieldLabel = "field"
+)
+
+type MetricConverter interface {
+	Convert(telegraf.Metric) (pdata.Metrics, error)
+}
+
+type metricConverter struct {
+	separateField bool
+}
+
+func newConverter(separateField bool) MetricConverter {
+	return metricConverter{
+		separateField: separateField,
+	}
+}
+
+// Convert converts telegraf.Metric to pdata.Metrics.
+func (mc metricConverter) Convert(m telegraf.Metric) (pdata.Metrics, error) {
+	ms := pdata.NewMetrics()
+	rms := ms.ResourceMetrics()
+	rms.Resize(1)
+	rm := rms.At(0)
+
+	// Attach tags as attributes - pipe through the metadata
+	for _, t := range m.TagList() {
+		rm.Resource().Attributes().InsertString(t.Key, t.Value)
+	}
+
+	rm.InstrumentationLibraryMetrics().Resize(1)
+	ilm := rm.InstrumentationLibraryMetrics().At(0)
+
+	il := ilm.InstrumentationLibrary()
+	il.SetName(typeStr)
+	il.SetVersion(versionStr)
+
+	tim := m.Time().UnixNano()
+
+	metrics := ilm.Metrics()
+
+	switch t := m.Type(); t {
+	case telegraf.Gauge:
+		for _, f := range m.FieldList() {
+			pm := pdata.NewMetric()
+
+			if mc.separateField {
+				pm.SetName(m.Name())
+			} else {
+				pm.SetName(m.Name() + "_" + f.Key)
+			}
+
+			switch v := f.Value.(type) {
+			case float64:
+				pm.SetDataType(pdata.MetricDataTypeDoubleGauge)
+				dps := pm.DoubleGauge().DataPoints()
+				dps.Resize(1)
+				dp := dps.At(0)
+				dp.SetValue(v)
+				dp.SetTimestamp(pdata.TimestampUnixNano(tim))
+				if mc.separateField {
+					dp.LabelsMap().Insert(fieldLabel, f.Key)
+				}
+
+			case int64, uint64:
+				pm.SetDataType(pdata.MetricDataTypeIntGauge)
+				dps := pm.IntGauge().DataPoints()
+				dps.Resize(1)
+				dp := dps.At(0)
+				switch vv := v.(type) {
+				case int64:
+					dp.SetValue(vv)
+				case uint64:
+					dp.SetValue(int64(vv))
+				}
+
+				dp.SetTimestamp(pdata.TimestampUnixNano(tim))
+				if mc.separateField {
+					dp.LabelsMap().Insert(fieldLabel, f.Key)
+				}
+
+			default:
+				return pdata.Metrics{},
+					fmt.Errorf("unknown data type in telegraf.Gauge metric: %T", v)
+			}
+			metrics.Append(pm)
+		}
+
+	case telegraf.Counter:
+		return pdata.Metrics{}, fmt.Errorf("unsupported metric type: telegraf.Counter")
+	case telegraf.Untyped:
+		return pdata.Metrics{}, fmt.Errorf("unsupported metric type: telegraf.Untyped")
+	case telegraf.Summary:
+		return pdata.Metrics{}, fmt.Errorf("unsupported metric type: telegraf.Summary")
+	case telegraf.Histogram:
+		return pdata.Metrics{}, fmt.Errorf("unsupported metric type: telegraf.Histogram")
+
+	default:
+		return pdata.Metrics{}, fmt.Errorf("unknown metric type: %T", t)
+	}
+
+	return ms, nil
+}