Convert Zipkin receiver and exporter to use OTLP and fix translation bugs

exporter/zipkinexporter/factory.go

@@ -62,7 +62,7 @@ func (f *Factory) CreateDefaultConfig() configmodels.Exporter {
 }
 
 // CreateTraceExporter creates a trace exporter based on this config.
-func (f *Factory) CreateTraceExporter(_ *zap.Logger, config configmodels.Exporter) (component.TraceExporterOld, error) {
+func (f *Factory) CreateTraceExporter(_ *zap.Logger, config configmodels.Exporter) (component.TraceExporter, error) {
 	zc := config.(*Config)
 
 	if zc.Endpoint == "" {
@@ -74,6 +74,6 @@ func (f *Factory) CreateTraceExporter(_ *zap.Logger, config configmodels.Exporte
 }
 
 // CreateMetricsExporter creates a metrics exporter based on this config.
-func (f *Factory) CreateMetricsExporter(_ *zap.Logger, _ configmodels.Exporter) (component.MetricsExporterOld, error) {
+func (f *Factory) CreateMetricsExporter(_ *zap.Logger, _ configmodels.Exporter) (component.MetricsExporter, error) {
 	return nil, configerror.ErrDataTypeIsNotSupported
 }

exporter/zipkinexporter/zipkin.go

@@ -20,14 +20,12 @@ import (
 	"fmt"
 	"net/http"
 
-	resourcepb "github.com/census-instrumentation/opencensus-proto/gen-go/resource/v1"
-	zipkinmodel "github.com/openzipkin/zipkin-go/model"
 	zipkinproto "github.com/openzipkin/zipkin-go/proto/v2"
 	zipkinreporter "github.com/openzipkin/zipkin-go/reporter"
 
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/consumer/consumerdata"
 	"go.opentelemetry.io/collector/consumer/consumererror"
+	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/collector/translator/trace/zipkin"
 )
@@ -46,12 +44,12 @@ type zipkinExporter struct {
 }
 
 // newTraceExporter creates an zipkin trace exporter.
-func newTraceExporter(config *Config) (component.TraceExporterOld, error) {
+func newTraceExporter(config *Config) (component.TraceExporter, error) {
 	ze, err := createZipkinExporter(config)
 	if err != nil {
 		return nil, err
 	}
-	zexp, err := exporterhelper.NewTraceExporterOld(config, ze.PushTraceData)
+	zexp, err := exporterhelper.NewTraceExporter(config, ze.PushTraceData)
 	if err != nil {
 		return nil, err
 	}
@@ -83,36 +81,30 @@ func createZipkinExporter(cfg *Config) (*zipkinExporter, error) {
 	return ze, nil
 }
 
-func (ze *zipkinExporter) PushTraceData(ctx context.Context, td consumerdata.TraceData) (int, error) {
-	tbatch := make([]*zipkinmodel.SpanModel, 0, len(td.Spans))
-	var resource *resourcepb.Resource = td.Resource
-
-	for _, span := range td.Spans {
-		zs, err := zipkin.OCSpanProtoToZipkin(td.Node, resource, span, ze.defaultServiceName)
-		if err != nil {
-			return len(td.Spans), consumererror.Permanent(fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err))
-		}
-		tbatch = append(tbatch, zs)
+func (ze *zipkinExporter) PushTraceData(ctx context.Context, td pdata.Traces) (int, error) {
+	tbatch, err := zipkin.InternalTracesToZipkinSpans(td)
+	if err != nil {
+		return td.SpanCount(), consumererror.Permanent(fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err))
 	}
 
 	body, err := ze.serializer.Serialize(tbatch)
 	if err != nil {
-		return len(td.Spans), consumererror.Permanent(fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err))
+		return td.SpanCount(), consumererror.Permanent(fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err))
 	}
 
 	req, err := http.NewRequestWithContext(ctx, "POST", ze.url, bytes.NewReader(body))
 	if err != nil {
-		return len(td.Spans), fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err)
+		return td.SpanCount(), fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err)
 	}
 	req.Header.Set("Content-Type", ze.serializer.ContentType())
 
 	resp, err := ze.client.Do(req)
 	if err != nil {
-		return len(td.Spans), fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err)
+		return td.SpanCount(), fmt.Errorf("failed to push trace data via Zipkin exporter: %w", err)
 	}
 	_ = resp.Body.Close()
 	if resp.StatusCode < 200 || resp.StatusCode > 299 {
-		return len(td.Spans), fmt.Errorf("failed the request with status code %d", resp.StatusCode)
+		return td.SpanCount(), fmt.Errorf("failed the request with status code %d", resp.StatusCode)
 	}
 	return 0, nil
 }

exporter/zipkinexporter/zipkin_test.go

@@ -75,7 +75,7 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 	mzr := newMockZipkinReporter(cst.URL)
 
 	// Run the Zipkin receiver to "receive spans upload from a client application"
-	zexp := processor.NewTraceFanOutConnectorOld([]consumer.TraceConsumerOld{tes})
+	zexp := processor.NewTraceFanOutConnector([]consumer.TraceConsumer{tes})
 	addr := testutil.GetAvailableLocalAddress(t)
 	cfg := &zipkinreceiver.Config{
 		ReceiverSettings: configmodels.ReceiverSettings{
@@ -316,7 +316,7 @@ func TestZipkinExporter_roundtripProto(t *testing.T) {
 	mzr.serializer = zipkinproto.SpanSerializer{}
 
 	// Run the Zipkin receiver to "receive spans upload from a client application"
-	zexp := processor.NewTraceFanOutConnectorOld([]consumer.TraceConsumerOld{tes})
+	zexp := processor.NewTraceFanOutConnector([]consumer.TraceConsumer{tes})
 	port := testutil.GetAvailablePort(t)
 	cfg := &zipkinreceiver.Config{
 		ReceiverSettings: configmodels.ReceiverSettings{

receiver/hostmetricsreceiver/internal/scraper/processscraper/process.go

@@ -58,7 +58,7 @@ func (m *processMetadata) initializeResource(resource pdata.Resource) {
 }
 
 func (m *processMetadata) insertPid(attr pdata.AttributeMap) {
-	attr.InsertInt(conventions.AttributeProcessID, int64(m.pid))
+	attr.InsertInt(conventions.AttributeProcessPID, int64(m.pid))
 }
 
 func (m *processMetadata) insertExecutable(attr pdata.AttributeMap) {
@@ -86,7 +86,7 @@ func (m *processMetadata) insertUsername(attr pdata.AttributeMap) {
 		return
 	}
 
-	attr.InsertString(conventions.AttributeProcessUsername, m.username)
+	attr.InsertString(conventions.AttributeProcessOwner, m.username)
 }
 
 // processHandles provides a wrapper around []*process.Process

receiver/hostmetricsreceiver/internal/scraper/processscraper/process_scraper_test.go

@@ -78,12 +78,12 @@ func TestScrapeMetrics(t *testing.T) {
 func assertResourceAttributes(t *testing.T, resourceMetrics pdata.ResourceMetricsSlice) {
 	for i := 0; i < resourceMetrics.Len(); i++ {
 		attr := resourceMetrics.At(0).Resource().Attributes()
-		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessID)
+		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessPID)
 		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessExecutableName)
 		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessExecutablePath)
 		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessCommand)
 		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessCommandLine)
-		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessUsername)
+		internal.AssertContainsAttribute(t, attr, conventions.AttributeProcessOwner)
 	}
 }
 

