From 9cb2cfa1de43998fbbe0cd9a42c0314e87d22161 Mon Sep 17 00:00:00 2001 From: skartikey <1942366+skartikey@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:38:20 +0100 Subject: [PATCH] fix(influx2otel): accept integer count, sum, and bucket values for histograms and summaries Influx line protocol integer fields (for example count=1i) decode as int64, but the V1 histogram and summary converters only accepted float64 and rejected them with "unsupported histogram count value type int64", dropping the metric. Add toFloat64/toUint64 helpers that also accept int64 and uint64 (matching the gauge and sum converters) and use them for the count, sum, and bucket/quantile values in convertHistogramV1 and convertSummaryV1. Related to https://github.com/influxdata/telegraf/issues/14371 --- influx2otel/metrics_telegraf_prometheus_v1.go | 71 ++++++++--- .../metrics_telegraf_prometheus_v1_test.go | 113 ++++++++++++++++++ 2 files changed, 165 insertions(+), 19 deletions(-) diff --git a/influx2otel/metrics_telegraf_prometheus_v1.go b/influx2otel/metrics_telegraf_prometheus_v1.go index 765796f..5276487 100644 --- a/influx2otel/metrics_telegraf_prometheus_v1.go +++ b/influx2otel/metrics_telegraf_prometheus_v1.go @@ -69,6 +69,37 @@ func isStringNumeric(s string) bool { return err == nil } +// toFloat64 coerces a numeric line protocol field value to float64. Integer +// fields (for example count=1i) arrive as int64 or uint64, so accept those in +// addition to float64. +func toFloat64(value interface{}) (float64, bool) { + switch v := value.(type) { + case float64: + return v, true + case int64: + return float64(v), true + case uint64: + return float64(v), true + default: + return 0, false + } +} + +// toUint64 coerces a numeric line protocol field value to uint64, accepting the +// same integer field types as toFloat64. +func toUint64(value interface{}) (uint64, bool) { + switch v := value.(type) { + case float64: + return uint64(v), true + case int64: + return uint64(v), true + case uint64: + return v, true + default: + return 0, false + } +} + func (b *MetricsBatch) convertGaugeV1(measurement string, tags map[string]string, fields map[string]interface{}, ts time.Time) error { if fieldValue, found := fields[common.MetricGaugeFieldKey]; found { var floatValue *float64 @@ -259,26 +290,27 @@ func (b *MetricsBatch) convertHistogramV1(measurement string, tags map[string]st for k, vi := range fields { if k == common.MetricHistogramCountFieldKey { foundCount = true - if vCount, ok := vi.(float64); !ok { + vCount, ok := toUint64(vi) + if !ok { return fmt.Errorf("unsupported histogram count value type %T", vi) - } else { - count = uint64(vCount) } + count = vCount } else if k == common.MetricHistogramSumFieldKey { foundSum = true - var ok bool - if sum, ok = vi.(float64); !ok { + vSum, ok := toFloat64(vi) + if !ok { return fmt.Errorf("unsupported histogram sum value type %T", vi) } + sum = vSum } else if explicitBound, err := strconv.ParseFloat(k, 64); err == nil { - if vBucketCount, ok := vi.(float64); !ok { + vBucketCount, ok := toUint64(vi) + if !ok { return fmt.Errorf("unsupported histogram bucket bound value type %T", vi) - } else { - explicitBounds = append(explicitBounds, explicitBound) - bucketCounts = append(bucketCounts, uint64(vBucketCount)) } + explicitBounds = append(explicitBounds, explicitBound) + bucketCounts = append(bucketCounts, vBucketCount) } else if k == common.AttributeStartTimeStatsd { } else { b.logger.Debug("skipping unrecognized histogram field", "field", k, "value", vi) @@ -339,27 +371,28 @@ func (b *MetricsBatch) convertSummaryV1(measurement string, tags map[string]stri for k, vi := range fields { if k == common.MetricSummaryCountFieldKey { foundCount = true - if vCount, ok := vi.(float64); !ok { + vCount, ok := toUint64(vi) + if !ok { return fmt.Errorf("unsupported summary count value type %T", vi) - } else { - count = uint64(vCount) } + count = vCount } else if k == common.MetricSummarySumFieldKey { foundSum = true - var ok bool - if sum, ok = vi.(float64); !ok { + vSum, ok := toFloat64(vi) + if !ok { return fmt.Errorf("unsupported summary sum value type %T", vi) } + sum = vSum } else if quantile, err := strconv.ParseFloat(k, 64); err == nil { - if value, ok := vi.(float64); !ok { + value, ok := toFloat64(vi) + if !ok { return fmt.Errorf("unsupported summary bucket bound value type %T", vi) - } else { - valueAtQuantile := quantileValues.AppendEmpty() - valueAtQuantile.SetQuantile(quantile) - valueAtQuantile.SetValue(value) } + valueAtQuantile := quantileValues.AppendEmpty() + valueAtQuantile.SetQuantile(quantile) + valueAtQuantile.SetValue(value) } else if k == common.AttributeStartTimeStatsd { } else { b.logger.Debug("skipping unrecognized summary field", "field", k, "value", vi) diff --git a/influx2otel/metrics_telegraf_prometheus_v1_test.go b/influx2otel/metrics_telegraf_prometheus_v1_test.go index d883076..3d1bb10 100644 --- a/influx2otel/metrics_telegraf_prometheus_v1_test.go +++ b/influx2otel/metrics_telegraf_prometheus_v1_test.go @@ -534,3 +534,116 @@ func TestAddPoint_v1_untypedSummary(t *testing.T) { assertMetricsEqual(t, expect, b.GetMetrics()) } + +func TestAddPoint_v1_histogramIntegerFields(t *testing.T) { + c, err := influx2otel.NewLineProtocolToOtelMetrics(new(common.NoopLogger)) + require.NoError(t, err) + + // Integer line protocol fields (for example count=144320i) arrive as int64, + // which must be accepted rather than rejected as an unsupported value type. + b := c.NewBatch() + err = b.AddPoint("http_request_duration_seconds", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "method": "post", + "code": "200", + }, + map[string]interface{}{ + "count": int64(144320), + "sum": int64(53423), + "0.05": int64(24054), + "0.1": int64(33444), + "0.2": int64(100392), + "0.5": int64(129389), + "1": int64(133988), + "+Inf": int64(144320), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeHistogram) + 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_request_duration_seconds") + m.SetEmptyHistogram() + m.Histogram().SetAggregationTemporality(pmetric.AggregationTemporalityCumulative) + dp := m.Histogram().DataPoints().AppendEmpty() + dp.Attributes().PutStr("code", "200") + dp.Attributes().PutStr("method", "post") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetCount(144320) + dp.SetSum(53423) + dp.BucketCounts().FromRaw([]uint64{24054, 9390, 66948, 28997, 4599, 10332}) + dp.ExplicitBounds().FromRaw([]float64{0.05, 0.1, 0.2, 0.5, 1}) + + assertMetricsEqual(t, expect, b.GetMetrics()) +} + +func TestAddPoint_v1_summaryIntegerFields(t *testing.T) { + c, err := influx2otel.NewLineProtocolToOtelMetrics(new(common.NoopLogger)) + require.NoError(t, err) + + // Integer line protocol fields (for example count=2693i) arrive as int64, + // which must be accepted rather than rejected as an unsupported value type. + b := c.NewBatch() + err = b.AddPoint("rpc_duration_seconds", + map[string]string{ + "container.name": "42", + "otel.library.name": "My Library", + "otel.library.version": "latest", + "method": "post", + "code": "200", + }, + map[string]interface{}{ + "count": int64(2693), + "sum": int64(17560473), + "0.01": int64(3102), + "0.05": int64(3272), + "0.5": int64(4773), + "0.9": int64(9001), + "0.99": int64(76656), + }, + time.Unix(0, 1395066363000000123).UTC(), + common.InfluxMetricValueTypeSummary) + 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("rpc_duration_seconds") + m.SetEmptySummary() + dp := m.Summary().DataPoints().AppendEmpty() + dp.Attributes().PutStr("code", "200") + dp.Attributes().PutStr("method", "post") + dp.SetTimestamp(pcommon.NewTimestampFromTime(time.Unix(0, 1395066363000000123))) + dp.SetCount(2693) + dp.SetSum(17560473) + qv := dp.QuantileValues().AppendEmpty() + qv.SetQuantile(0.01) + qv.SetValue(3102) + qv = dp.QuantileValues().AppendEmpty() + qv.SetQuantile(0.05) + qv.SetValue(3272) + qv = dp.QuantileValues().AppendEmpty() + qv.SetQuantile(0.5) + qv.SetValue(4773) + qv = dp.QuantileValues().AppendEmpty() + qv.SetQuantile(0.9) + qv.SetValue(9001) + qv = dp.QuantileValues().AppendEmpty() + qv.SetQuantile(0.99) + qv.SetValue(76656) + + assertMetricsEqual(t, expect, b.GetMetrics()) +}