Switch to otel

Author: simonswine

pkg/phlaredb/query/iters.go

@@ -10,9 +10,10 @@ import (
 	"sync"
 
 	"github.com/grafana/dskit/multierror"
-	"github.com/opentracing/opentracing-go"
-	"github.com/opentracing/opentracing-go/log"
 	"github.com/segmentio/parquet-go"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 
 	"github.com/grafana/phlare/pkg/iter"
 )
@@ -39,6 +40,10 @@ type RowNumberWithDefinitionLevel struct {
 	DefinitionLevel int
 }
 
+func (r *RowNumberWithDefinitionLevel) String() string {
+	return fmt.Sprintf("%v:%v", r.RowNumber, r.DefinitionLevel)
+}
+
 // EmptyRowNumber creates an empty invalid row number.
 func EmptyRowNumber() RowNumber {
 	return RowNumber{-1, -1, -1, -1, -1, -1}
@@ -782,7 +787,7 @@ type SyncIterator struct {
 	filter     *InstrumentedPredicate
 
 	// Status
-	span            opentracing.Span
+	span            trace.Span
 	metrics         *Metrics
 	curr            RowNumber
 	currRowGroup    parquet.RowGroup
@@ -839,10 +844,12 @@ func NewSyncIterator(ctx context.Context, rgs []parquet.RowGroup, column int, co
 		rn.Skip(rg.NumRows())
 	}
 
-	span, _ := opentracing.StartSpanFromContext(ctx, "syncIterator", opentracing.Tags{
-		"columnIndex": column,
-		"column":      columnName,
-	})
+	tr := otel.Tracer("query")
+
+	ctx, span := tr.Start(ctx, "syncIterator", trace.WithAttributes(
+		attribute.String("column", columnName),
+		attribute.Int("columnIndex", column),
+	))
 
 	return &SyncIterator{
 		table:      strings.ToLower(rgs[0].Schema().Name()) + "s",
@@ -1012,11 +1019,12 @@ func (c *SyncIterator) seekPages(seekTo RowNumber, definitionLevel int) (done bo
 				return true, err
 			}
 			c.metrics.pageReadsTotal.WithLabelValues(c.table, c.columnName).Add(1)
-			c.span.LogFields(
-				log.String("msg", "reading page (seekPages)"),
-				log.Int64("page_num_values", pg.NumValues()),
-				log.Int64("page_size", pg.Size()),
-			)
+			c.span.AddEvent(
+				"read page (seekPages)",
+				trace.WithAttributes(
+					attribute.Int64("page_num_values", pg.NumValues()),
+					attribute.Int64("page_size", pg.Size()),
+				))
 
 			// Skip based on row number?
 			newRN := c.curr
@@ -1073,11 +1081,12 @@ func (c *SyncIterator) next() (RowNumber, *parquet.Value, error) {
 				return EmptyRowNumber(), nil, err
 			}
 			c.metrics.pageReadsTotal.WithLabelValues(c.table, c.columnName).Add(1)
-			c.span.LogFields(
-				log.String("msg", "reading page (next)"),
-				log.Int64("page_num_values", pg.NumValues()),
-				log.Int64("page_size", pg.Size()),
-			)
+			c.span.AddEvent(
+				"read page (next)",
+				trace.WithAttributes(
+					attribute.Int64("page_num_values", pg.NumValues()),
+					attribute.Int64("page_size", pg.Size()),
+				))
 
 			if c.filter != nil && !c.filter.KeepPage(pg) {
 				// This page filtered out
@@ -1195,12 +1204,14 @@ func (c *SyncIterator) Err() error {
 func (c *SyncIterator) Close() error {
 	c.closeCurrRowGroup()
 
-	c.span.SetTag("inspectedColumnChunks", c.filter.InspectedColumnChunks.Load())
-	c.span.SetTag("inspectedPages", c.filter.InspectedPages.Load())
-	c.span.SetTag("inspectedValues", c.filter.InspectedValues.Load())
-	c.span.SetTag("keptColumnChunks", c.filter.KeptColumnChunks.Load())
-	c.span.SetTag("keptPages", c.filter.KeptPages.Load())
-	c.span.SetTag("keptValues", c.filter.KeptValues.Load())
-	c.span.Finish()
+	c.span.SetAttributes(attribute.Int64("inspectedColumnChunks", c.filter.InspectedColumnChunks.Load()))
+	/*
+		c.span.SetTag("inspectedPages", c.filter.InspectedPages.Load())
+		c.span.SetTag("inspectedValues", c.filter.InspectedValues.Load())
+		c.span.SetTag("keptColumnChunks", c.filter.KeptColumnChunks.Load())
+		c.span.SetTag("keptPages", c.filter.KeptPages.Load())
+		c.span.SetTag("keptValues", c.filter.KeptValues.Load())
+	*/
+	c.span.End()
 	return nil
 }

pkg/phlaredb/query/iters_test.go

@@ -348,14 +348,18 @@ func createFileWith[T any](t testing.TB, rows []T, rowGroups int) *parquet.File
 	require.NoError(t, err)
 	t.Logf("Created temp file %s", f.Name())
 
-	half := len(rows) / rowGroups
+	perRG := len(rows) / rowGroups
 
 	w := parquet.NewGenericWriter[T](f)
-	_, err = w.Write(rows[0:half])
-	require.NoError(t, err)
-	require.NoError(t, w.Flush())
 
-	_, err = w.Write(rows[half:])
+	for i := 0; i < (rowGroups - 1); i++ {
+		_, err = w.Write(rows[0:perRG])
+		require.NoError(t, err)
+		require.NoError(t, w.Flush())
+		rows = rows[perRG:]
+	}
+
+	_, err = w.Write(rows)
 	require.NoError(t, err)
 	require.NoError(t, w.Flush())
 
@@ -380,18 +384,15 @@ type iteratorTracer struct {
 
 func (i iteratorTracer) Next() bool {
 	i.nextCount++
-	//posBefore := i.it.At()
+	posBefore := i.it.At()
 	result := i.it.Next()
-	//posAfter := i.it.At()
-	/*
-		i.span.AddAttributes.LogKV(
-			"event", "next",
-			"result", result,
-			"column", i.name,
-			"posBefore", posBefore,
-			"posAfter", posAfter,
-		)
-	*/
+	posAfter := i.it.At()
+	i.span.AddEvent("next", trace.WithAttributes(
+		attribute.String("column", i.name),
+		attribute.Bool("result", result),
+		attribute.Stringer("posBefore", posBefore),
+		attribute.Stringer("posAfter", posAfter),
+	))
 	return result
 }
 
@@ -412,22 +413,13 @@ func (i iteratorTracer) Seek(pos RowNumberWithDefinitionLevel) bool {
 	posBefore := i.it.At()
 	result := i.it.Seek(pos)
 	posAfter := i.it.At()
-
 	i.span.AddEvent("seek", trace.WithAttributes(
 		attribute.String("column", i.name),
+		attribute.Bool("result", result),
+		attribute.Stringer("seekTo", &pos),
 		attribute.Stringer("posBefore", posBefore),
 		attribute.Stringer("posAfter", posAfter),
 	))
-	/*
-		i.span.LogKV(
-			"event", "seek",
-			"result", result,
-			"column", i.name,
-			"seekTo", pos,
-			"posBefore", posBefore,
-			"posAfter", posAfter,
-		)
-	*/
 	return result
 }
 
@@ -544,12 +536,9 @@ func TestBinaryJoinIterator(t *testing.T) {
 
 			results := 0
 			for it.Next() {
-				/*
-					span.LogKV(
-						"event", "match",
-						"pos", it.At().RowNumber,
-					)
-				*/
+				span.AddEvent("match", trace.WithAttributes(
+					attribute.Stringer("element", it.At()),
+				))
 				results++
 			}
 			require.NoError(t, it.Err())