diff --git a/go.mod b/go.mod index 0776bc758d20b..5ff4e43f4b222 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,8 @@ go 1.26.0 godebug x509negativeserial=1 +replace github.com/influxdata/influxdb-observability/influx2otel => github.com/mchrome/influxdb-observability/influx2otel v0.0.0-20260623155732-db200540d054 + require ( cloud.google.com/go/auth v0.20.0 cloud.google.com/go/bigquery v1.77.0 diff --git a/go.sum b/go.sum index 78e478c30f1a5..790d0f270d2f0 100644 --- a/go.sum +++ b/go.sum @@ -1661,8 +1661,6 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/influxdata/influxdb-observability/common v0.5.12 h1:4YwZ+vsodz6VfoiX+ZqVotmnyCa9vCCPksSBK/WLjBs= github.com/influxdata/influxdb-observability/common v0.5.12/go.mod h1:u+CABnGO/F1IK51pDlZQroh4+igJNo695XrbLGDBhVc= -github.com/influxdata/influxdb-observability/influx2otel v0.5.12 h1:u0lNE3+63rILk4mtmCYsNyczH/1wEXnM+1aBzBe5akk= -github.com/influxdata/influxdb-observability/influx2otel v0.5.12/go.mod h1:bM407XIJYnrJYJ9Q3q2ytDSOyFhiYmGm0Sz1Qf48RPk= github.com/influxdata/influxdb-observability/otel2influx v0.5.12 h1:t9gmVOOHbZyEAvIYSoO97Tde1KArVtiYdM0/0Dhmuio= github.com/influxdata/influxdb-observability/otel2influx v0.5.12/go.mod h1:YGsb8xYfjHvcr2y0+Nj7kOHMTw7fWDbAA4g/qJKkvaU= github.com/influxdata/line-protocol-corpus v0.0.0-20210519164801-ca6fa5da0184/go.mod h1:03nmhxzZ7Xk2pdG+lmMd7mHDfeVOYFyhOgwO61qWU98= @@ -1866,6 +1864,8 @@ github.com/mattn/go-runewidth v0.0.21 h1:jJKAZiQH+2mIinzCJIaIG9Be1+0NR+5sz/lYEEj github.com/mattn/go-runewidth v0.0.21/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= github.com/mattn/go-sqlite3 v1.14.14/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/mchrome/influxdb-observability/influx2otel v0.0.0-20260623155732-db200540d054 h1:WXWL5QDbST9SMghx58SgxKadDTS1YqFyngMS+7BHnPk= +github.com/mchrome/influxdb-observability/influx2otel v0.0.0-20260623155732-db200540d054/go.mod h1:YgNF+ESSKXizCNRvkIAFMS8rWaP8t/86GLuycJ7vOTw= github.com/mdlayher/apcupsd v0.0.0-20220319200143-473c7b5f3c6a h1:JOlLsLUQnokTyWWwEvOVoKH3XUl6oDMP8jisO54l6J8= github.com/mdlayher/apcupsd v0.0.0-20220319200143-473c7b5f3c6a/go.mod h1:960H6oqSawdujauTeLX9BOx+ZdYX0TdG9xE9br5bino= github.com/mdlayher/ethtool v0.0.0-20210210192532-2b88debcdd43/go.mod h1:+t7E0lkKfbBsebllff1xdTmyJt8lH37niI6kwFk9OTo= diff --git a/plugins/outputs/opentelemetry/README.md b/plugins/outputs/opentelemetry/README.md index e3322cc10103d..adceb2e4ee321 100644 --- a/plugins/outputs/opentelemetry/README.md +++ b/plugins/outputs/opentelemetry/README.md @@ -50,6 +50,9 @@ plugin ordering. See [CONFIGURATION.md][CONFIGURATION.md] for more details. ## Supports: "gzip", "none" # compression = "gzip" + ## Characters for separating LineProtocol's metric name and field when converting to OpenTelemetry's metric name + # name_separator = "_" + ## Optional proxy settings for HTTP endpoints (service_address starts with http:// or https://) # use_system_proxy = false # http_proxy_url = "" diff --git a/plugins/outputs/opentelemetry/opentelemetry.go b/plugins/outputs/opentelemetry/opentelemetry.go index ccaed80960a99..11b4017e2180c 100644 --- a/plugins/outputs/opentelemetry/opentelemetry.go +++ b/plugins/outputs/opentelemetry/opentelemetry.go @@ -34,12 +34,13 @@ type OpenTelemetry struct { EncodingType string `toml:"encoding_type"` tls.ClientConfig - Timeout config.Duration `toml:"timeout"` - Compression string `toml:"compression"` - Token config.Secret `toml:"token"` - Headers map[string]string `toml:"headers"` - Attributes map[string]string `toml:"attributes"` - Coralogix *CoralogixConfig `toml:"coralogix"` + Timeout config.Duration `toml:"timeout"` + Compression string `toml:"compression"` + NameSeparator string `toml:"name_separator"` + Token config.Secret `toml:"token"` + Headers map[string]string `toml:"headers"` + Attributes map[string]string `toml:"attributes"` + Coralogix *CoralogixConfig `toml:"coralogix"` proxy.HTTPProxy proxy.TCPProxy @@ -105,7 +106,7 @@ func (o *OpenTelemetry) Connect() error { o.Headers["Authorization"] = "Bearer " + o.Coralogix.PrivateKey } - metricsConverter, err := influx2otel.NewLineProtocolToOtelMetrics(logger) + metricsConverter, err := influx2otel.NewLineProtocolToOtelMetricsWithSeparator(logger, o.NameSeparator) if err != nil { return err } @@ -235,6 +236,7 @@ const ( defaultServiceAddress = "localhost:4317" defaultTimeout = config.Duration(5 * time.Second) defaultCompression = "gzip" + defaultNameSeparator = "_" ) func init() { @@ -243,6 +245,7 @@ func init() { ServiceAddress: defaultServiceAddress, Timeout: defaultTimeout, Compression: defaultCompression, + NameSeparator: defaultNameSeparator, } }) } diff --git a/plugins/outputs/opentelemetry/opentelemetry_test.go b/plugins/outputs/opentelemetry/opentelemetry_test.go index b74ba43a7ed46..81f531563b668 100644 --- a/plugins/outputs/opentelemetry/opentelemetry_test.go +++ b/plugins/outputs/opentelemetry/opentelemetry_test.go @@ -9,8 +9,6 @@ import ( "testing" "time" - "github.com/influxdata/influxdb-observability/common" - "github.com/influxdata/influxdb-observability/influx2otel" "github.com/stretchr/testify/require" "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/pmetric" @@ -45,20 +43,18 @@ func TestOpenTelemetry(t *testing.T) { m := newMockOtelService(t) t.Cleanup(m.Cleanup) - metricsConverter, err := influx2otel.NewLineProtocolToOtelMetrics(common.NoopLogger{}) - require.NoError(t, err) plugin := &OpenTelemetry{ - ServiceAddress: m.Address(), - Timeout: config.Duration(time.Second), - Headers: map[string]string{"test": "header1"}, - Attributes: map[string]string{"attr-key": "attr-val"}, - metricsConverter: metricsConverter, + ServiceAddress: m.Address(), + Timeout: config.Duration(time.Second), + Headers: map[string]string{"test": "header1"}, + Attributes: map[string]string{"attr-key": "attr-val"}, otlpMetricClient: &gRPCClient{ grpcClientConn: m.GrpcClient(), metricsServiceClient: pmetricotlp.NewGRPCClient(m.GrpcClient()), }, Log: testutil.Logger{}, } + require.NoError(t, plugin.Connect()) input := metric.New( "cpu_temp", @@ -145,17 +141,13 @@ func TestOpenTelemetryHTTPProtobuf(t *testing.T) { })) defer server.Close() - metricsConverter, err := influx2otel.NewLineProtocolToOtelMetrics(common.NoopLogger{}) - require.NoError(t, err) - plugin := &OpenTelemetry{ - ServiceAddress: server.URL, - EncodingType: "protobuf", - Timeout: config.Duration(time.Second), - Attributes: map[string]string{"attr-key": "attr-val"}, - Compression: "none", - metricsConverter: metricsConverter, - Log: testutil.Logger{}, + ServiceAddress: server.URL, + EncodingType: "protobuf", + Timeout: config.Duration(time.Second), + Attributes: map[string]string{"attr-key": "attr-val"}, + Compression: "none", + Log: testutil.Logger{}, } require.NoError(t, plugin.Connect()) @@ -210,6 +202,7 @@ func TestOpenTelemetryHTTPJSON(t *testing.T) { body, err := io.ReadAll(r.Body) if err != nil { t.Error(err) + w.WriteHeader(http.StatusInternalServerError) return } @@ -243,17 +236,13 @@ func TestOpenTelemetryHTTPJSON(t *testing.T) { })) defer server.Close() - metricsConverter, err := influx2otel.NewLineProtocolToOtelMetrics(common.NoopLogger{}) - require.NoError(t, err) - plugin := &OpenTelemetry{ - ServiceAddress: server.URL, - EncodingType: "json", - Timeout: config.Duration(time.Second), - Attributes: map[string]string{"attr-key": "attr-val"}, - Compression: "none", - metricsConverter: metricsConverter, - Log: testutil.Logger{}, + ServiceAddress: server.URL, + EncodingType: "json", + Timeout: config.Duration(time.Second), + Attributes: map[string]string{"attr-key": "attr-val"}, + Compression: "none", + Log: testutil.Logger{}, } require.NoError(t, plugin.Connect()) @@ -284,6 +273,322 @@ func TestOpenTelemetryHTTPJSON(t *testing.T) { require.JSONEq(t, string(expectJSON), string(gotJSON)) } +func TestOpenTelemetrySeparator(t *testing.T) { + expect := pmetric.NewMetrics() + { + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("host.name", "potato") + rm.Resource().Attributes().PutStr("attr-key", "attr-val") + ilm := rm.ScopeMetrics().AppendEmpty() + ilm.Scope().SetName("My Library Name") + m := ilm.Metrics().AppendEmpty() + m.SetName("mem.used_percent") + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("foo", "bar") + dp.SetTimestamp(pcommon.Timestamp(1622848686000000000)) + dp.SetDoubleValue(87.332) + } + m := newMockOtelService(t) + t.Cleanup(m.Cleanup) + + plugin := &OpenTelemetry{ + ServiceAddress: m.Address(), + Timeout: config.Duration(time.Second), + Headers: map[string]string{"test": "header1"}, + Attributes: map[string]string{"attr-key": "attr-val"}, + NameSeparator: ".", + otlpMetricClient: &gRPCClient{ + grpcClientConn: m.GrpcClient(), + metricsServiceClient: pmetricotlp.NewGRPCClient(m.GrpcClient()), + }, + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Connect()) + + input := metric.New( + "mem", + map[string]string{ + "foo": "bar", + "otel.library.name": "My Library Name", + "host.name": "potato", + }, + map[string]interface{}{ + "used_percent": 87.332, + }, + time.Unix(0, 1622848686000000000), + telegraf.Gauge, + ) + + require.NoError(t, plugin.Write([]telegraf.Metric{input})) + + got := m.GotMetrics() + + marshaller := pmetric.JSONMarshaler{} + expectJSON, err := marshaller.MarshalMetrics(expect) + require.NoError(t, err) + + gotJSON, err := marshaller.MarshalMetrics(got) + require.NoError(t, err) + + require.JSONEq(t, string(expectJSON), string(gotJSON)) +} + +func TestOpenTelemetryDefaultSeparator(t *testing.T) { + expect := pmetric.NewMetrics() + { + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("host.name", "potato") + rm.Resource().Attributes().PutStr("attr-key", "attr-val") + ilm := rm.ScopeMetrics().AppendEmpty() + ilm.Scope().SetName("My Library Name") + m := ilm.Metrics().AppendEmpty() + m.SetName("mem_used_percent") + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("foo", "bar") + dp.SetTimestamp(pcommon.Timestamp(1622848686000000000)) + dp.SetDoubleValue(87.332) + } + m := newMockOtelService(t) + t.Cleanup(m.Cleanup) + + plugin := &OpenTelemetry{ + ServiceAddress: m.Address(), + Timeout: config.Duration(time.Second), + Headers: map[string]string{"test": "header1"}, + Attributes: map[string]string{"attr-key": "attr-val"}, + NameSeparator: defaultNameSeparator, + otlpMetricClient: &gRPCClient{ + grpcClientConn: m.GrpcClient(), + metricsServiceClient: pmetricotlp.NewGRPCClient(m.GrpcClient()), + }, + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Connect()) + + input := metric.New( + "mem", + map[string]string{ + "foo": "bar", + "otel.library.name": "My Library Name", + "host.name": "potato", + }, + map[string]interface{}{ + "used_percent": 87.332, + }, + time.Unix(0, 1622848686000000000), + telegraf.Gauge, + ) + + require.NoError(t, plugin.Write([]telegraf.Metric{input})) + + got := m.GotMetrics() + + marshaller := pmetric.JSONMarshaler{} + expectJSON, err := marshaller.MarshalMetrics(expect) + require.NoError(t, err) + + gotJSON, err := marshaller.MarshalMetrics(got) + require.NoError(t, err) + + require.JSONEq(t, string(expectJSON), string(gotJSON)) +} + +func TestOpenTelemetrySeparatorHTTPProtobuf(t *testing.T) { + expect := pmetric.NewMetrics() + { + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("host.name", "potato") + rm.Resource().Attributes().PutStr("attr-key", "attr-val") + ilm := rm.ScopeMetrics().AppendEmpty() + ilm.Scope().SetName("My Library Name") + m := ilm.Metrics().AppendEmpty() + m.SetName("mem.used_percent") + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("foo", "bar") + dp.SetTimestamp(pcommon.Timestamp(1622848686000000000)) + dp.SetDoubleValue(87.332) + } + + var receivedMetrics pmetric.Metrics + var receivedContentType string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + receivedContentType = r.Header.Get("Content-Type") + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + req := pmetricotlp.NewExportRequest() + err = req.UnmarshalProto(body) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + receivedMetrics = pmetric.NewMetrics() + req.Metrics().CopyTo(receivedMetrics) + + resp := pmetricotlp.NewExportResponse() + respBytes, err := resp.MarshalProto() + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/x-protobuf") + w.WriteHeader(http.StatusOK) + _, err = w.Write(respBytes) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + })) + defer server.Close() + + plugin := &OpenTelemetry{ + ServiceAddress: server.URL, + EncodingType: "protobuf", + Timeout: config.Duration(time.Second), + Attributes: map[string]string{"attr-key": "attr-val"}, + Compression: "none", + NameSeparator: ".", + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Connect()) + + input := metric.New( + "mem", + map[string]string{ + "foo": "bar", + "otel.library.name": "My Library Name", + "host.name": "potato", + }, + map[string]interface{}{ + "used_percent": 87.332, + }, + time.Unix(0, 1622848686000000000), + telegraf.Gauge, + ) + + require.NoError(t, plugin.Write([]telegraf.Metric{input})) + + require.Equal(t, "application/x-protobuf", receivedContentType) + + marshaller := pmetric.JSONMarshaler{} + expectJSON, err := marshaller.MarshalMetrics(expect) + require.NoError(t, err) + + gotJSON, err := marshaller.MarshalMetrics(receivedMetrics) + require.NoError(t, err) + + require.JSONEq(t, string(expectJSON), string(gotJSON)) +} + +func TestOpenTelemetrySeparatorHTTPJSON(t *testing.T) { + expect := pmetric.NewMetrics() + { + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("host.name", "potato") + rm.Resource().Attributes().PutStr("attr-key", "attr-val") + ilm := rm.ScopeMetrics().AppendEmpty() + ilm.Scope().SetName("My Library Name") + m := ilm.Metrics().AppendEmpty() + m.SetName("mem.used_percent") + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("foo", "bar") + dp.SetTimestamp(pcommon.Timestamp(1622848686000000000)) + dp.SetDoubleValue(87.332) + } + + var receivedMetrics pmetric.Metrics + var receivedContentType string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + receivedContentType = r.Header.Get("Content-Type") + body, err := io.ReadAll(r.Body) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + req := pmetricotlp.NewExportRequest() + err = req.UnmarshalJSON(body) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + receivedMetrics = pmetric.NewMetrics() + req.Metrics().CopyTo(receivedMetrics) + + resp := pmetricotlp.NewExportResponse() + respBytes, err := resp.MarshalJSON() + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, err = w.Write(respBytes) + if err != nil { + t.Error(err) + w.WriteHeader(http.StatusInternalServerError) + return + } + })) + defer server.Close() + + plugin := &OpenTelemetry{ + ServiceAddress: server.URL, + EncodingType: "json", + Timeout: config.Duration(time.Second), + Attributes: map[string]string{"attr-key": "attr-val"}, + Compression: "none", + NameSeparator: ".", + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Connect()) + + input := metric.New( + "mem", + map[string]string{ + "foo": "bar", + "otel.library.name": "My Library Name", + "host.name": "potato", + }, + map[string]interface{}{ + "used_percent": 87.332, + }, + time.Unix(0, 1622848686000000000), + telegraf.Gauge, + ) + + require.NoError(t, plugin.Write([]telegraf.Metric{input})) + + require.Equal(t, "application/json", receivedContentType) + + marshaller := pmetric.JSONMarshaler{} + expectJSON, err := marshaller.MarshalMetrics(expect) + require.NoError(t, err) + + gotJSON, err := marshaller.MarshalMetrics(receivedMetrics) + require.NoError(t, err) + + require.JSONEq(t, string(expectJSON), string(gotJSON)) +} + var _ pmetricotlp.GRPCServer = (*mockOtelService)(nil) type mockOtelService struct { diff --git a/plugins/outputs/opentelemetry/sample.conf b/plugins/outputs/opentelemetry/sample.conf index ed462e9009601..ae8973eef6095 100644 --- a/plugins/outputs/opentelemetry/sample.conf +++ b/plugins/outputs/opentelemetry/sample.conf @@ -28,6 +28,9 @@ ## Supports: "gzip", "none" # compression = "gzip" + ## Characters for separating LineProtocol's metric name and field when converting to OpenTelemetry's metric name + # name_separator = "_" + ## Optional proxy settings for HTTP endpoints (service_address starts with http:// or https://) # use_system_proxy = false # http_proxy_url = ""