Migrate metrics export from OTel Arrow to OTLP

exporter.go

@@ -9,9 +9,9 @@ import (
 	"time"
 
 	"github.com/oklog/run"
-	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
-	"github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
+	"google.golang.org/grpc"
 )
 
 type Producer interface {
@@ -24,58 +24,37 @@ type ProducerConfig struct {
 	ScopeName string
 }
 
-type stream struct {
-	endpoint      arrowpb.ArrowMetricsService_ArrowMetricsClient
-	arrowProducer *arrow_record.Producer
-}
-
 type Exporter struct {
-	client arrowpb.ArrowMetricsServiceClient
+	client pmetricotlp.GRPCClient
 	// NB: someday we might want to have several producer groups,
 	// each of which collects at different intervals.
 	// For now we are only collecting one scope (GPU metrics)
 	// so the global interval is fine.
 	interval      time.Duration
 	producers     []ProducerConfig
 	resourceAttrs map[string]any
-	stream        *stream
 }
 
-func NewExporter(client arrowpb.ArrowMetricsServiceClient, interval time.Duration, resourceAttrs map[string]any) Exporter {
+func NewExporter(conn *grpc.ClientConn, interval time.Duration, resourceAttrs map[string]any) Exporter {
 	return Exporter{
-		client:        client,
+		client:        pmetricotlp.NewGRPCClient(conn),
 		interval:      interval,
 		resourceAttrs: resourceAttrs,
-		stream:        nil,
 	}
 }
 
 func (e *Exporter) AddProducer(p ProducerConfig) {
 	e.producers = append(e.producers, p)
 }
 
-func (e *Exporter) makeStream(ctx context.Context) error {
-	slog.Debug("making new stream")
-	endpoint, err := e.client.ArrowMetrics(ctx)
-	if err != nil {
-		return err
-	}
-	p := arrow_record.NewProducer()
-	e.stream = &stream{
-		endpoint:      endpoint,
-		arrowProducer: p,
-	}
-	return nil
-}
-
 func (e *Exporter) report(ctx context.Context) error {
 	m := pmetric.NewMetrics()
 	r := m.ResourceMetrics().AppendEmpty()
 	if err := r.Resource().Attributes().FromRaw(e.resourceAttrs); err != nil {
 		return err
 	}
 	for _, p := range e.producers {
-		slog.Debug("Running arrow metrics producer", "scope", p.ScopeName)
+		slog.Debug("Running metrics producer", "scope", p.ScopeName)
 		s := r.ScopeMetrics().AppendEmpty()
 		s.Scope().SetName(p.ScopeName)
 		ms := s.Metrics()
@@ -85,61 +64,25 @@ func (e *Exporter) report(ctx context.Context) error {
 	}
 
 	dpc := m.DataPointCount()
-	slog.Debug("About to report arrow metrics", "data points", dpc)
-
-	retriesRemaining := 1
-	var err error
-	var arrow *arrowpb.BatchArrowRecords
-	for retriesRemaining >= 0 {
-		retriesRemaining--
-		if e.stream == nil {
-			err = e.makeStream(ctx)
-			if err != nil {
-				// if we failed to create a new stream, don't retry.
-				// The point of the retry loop is to handle the stream
-				// legitimately going away e.g. due to the server
-				// having specified max_connection_age,
-				// not unexpected issues creating a new stream.
-				break
-			}
-		}
-		arrow, err = e.stream.arrowProducer.BatchArrowRecordsFromMetrics(m)
-		if err != nil {
-			slog.Warn("Error on produce", "error", err)
-			e.stream = nil
-			continue
-		}
-		start := time.Now()
-		err = e.stream.endpoint.Send(arrow)
-		if err != nil {
-			slog.Warn("Error on send", "error", err)
-			e.stream = nil
-			continue
-		}
-		batchStatus, err := e.stream.endpoint.Recv()
-		if err != nil {
-			slog.Warn("Error on recv", "error", err)
-			return err
-		}
-		if batchStatus.GetStatusCode() != arrowpb.StatusCode_OK {
-			slog.Warn("unexpected status code",
-				"status", batchStatus.GetStatusCode(),
-				"message", batchStatus.GetStatusMessage(),
-			)
-			return err
-		}
-
-		slog.Info("Send succeeded",
-			"data points", dpc,
-			"duration", time.Since(start),
-		)
-		break
-	}
+	slog.Debug("About to report otlp metrics", "data points", dpc)
 
+	req := pmetricotlp.NewExportRequestFromMetrics(m)
+	start := time.Now()
+	resp, err := e.client.Export(ctx, req)
 	if err != nil {
-		return err
+		return fmt.Errorf("otlp export failed: %w", err)
+	}
+	if ps := resp.PartialSuccess(); ps.RejectedDataPoints() > 0 || ps.ErrorMessage() != "" {
+		slog.Warn("otlp partial success",
+			"rejected", ps.RejectedDataPoints(),
+			"message", ps.ErrorMessage(),
+		)
 	}
 
+	slog.Info("Send succeeded",
+		"data points", dpc,
+		"duration", time.Since(start),
+	)
 	return nil
 }
 
@@ -158,7 +101,7 @@ func (e *Exporter) Collect(ctx context.Context) error {
 }
 
 func (e *Exporter) Start(ctx context.Context) error {
-	slog.Info("running arrow metrics exporter", "producers", len(e.producers))
+	slog.Info("running otlp metrics exporter", "producers", len(e.producers))
 	if len(e.producers) == 0 {
 		return errors.New("no producers configured")
 	}
@@ -170,7 +113,7 @@ func (e *Exporter) Start(ctx context.Context) error {
 			return nil
 		case <-tick.C:
 			if err := e.report(ctx); err != nil {
-				return fmt.Errorf("failed to send arrow metrics: %v", err)
+				return fmt.Errorf("failed to send otlp metrics: %v", err)
 			}
 			tick.Reset(addJitter(e.interval, 0.2))
 		}

go.mod

@@ -10,7 +10,6 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0
 	github.com/oklog/run v1.1.0
-	github.com/open-telemetry/otel-arrow v0.34.0
 	github.com/prometheus/client_golang v1.21.1
 	github.com/sirupsen/logrus v1.9.3
 	go.opentelemetry.io/collector/pdata v1.28.1
@@ -22,41 +21,24 @@ require (
 )
 
 require (
-	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
-	github.com/apache/arrow/go/v17 v17.0.0 // indirect
-	github.com/axiomhq/hyperloglog v0.0.0-20230201085229-3ddf4bad03dc // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
-	github.com/fxamacker/cbor/v2 v2.4.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/goccy/go-json v0.10.3 // indirect
-	github.com/google/flatbuffers v24.3.25+incompatible // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/compress v1.17.11 // indirect
-	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.62.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
-	github.com/x448/float16 v0.8.4 // indirect
-	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
-	golang.org/x/mod v0.21.0 // indirect
 	golang.org/x/net v0.37.0 // indirect
-	golang.org/x/sync v0.12.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/text v0.23.0 // indirect
-	golang.org/x/tools v0.25.0 // indirect
-	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
 )