Fix translation from otlp.Request to pdata representation

CHANGELOG.md

@@ -9,6 +9,7 @@
 ### 💡 Enhancements 💡
 
 ### 🧰 Bug fixes 🧰
+- Fix translation from otlp.Request to pdata representation, changes to the returned pdata not all reflected to the otlp.Request (#5197)
 
 ## v0.49.0 Beta
 

pdata/internal/logs.go

@@ -15,25 +15,50 @@
 package internal // import "go.opentelemetry.io/collector/pdata/internal"
 
 import (
+	otlpcollectorlog "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/logs/v1"
 	otlplogs "go.opentelemetry.io/collector/pdata/internal/data/protogen/logs/v1"
 )
 
+// LogsToOtlp internal helper to convert Logs to otlp request representation.
+func LogsToOtlp(mw Logs) *otlpcollectorlog.ExportLogsServiceRequest {
+	return mw.orig
+}
+
+// LogsFromOtlp internal helper to convert otlp request representation to Logs.
+func LogsFromOtlp(orig *otlpcollectorlog.ExportLogsServiceRequest) Logs {
+	return Logs{orig: orig}
+}
+
+// LogsToProto internal helper to convert Logs to protobuf representation.
+func LogsToProto(l Logs) otlplogs.LogsData {
+	return otlplogs.LogsData{
+		ResourceLogs: l.orig.ResourceLogs,
+	}
+}
+
+// LogsFromProto internal helper to convert protobuf representation to Logs.
+func LogsFromProto(orig otlplogs.LogsData) Logs {
+	return Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{
+		ResourceLogs: orig.ResourceLogs,
+	}}
+}
+
 // Logs is the top-level struct that is propagated through the logs pipeline.
 // Use NewLogs to create new instance, zero-initialized instance is not valid for use.
 type Logs struct {
-	orig *otlplogs.LogsData
+	orig *otlpcollectorlog.ExportLogsServiceRequest
 }
 
 // NewLogs creates a new Logs.
 func NewLogs() Logs {
-	return Logs{orig: &otlplogs.LogsData{}}
+	return Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{}}
 }
 
 // MoveTo moves all properties from the current struct to dest
 // resetting the current instance to its zero value.
 func (ld Logs) MoveTo(dest Logs) {
 	*dest.orig = *ld.orig
-	*ld.orig = otlplogs.LogsData{}
+	*ld.orig = otlpcollectorlog.ExportLogsServiceRequest{}
 }
 
 // Clone returns a copy of Logs.

pdata/internal/logs_test.go

@@ -22,6 +22,7 @@ import (
 	goproto "google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 
+	otlpcollectorlog "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/logs/v1"
 	otlplogs "go.opentelemetry.io/collector/pdata/internal/data/protogen/logs/v1"
 )
 