receiver/zipkinreceiver/factory.go

@@ -67,13 +67,13 @@ func (f *Factory) CreateTraceReceiver(
 	ctx context.Context,
 	logger *zap.Logger,
 	cfg configmodels.Receiver,
-	nextConsumer consumer.TraceConsumerOld,
+	nextConsumer consumer.TraceConsumer,
 ) (component.TraceReceiver, error) {
 	rCfg := cfg.(*Config)
 	return New(rCfg, nextConsumer)
 }
 
 // CreateMetricsReceiver creates a metrics receiver based on provided config.
-func (f *Factory) CreateMetricsReceiver(ctx context.Context, logger *zap.Logger, cfg configmodels.Receiver, nextConsumer consumer.MetricsConsumerOld) (component.MetricsReceiver, error) {
+func (f *Factory) CreateMetricsReceiver(ctx context.Context, logger *zap.Logger, cfg configmodels.Receiver, nextConsumer consumer.MetricsConsumer) (component.MetricsReceiver, error) {
 	return nil, configerror.ErrDataTypeIsNotSupported
 }

receiver/zipkinreceiver/factory_test.go

@@ -23,7 +23,7 @@ import (
 
 	"go.opentelemetry.io/collector/config/configcheck"
 	"go.opentelemetry.io/collector/config/configerror"
-	"go.opentelemetry.io/collector/consumer/consumerdata"
+	"go.opentelemetry.io/collector/consumer/pdata"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
@@ -36,7 +36,7 @@ func TestCreateDefaultConfig(t *testing.T) {
 type mockTraceConsumer struct {
 }
 
-func (m *mockTraceConsumer) ConsumeTraceData(ctx context.Context, td consumerdata.TraceData) error {
+func (m *mockTraceConsumer) ConsumeTraces(ctx context.Context, td pdata.Traces) error {
 	return nil
 }