diff --git a/influx2otel/metrics.go b/influx2otel/metrics.go index 7c37cf9..81db952 100644 --- a/influx2otel/metrics.go +++ b/influx2otel/metrics.go @@ -16,12 +16,21 @@ import ( ) type LineProtocolToOtelMetrics struct { - logger common.Logger + logger common.Logger + nameSeparator string } func NewLineProtocolToOtelMetrics(logger common.Logger) (*LineProtocolToOtelMetrics, error) { return &LineProtocolToOtelMetrics{ - logger: logger, + logger: logger, + nameSeparator: "_", + }, nil +} + +func NewLineProtocolToOtelMetricsWithSeparator(logger common.Logger, nameSeparator string) (*LineProtocolToOtelMetrics, error) { + return &LineProtocolToOtelMetrics{ + logger: logger, + nameSeparator: nameSeparator, }, nil } @@ -33,7 +42,8 @@ func (c *LineProtocolToOtelMetrics) NewBatch() *MetricsBatch { histogramDataPointsByMDPK: make(map[pmetric.Metric]map[dataPointKey]pmetric.HistogramDataPoint), summaryDataPointsByMDPK: make(map[pmetric.Metric]map[dataPointKey]pmetric.SummaryDataPoint), - logger: c.logger, + logger: c.logger, + nameSeparator: c.nameSeparator, } } @@ -44,7 +54,8 @@ type MetricsBatch struct { histogramDataPointsByMDPK map[pmetric.Metric]map[dataPointKey]pmetric.HistogramDataPoint summaryDataPointsByMDPK map[pmetric.Metric]map[dataPointKey]pmetric.SummaryDataPoint - logger common.Logger + logger common.Logger + nameSeparator string } // measurement - metric name @@ -235,7 +246,7 @@ func (b *MetricsBatch) addPointWithUnknownSchema(measurement string, tags map[st continue } - metricName := fmt.Sprintf("%s_%s", measurement, k) + metricName := fmt.Sprintf("%s%s%s", measurement, b.nameSeparator, k) metric, attributes, err := b.lookupMetric(metricName, tags, common.InfluxMetricValueTypeGauge) if err != nil { return err diff --git a/influx2otel/metrics_telegraf_prometheus_v1.go b/influx2otel/metrics_telegraf_prometheus_v1.go index 765796f..03618b8 100644 --- a/influx2otel/metrics_telegraf_prometheus_v1.go +++ b/influx2otel/metrics_telegraf_prometheus_v1.go @@ -130,7 +130,7 @@ func (b *MetricsBatch) convertGaugeV1(measurement string, tags map[string]string continue } - metricName := fmt.Sprintf("%s_%s", measurement, k) + metricName := fmt.Sprintf("%s%s%s", measurement, b.nameSeparator, k) metric, attributes, err := b.lookupMetric(metricName, tags, common.InfluxMetricValueTypeGauge) if err != nil { return err @@ -220,7 +220,7 @@ func (b *MetricsBatch) convertSumV1(measurement string, tags map[string]string, continue } - metricName := fmt.Sprintf("%s_%s", measurement, k) + metricName := fmt.Sprintf("%s%s%s", measurement, b.nameSeparator, k) metric, attributes, err := b.lookupMetric(metricName, tags, common.InfluxMetricValueTypeSum) if err != nil { return err diff --git a/influx2otel/metrics_telegraf_prometheus_v1_test.go b/influx2otel/metrics_telegraf_prometheus_v1_test.go index d883076..f0405d8 100644 --- a/influx2otel/metrics_telegraf_prometheus_v1_test.go +++ b/influx2otel/metrics_telegraf_prometheus_v1_test.go @@ -534,3 +534,167 @@ func TestAddPoint_v1_untypedSummary(t *testing.T) { assertMetricsEqual(t, expect, b.GetMetrics()) } + +func TestAddPoint_v1_gauge_separator(t *testing.T) { + // Use a dot to separate name and fields + c, err := influx2otel.NewLineProtocolToOtelMetricsWithSeparator(new(common.NoopLogger), ".") + require.NoError(t, err) + + b := c.NewBatch() + err = b.AddPoint("cache_age_seconds", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "engine_id": "0", + }, + map[string]interface{}{ + "gauge": float64(23.9), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeGauge) + require.NoError(t, err) + + err = b.AddPoint("cache_age_seconds", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "engine_id": "1", + }, + map[string]interface{}{ + "custom_gauge": float64(11.9), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeGauge) + require.NoError(t, err) + + expect := pmetric.NewMetrics() + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("container.name", "42") + isMetrics := rm.ScopeMetrics().AppendEmpty() + isMetrics.Scope().SetName("My Library") + isMetrics.Scope().SetVersion("latest") + m := isMetrics.Metrics().AppendEmpty() + m.SetName("cache_age_seconds") + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("engine_id", "0") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetDoubleValue(23.9) + m = isMetrics.Metrics().AppendEmpty() + m.SetName("cache_age_seconds.custom_gauge") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp = m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("engine_id", "1") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetDoubleValue(11.9) + + assertMetricsEqual(t, expect, b.GetMetrics()) +} +func TestAddPoint_v1_sum_separator(t *testing.T) { + // Use a dot to separate name and fields + c, err := influx2otel.NewLineProtocolToOtelMetricsWithSeparator(new(common.NoopLogger), ".") + require.NoError(t, err) + + b := c.NewBatch() + err = b.AddPoint("http_requests_total", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "method": "post", + "code": "200", + }, + map[string]interface{}{ + "counter": float64(1027), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeSum) + require.NoError(t, err) + + err = b.AddPoint("http_requests_total", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "method": "post", + "code": "400", + }, + map[string]interface{}{ + "custom_counter": float64(3), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeSum) + require.NoError(t, err) + + expect := pmetric.NewMetrics() + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("container.name", "42") + isMetrics := rm.ScopeMetrics().AppendEmpty() + isMetrics.Scope().SetName("My Library") + isMetrics.Scope().SetVersion("latest") + m := isMetrics.Metrics().AppendEmpty() + m.SetName("http_requests_total") + m.SetEmptySum() + m.Sum().SetIsMonotonic(true) + m.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) + dp := m.Sum().DataPoints().AppendEmpty() + dp.Attributes().PutStr("code", "200") + dp.Attributes().PutStr("method", "post") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetDoubleValue(1027) + m = isMetrics.Metrics().AppendEmpty() + m.SetName("http_requests_total.custom_counter") // Uses a dot to separate name and fields + m.SetEmptySum() + m.Sum().SetIsMonotonic(true) + m.Sum().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) + dp = m.Sum().DataPoints().AppendEmpty() + dp.Attributes().PutStr("code", "400") + dp.Attributes().PutStr("method", "post") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetDoubleValue(3) + + assertMetricsEqual(t, expect, b.GetMetrics()) +} +func TestAddPoint_v1_untyped_separator(t *testing.T) { + // Use a dot to separate name and fields + c, err := influx2otel.NewLineProtocolToOtelMetricsWithSeparator(new(common.NoopLogger), ".") + require.NoError(t, err) + + b := c.NewBatch() + err = b.AddPoint("some_custom_metric", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + }, + map[string]any{ + "count": int64(1), + "something_else": float64(2.3), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeUntyped) + require.NoError(t, err) + + expect := pmetric.NewMetrics() + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("container.name", "42") + isMetrics := rm.ScopeMetrics().AppendEmpty() + isMetrics.Scope().SetName("My Library") + isMetrics.Scope().SetVersion("latest") + m := isMetrics.Metrics().AppendEmpty() + m.SetName("some_custom_metric.count") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetIntValue(1) + m = isMetrics.Metrics().AppendEmpty() + m.SetName("some_custom_metric.something_else") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp = m.Gauge().DataPoints().AppendEmpty() + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetDoubleValue(2.3) + + assertMetricsEqual(t, expect, b.GetMetrics()) +} diff --git a/influx2otel/metrics_unknown_schema_test.go b/influx2otel/metrics_unknown_schema_test.go index 9f5456d..a9e0c58 100644 --- a/influx2otel/metrics_unknown_schema_test.go +++ b/influx2otel/metrics_unknown_schema_test.go @@ -68,3 +68,60 @@ func TestUnknownSchema(t *testing.T) { assertMetricsEqual(t, expect, b.GetMetrics()) } + +func TestUnknownSchema_CustomSeparator(t *testing.T) { + // Use a dot to separate name and fields + c, err := influx2otel.NewLineProtocolToOtelMetricsWithSeparator(new(common.NoopLogger), ".") + require.NoError(t, err) + + b := c.NewBatch() + err = b.AddPoint("cpu", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "cpu": "cpu4", + "host": "777348dc6343", + }, + map[string]interface{}{ + "usage_user": 0.10090817356207936, + "usage_system": 0.3027245206862381, + "some_int_key": int64(7), + }, + time.Unix(0, 1395066363000000123), + common.InfluxMetricValueTypeUntyped) + require.NoError(t, err) + + expect := pmetric.NewMetrics() + rm := expect.ResourceMetrics().AppendEmpty() + rm.Resource().Attributes().PutStr("container.name", "42") + isMetrics := rm.ScopeMetrics().AppendEmpty() + isMetrics.Scope().SetName("My Library") + isMetrics.Scope().SetVersion("latest") + m := isMetrics.Metrics().AppendEmpty() + m.SetName("cpu.usage_user") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp := m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("cpu", "cpu4") + dp.Attributes().PutStr("host", "777348dc6343") + dp.SetTimestamp(pcommon.Timestamp(1395066363000000123)) + dp.SetDoubleValue(0.10090817356207936) + m = isMetrics.Metrics().AppendEmpty() + m.SetName("cpu.usage_system") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp = m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("cpu", "cpu4") + dp.Attributes().PutStr("host", "777348dc6343") + dp.SetTimestamp(pcommon.Timestamp(1395066363000000123)) + dp.SetDoubleValue(0.3027245206862381) + m = isMetrics.Metrics().AppendEmpty() + m.SetName("cpu.some_int_key") // Uses a dot to separate name and fields + m.SetEmptyGauge() + dp = m.Gauge().DataPoints().AppendEmpty() + dp.Attributes().PutStr("cpu", "cpu4") + dp.Attributes().PutStr("host", "777348dc6343") + dp.SetTimestamp(pcommon.Timestamp(1395066363000000123)) + dp.SetIntValue(7) + + assertMetricsEqual(t, expect, b.GetMetrics()) +}