@@ -51,17 +52,17 @@ func TestLogRecordCount(t *testing.T) {
 
 func TestLogRecordCountWithEmpty(t *testing.T) {
 	assert.Zero(t, NewLogs().LogRecordCount())
-	assert.Zero(t, Logs{orig: &otlplogs.LogsData{
+	assert.Zero(t, Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{
 		ResourceLogs: []*otlplogs.ResourceLogs{{}},
 	}}.LogRecordCount())
-	assert.Zero(t, Logs{orig: &otlplogs.LogsData{
+	assert.Zero(t, Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{
 		ResourceLogs: []*otlplogs.ResourceLogs{
 			{
 				ScopeLogs: []*otlplogs.ScopeLogs{{}},
 			},
 		},
 	}}.LogRecordCount())
-	assert.Equal(t, 1, Logs{orig: &otlplogs.LogsData{
+	assert.Equal(t, 1, Logs{orig: &otlpcollectorlog.ExportLogsServiceRequest{
 		ResourceLogs: []*otlplogs.ResourceLogs{
 			{
 				ScopeLogs: []*otlplogs.ScopeLogs{
@@ -74,10 +75,11 @@ func TestLogRecordCountWithEmpty(t *testing.T) {
 	}}.LogRecordCount())
 }
 
-func TestToFromLogProto(t *testing.T) {
-	logs := LogsFromOtlp(&otlplogs.LogsData{})
+func TestToFromLogOtlp(t *testing.T) {
+	otlp := &otlpcollectorlog.ExportLogsServiceRequest{}
+	logs := LogsFromOtlp(otlp)
 	assert.EqualValues(t, NewLogs(), logs)
-	assert.EqualValues(t, &otlplogs.LogsData{}, logs.orig)
+	assert.EqualValues(t, otlp, LogsToOtlp(logs))
 }
 
 func TestResourceLogsWireCompatibility(t *testing.T) {

pdata/internal/metrics.go

@@ -15,18 +15,43 @@
 package internal // import "go.opentelemetry.io/collector/pdata/internal"
 
 import (
+	otlpcollectormetrics "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/metrics/v1"
 	otlpmetrics "go.opentelemetry.io/collector/pdata/internal/data/protogen/metrics/v1"
 )
 
+// MetricsToOtlp internal helper to convert Metrics to otlp request representation.
+func MetricsToOtlp(mw Metrics) *otlpcollectormetrics.ExportMetricsServiceRequest {
+	return mw.orig
+}
+
+// MetricsFromOtlp internal helper to convert otlp request representation to Metrics.
+func MetricsFromOtlp(orig *otlpcollectormetrics.ExportMetricsServiceRequest) Metrics {
+	return Metrics{orig: orig}
+}
+
+// MetricsToProto internal helper to convert Metrics to protobuf representation.
+func MetricsToProto(l Metrics) otlpmetrics.MetricsData {
+	return otlpmetrics.MetricsData{
+		ResourceMetrics: l.orig.ResourceMetrics,
+	}
+}
+
+// MetricsFromProto internal helper to convert protobuf representation to Metrics.
+func MetricsFromProto(orig otlpmetrics.MetricsData) Metrics {
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
+		ResourceMetrics: orig.ResourceMetrics,
+	}}
+}
+
 // Metrics is the top-level struct that is propagated through the metrics pipeline.
 // Use NewMetrics to create new instance, zero-initialized instance is not valid for use.
 type Metrics struct {
-	orig *otlpmetrics.MetricsData
+	orig *otlpcollectormetrics.ExportMetricsServiceRequest
 }
 
 // NewMetrics creates a new Metrics.
 func NewMetrics() Metrics {
-	return Metrics{orig: &otlpmetrics.MetricsData{}}
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{}}
 }
 
 // Clone returns a copy of MetricData.
@@ -40,7 +65,7 @@ func (md Metrics) Clone() Metrics {
 // resetting the current instance to its zero value.
 func (md Metrics) MoveTo(dest Metrics) {
 	*dest.orig = *md.orig
-	*md.orig = otlpmetrics.MetricsData{}
+	*md.orig = otlpcollectormetrics.ExportMetricsServiceRequest{}
 }
 
 // ResourceMetrics returns the ResourceMetricsSlice associated with this Metrics.

pdata/internal/metrics_test.go

@@ -22,6 +22,7 @@ import (
 	goproto "google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/emptypb"
 
+	otlpcollectormetrics "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/metrics/v1"
 	otlpcommon "go.opentelemetry.io/collector/pdata/internal/data/protogen/common/v1"
 	otlpmetrics "go.opentelemetry.io/collector/pdata/internal/data/protogen/metrics/v1"
 	otlpresource "go.opentelemetry.io/collector/pdata/internal/data/protogen/resource/v1"
@@ -226,7 +227,7 @@ func TestMetricsMoveTo(t *testing.T) {
 }
 
 func TestOtlpToInternalReadOnly(t *testing.T) {
-	md := Metrics{orig: &otlpmetrics.MetricsData{
+	md := Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -314,7 +315,7 @@ func TestOtlpToInternalReadOnly(t *testing.T) {
 }
 
 func TestOtlpToFromInternalReadOnly(t *testing.T) {
-	md := MetricsFromOtlp(&otlpmetrics.MetricsData{
+	md := MetricsFromOtlp(&otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -346,7 +347,7 @@ func TestOtlpToFromInternalReadOnly(t *testing.T) {
 func TestOtlpToFromInternalGaugeMutating(t *testing.T) {
 	newAttributes := NewMapFromRaw(map[string]interface{}{"k": "v"})
 
-	md := MetricsFromOtlp(&otlpmetrics.MetricsData{
+	md := MetricsFromOtlp(&otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -429,7 +430,7 @@ func TestOtlpToFromInternalGaugeMutating(t *testing.T) {
 func TestOtlpToFromInternalSumMutating(t *testing.T) {
 	newAttributes := NewMapFromRaw(map[string]interface{}{"k": "v"})
 
-	md := MetricsFromOtlp(&otlpmetrics.MetricsData{
+	md := MetricsFromOtlp(&otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -514,7 +515,7 @@ func TestOtlpToFromInternalSumMutating(t *testing.T) {
 func TestOtlpToFromInternalHistogramMutating(t *testing.T) {
 	newAttributes := NewMapFromRaw(map[string]interface{}{"k": "v"})
 
-	md := MetricsFromOtlp(&otlpmetrics.MetricsData{
+	md := MetricsFromOtlp(&otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -598,7 +599,7 @@ func TestOtlpToFromInternalHistogramMutating(t *testing.T) {
 func TestOtlpToFromInternalExponentialHistogramMutating(t *testing.T) {
 	newAttributes := NewMapFromRaw(map[string]interface{}{"k": "v"})
 
-	md := MetricsFromOtlp(&otlpmetrics.MetricsData{
+	md := MetricsFromOtlp(&otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -707,7 +708,7 @@ func BenchmarkMetricsClone(b *testing.B) {
 }
 
 func BenchmarkOtlpToFromInternal_PassThrough(b *testing.B) {
-	req := &otlpmetrics.MetricsData{
+	req := &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -732,7 +733,7 @@ func BenchmarkOtlpToFromInternal_PassThrough(b *testing.B) {
 }
 
 func BenchmarkOtlpToFromInternal_Gauge_MutateOneLabel(b *testing.B) {
-	req := &otlpmetrics.MetricsData{
+	req := &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -758,7 +759,7 @@ func BenchmarkOtlpToFromInternal_Gauge_MutateOneLabel(b *testing.B) {
 }
 
 func BenchmarkOtlpToFromInternal_Sum_MutateOneLabel(b *testing.B) {
-	req := &otlpmetrics.MetricsData{
+	req := &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -784,7 +785,7 @@ func BenchmarkOtlpToFromInternal_Sum_MutateOneLabel(b *testing.B) {
 }
 
 func BenchmarkOtlpToFromInternal_HistogramPoints_MutateOneLabel(b *testing.B) {
-	req := &otlpmetrics.MetricsData{
+	req := &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				Resource: generateTestProtoResource(),
@@ -947,13 +948,13 @@ func generateTestProtoDoubleHistogramMetric() *otlpmetrics.Metric {
 }
 
 func generateMetricsEmptyResource() Metrics {
-	return Metrics{orig: &otlpmetrics.MetricsData{
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{{}},
 	}}
 }
 
 func generateMetricsEmptyInstrumentation() Metrics {
-	return Metrics{orig: &otlpmetrics.MetricsData{
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				ScopeMetrics: []*otlpmetrics.ScopeMetrics{{}},
@@ -963,7 +964,7 @@ func generateMetricsEmptyInstrumentation() Metrics {
 }
 
 func generateMetricsEmptyMetrics() Metrics {
-	return Metrics{orig: &otlpmetrics.MetricsData{
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				ScopeMetrics: []*otlpmetrics.ScopeMetrics{
@@ -977,7 +978,7 @@ func generateMetricsEmptyMetrics() Metrics {
 }
 
 func generateMetricsEmptyDataPoints() Metrics {
-	return Metrics{orig: &otlpmetrics.MetricsData{
+	return Metrics{orig: &otlpcollectormetrics.ExportMetricsServiceRequest{
 		ResourceMetrics: []*otlpmetrics.ResourceMetrics{
 			{
 				ScopeMetrics: []*otlpmetrics.ScopeMetrics{

pdata/internal/traces.go

@@ -15,25 +15,51 @@
 package internal // import "go.opentelemetry.io/collector/pdata/internal"
 
 import (
+	otlpcollectortrace "go.opentelemetry.io/collector/pdata/internal/data/protogen/collector/trace/v1"
 	otlptrace "go.opentelemetry.io/collector/pdata/internal/data/protogen/trace/v1"
 )
 
+// TracesToOtlp internal helper to convert Traces to otlp request representation.
+func TracesToOtlp(mw Traces) *otlpcollectortrace.ExportTraceServiceRequest {
+	return mw.orig
+}
+
+// TracesFromOtlp internal helper to convert otlp request representation to Traces.
+func TracesFromOtlp(orig *otlpcollectortrace.ExportTraceServiceRequest) Traces {
+	return Traces{orig: orig}
+}
+
+// TracesToProto internal helper to convert Traces to protobuf representation.
+func TracesToProto(mw Traces) otlptrace.TracesData {
+	return otlptrace.TracesData{
+		ResourceSpans: mw.orig.ResourceSpans,
+	}
+}
+
+// TracesFromProto internal helper to convert protobuf representation to Traces.
+func TracesFromProto(orig otlptrace.TracesData) Traces {
+	return Traces{orig: &otlpcollectortrace.ExportTraceServiceRequest{
+		ResourceSpans: orig.ResourceSpans,
+	}}
+}
+
 // Traces is the top-level struct that is propagated through the traces pipeline.
 // Use NewTraces to create new instance, zero-initialized instance is not valid for use.
 type Traces struct {
-	orig *otlptrace.TracesData
+	// When marhsal/unmarshal unless it is in the request for otlp protocol, convert to otlptrace.TracesData.
+	orig *otlpcollectortrace.ExportTraceServiceRequest
 }
 
 // NewTraces creates a new Traces.
 func NewTraces() Traces {
-	return Traces{orig: &otlptrace.TracesData{}}
+	return Traces{orig: &otlpcollectortrace.ExportTraceServiceRequest{}}
 }
 
 // MoveTo moves all properties from the current struct to dest
 // resetting the current instance to its zero value.
 func (td Traces) MoveTo(dest Traces) {
 	*dest.orig = *td.orig
-	*td.orig = otlptrace.TracesData{}
+	*td.orig = otlpcollectortrace.ExportTraceServiceRequest{}
 }
 
 // Clone returns a copy of Traces.