Pass logger down to FromMetrics

This commit passes logger down to FromMetrics and extends the converter
to printout some information about the incoming series timestamps and
start timestamps.

Only for sums.

Signed-off-by: Jesus Vazquez <jesus.vazquez@grafana.com>

format series

Signed-off-by: Jesus Vazquez <jesus.vazquez@grafana.com>

dont store, log right away

Signed-off-by: Jesus Vazquez <jesus.vazquez@grafana.com>

Author: jesusvazquez

storage/remote/otlptranslator/prometheusremotewrite/metrics_to_prw.go

@@ -21,7 +21,10 @@ import (
 	"errors"
 	"fmt"
 	"sort"
+	"strings"
 
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/multierr"
@@ -42,6 +45,12 @@ type Settings struct {
 	EnableCreatedTimestampZeroIngestion bool
 }
 
+type StartTsAndTs struct {
+	StartTs int64
+	Ts      int64
+	Labels  []prompb.Label
+}
+
 // PrometheusConverter converts from OTel write format to Prometheus remote write format.
 type PrometheusConverter struct {
 	unique    map[uint64]*prompb.TimeSeries
@@ -57,7 +66,7 @@ func NewPrometheusConverter() *PrometheusConverter {
 }
 
 // FromMetrics converts pmetric.Metrics to Prometheus remote write format.
-func (c *PrometheusConverter) FromMetrics(ctx context.Context, md pmetric.Metrics, settings Settings) (annots annotations.Annotations, errs error) {
+func (c *PrometheusConverter) FromMetrics(ctx context.Context, md pmetric.Metrics, settings Settings, logger log.Logger) (annots annotations.Annotations, errs error) {
 	c.everyN = everyNTimes{n: 128}
 	resourceMetricsSlice := md.ResourceMetrics()
 	for i := 0; i < resourceMetricsSlice.Len(); i++ {
@@ -108,7 +117,7 @@ func (c *PrometheusConverter) FromMetrics(ctx context.Context, md pmetric.Metric
 						errs = multierr.Append(errs, fmt.Errorf("empty data points. %s is dropped", metric.Name()))
 						break
 					}
-					if err := c.addSumNumberDataPoints(ctx, dataPoints, resource, metric, settings, promName); err != nil {
+					if err := c.addSumNumberDataPoints(ctx, dataPoints, resource, metric, settings, promName, logger); err != nil {
 						errs = multierr.Append(errs, err)
 						if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 							return
@@ -226,3 +235,22 @@ func (c *PrometheusConverter) addSample(sample *prompb.Sample, lbls []prompb.Lab
 	ts.Samples = append(ts.Samples, *sample)
 	return ts
 }
+
+// TODO(jesus.vazquez) This method is for debugging only and its meant to be removed soon.
+// trackStartTimestampForSeries logs the start timestamp for a series if it has been seen before.
+func (c *PrometheusConverter) trackStartTimestampForSeries(startTs, ts int64, lbls []prompb.Label, logger log.Logger) {
+	h := timeSeriesSignature(lbls)
+	if _, ok := c.unique[h]; ok {
+		var seriesBuilder strings.Builder
+		seriesBuilder.WriteString("{")
+		for i, l := range lbls {
+			if i > 0 {
+				seriesBuilder.WriteString(",")
+			}
+			seriesBuilder.WriteString(fmt.Sprintf("%s=%s", l.Name, l.Value))
+
+		}
+		seriesBuilder.WriteString("}")
+		level.Debug(logger).Log("labels", seriesBuilder.String(), "start_ts", startTs, "ts", ts)
+	}
+}

storage/remote/otlptranslator/prometheusremotewrite/metrics_to_prw_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-kit/log"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -36,7 +37,7 @@ func TestFromMetrics(t *testing.T) {
 		cancel()
 		payload := createExportRequest(5, 128, 128, 2, 0)
 
-		_, err := converter.FromMetrics(ctx, payload.Metrics(), Settings{})
+		_, err := converter.FromMetrics(ctx, payload.Metrics(), Settings{}, log.NewNopLogger())
 		require.ErrorIs(t, err, context.Canceled)
 	})
 
@@ -47,7 +48,7 @@ func TestFromMetrics(t *testing.T) {
 		t.Cleanup(cancel)
 		payload := createExportRequest(5, 128, 128, 2, 0)
 
-		_, err := converter.FromMetrics(ctx, payload.Metrics(), Settings{})
+		_, err := converter.FromMetrics(ctx, payload.Metrics(), Settings{}, log.NewNopLogger())
 		require.ErrorIs(t, err, context.DeadlineExceeded)
 	})
 
@@ -74,7 +75,7 @@ func TestFromMetrics(t *testing.T) {
 		}
 
 		converter := NewPrometheusConverter()
-		annots, err := converter.FromMetrics(context.Background(), request.Metrics(), Settings{})
+		annots, err := converter.FromMetrics(context.Background(), request.Metrics(), Settings{}, log.NewNopLogger())
 		require.NoError(t, err)
 		require.NotEmpty(t, annots)
 		ws, infos := annots.AsStrings("", 0, 0)
@@ -107,7 +108,7 @@ func BenchmarkPrometheusConverter_FromMetrics(b *testing.B) {
 
 											for i := 0; i < b.N; i++ {
 												converter := NewPrometheusConverter()
-												annots, err := converter.FromMetrics(context.Background(), payload.Metrics(), Settings{})
+												annots, err := converter.FromMetrics(context.Background(), payload.Metrics(), Settings{}, log.NewNopLogger())
 												require.NoError(b, err)
 												require.Empty(b, annots)
 												require.NotNil(b, converter.TimeSeries())

storage/remote/otlptranslator/prometheusremotewrite/number_data_points.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"math"
 
+	"github.com/go-kit/log"
 	"github.com/prometheus/common/model"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
@@ -65,14 +66,14 @@ func (c *PrometheusConverter) addGaugeNumberDataPoints(ctx context.Context, data
 }
 
 func (c *PrometheusConverter) addSumNumberDataPoints(ctx context.Context, dataPoints pmetric.NumberDataPointSlice,
-	resource pcommon.Resource, metric pmetric.Metric, settings Settings, name string) error {
+	resource pcommon.Resource, metric pmetric.Metric, settings Settings, name string, logger log.Logger) error {
 	for x := 0; x < dataPoints.Len(); x++ {
 		if err := c.everyN.checkContext(ctx); err != nil {
 			return err
 		}
 
 		pt := dataPoints.At(x)
-		startTimestampNs := pt.StartTimestamp()
+		timestamp := convertTimeStamp(pt.Timestamp())
 		startTimestampMs := convertTimeStamp(pt.StartTimestamp())
 		lbls := createAttributes(
 			resource,
@@ -83,7 +84,7 @@ func (c *PrometheusConverter) addSumNumberDataPoints(ctx context.Context, dataPo
 			model.MetricNameLabel,
 			name,
 		)
-		timestamp := convertTimeStamp(pt.Timestamp())
+
 		sample := &prompb.Sample{
 			Timestamp: timestamp,
 		}
@@ -111,7 +112,7 @@ func (c *PrometheusConverter) addSumNumberDataPoints(ctx context.Context, dataPo
 
 		// add created time series if needed
 		if settings.ExportCreatedMetric && isMonotonic {
-			if startTimestampNs == 0 {
+			if startTimestampMs == 0 {
 				return nil
 			}
 
@@ -125,6 +126,7 @@ func (c *PrometheusConverter) addSumNumberDataPoints(ctx context.Context, dataPo
 			}
 			c.addTimeSeriesIfNeeded(createdLabels, startTimestampMs, pt.Timestamp())
 		}
+		c.trackStartTimestampForSeries(startTimestampMs, timestamp, lbls, logger)
 	}
 
 	return nil

storage/remote/otlptranslator/prometheusremotewrite/number_data_points_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-kit/log"
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -255,6 +256,7 @@ func TestPrometheusConverter_addSumNumberDataPoints(t *testing.T) {
 					EnableCreatedTimestampZeroIngestion: true,
 				},
 				metric.Name(),
+				log.NewNopLogger(),
 			)
 
 			assert.Equal(t, tt.want(), converter.unique)

storage/remote/write_handler.go

@@ -518,7 +518,7 @@ func (h *otlpWriteHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		AddMetricSuffixes:                   true,
 		PromoteResourceAttributes:           otlpCfg.PromoteResourceAttributes,
 		EnableCreatedTimestampZeroIngestion: h.enableCTZeroIngestion,
-	})
+	}, h.logger)
 	if err != nil {
 		level.Warn(h.logger).Log("msg", "Error translating OTLP metrics to Prometheus write request", "err", err)
 	}