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

exporter/zipkinexporter/config_test.go

@@ -15,12 +15,15 @@
 package zipkinexporter
 
 import (
+	"context"
 	"path"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configtest"
 )
@@ -48,4 +51,7 @@ func TestLoadConfig(t *testing.T) {
 	assert.Equal(t, "zipkin/2", e1.(*Config).Name())
 	assert.Equal(t, "https://somedest:1234/api/v2/spans", e1.(*Config).Endpoint)
 	assert.Equal(t, "proto", e1.(*Config).Format)
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
+	_, err = factory.CreateTraceExporter(context.Background(), params, e1)
+	require.NoError(t, err)
 }

exporter/zipkinexporter/factory.go

@@ -36,14 +36,15 @@ const (
 	defaultServiceName string = "<missing service name>"
 )
 
-// NewFactory creates a factory for OTLP exporter.
+// NewFactory creates a factory for Zipkin exporter
 func NewFactory() component.ExporterFactory {
 	return exporterhelper.NewFactory(
 		typeStr,
 		createDefaultConfig,
 		exporterhelper.WithTraces(createTraceExporter))
 }
 
+// CreateDefaultConfig creates the default configuration for exporter.
 func createDefaultConfig() configmodels.Exporter {
 	return &Config{
 		ExporterSettings: configmodels.ExporterSettings{
@@ -60,12 +61,13 @@ func createDefaultConfig() configmodels.Exporter {
 	}
 }
 
+// CreateTraceExporter creates a trace exporter based on this config.
 func createTraceExporter(
 	_ context.Context,
 	_ component.ExporterCreateParams,
-	cfg configmodels.Exporter,
+	config configmodels.Exporter,
 ) (component.TraceExporter, error) {
-	zc := cfg.(*Config)
+	zc := config.(*Config)
 
 	if zc.Endpoint == "" {
 		// TODO https://github.com/open-telemetry/opentelemetry-collector/issues/215

exporter/zipkinexporter/factory_test.go

@@ -26,24 +26,28 @@ import (
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
-	cfg := createDefaultConfig()
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
 	assert.NotNil(t, cfg, "failed to create default config")
 	assert.NoError(t, configcheck.ValidateConfig(cfg))
 }
 
 func TestCreateInstanceViaFactory(t *testing.T) {
-	cfg := createDefaultConfig()
+	factory := NewFactory()
+
+	cfg := factory.CreateDefaultConfig()
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
 
 	// Default config doesn't have default endpoint so creating from it should
 	// fail.
-	ze, err := createTraceExporter(context.Background(), component.ExporterCreateParams{Logger: zap.NewNop()}, cfg)
+	ze, err := factory.CreateTraceExporter(context.Background(), params, cfg)
 	assert.Error(t, err)
 	assert.Nil(t, ze)
 
 	// URL doesn't have a default value so set it directly.
 	zeCfg := cfg.(*Config)
 	zeCfg.Endpoint = "http://some.location.org:9411/api/v2/spans"
-	ze, err = createTraceExporter(context.Background(), component.ExporterCreateParams{Logger: zap.NewNop()}, cfg)
+	ze, err = factory.CreateTraceExporter(context.Background(), params, cfg)
 	assert.NoError(t, err)
 	assert.NotNil(t, ze)
 }

exporter/zipkinexporter/test_utils.go

@@ -0,0 +1,53 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package zipkinexporter
+
+import (
+	"encoding/json"
+	"testing"
+
+	zipkinmodel "github.com/openzipkin/zipkin-go/model"
+	"github.com/stretchr/testify/require"
+)
+
+func unmarshalZipkinSpanArrayToMap(t *testing.T, jsonStr string) map[zipkinmodel.ID]*zipkinmodel.SpanModel {
+	var i interface{}
+
+	err := json.Unmarshal([]byte(jsonStr), &i)
+	require.NoError(t, err)
+
+	results := make(map[zipkinmodel.ID]*zipkinmodel.SpanModel)
+
+	switch x := i.(type) {
+	case []interface{}:
+		for _, j := range x {
+			span := jsonToSpan(t, j)
+			results[span.ID] = span
+		}
+	default:
+		span := jsonToSpan(t, x)
+		results[span.ID] = span
+	}
+	return results
+}
+
+func jsonToSpan(t *testing.T, j interface{}) *zipkinmodel.SpanModel {
+	b, err := json.Marshal(j)
+	require.NoError(t, err)
+	span := &zipkinmodel.SpanModel{}
+	err = span.UnmarshalJSON(b)
+	require.NoError(t, err)
+	return span
+}

exporter/zipkinexporter/zipkin.go

@@ -20,15 +20,13 @@ import (
 	"fmt"
 	"net/http"
 
-	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/consumererror"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
-	"go.opentelemetry.io/collector/translator/internaldata"
 	"go.opentelemetry.io/collector/translator/trace/zipkin"
 )
 
@@ -51,7 +49,7 @@ func newTraceExporter(config *Config) (component.TraceExporter, error) {
 	if err != nil {
 		return nil, err
 	}
-	zexp, err := exporterhelper.NewTraceExporter(config, ze.pushTraceData)
+	zexp, err := exporterhelper.NewTraceExporter(config, ze.PushTraceData)
 	if err != nil {
 		return nil, err
 	}
@@ -83,39 +81,30 @@ func createZipkinExporter(cfg *Config) (*zipkinExporter, error) {
 	return ze, nil
 }
 
-func (ze *zipkinExporter) pushTraceData(ctx context.Context, td pdata.Traces) (int, error) {
-	numSpans := td.SpanCount()
-	octds := internaldata.TraceDataToOC(td)
-
-	tbatch := make([]*zipkinmodel.SpanModel, 0, numSpans)
-	for _, octd := range octds {
-		for _, span := range octd.Spans {
-			zs, err := zipkin.OCSpanProtoToZipkin(octd.Node, octd.Resource, span, ze.defaultServiceName)
-			if err != nil {
-				return numSpans, 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 numSpans, 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 numSpans, 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 numSpans, 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 numSpans, 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

@@ -36,6 +36,8 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/receiver/zipkinreceiver"
 	"go.opentelemetry.io/collector/testutil"
 )
@@ -51,7 +53,7 @@ import (
 // The rest of the fields should match up exactly
 func TestZipkinExporter_roundtripJSON(t *testing.T) {
 	buf := new(bytes.Buffer)
-	var sizes []int64
+	sizes := []int64{}
 	cst := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		s, _ := io.Copy(buf, r.Body)
 		sizes = append(sizes, s)
@@ -65,15 +67,16 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 		},
 		Format: "json",
 	}
-	zexp, err := newTraceExporter(config)
+	tes, err := newTraceExporter(config)
 	assert.NoError(t, err)
-	require.NotNil(t, zexp)
+	require.NotNil(t, tes)
 
 	// The test requires the spans from zipkinSpansJSONJavaLibrary to be sent in a single batch, use
 	// a mock to ensure that this happens as intended.
 	mzr := newMockZipkinReporter(cst.URL)
 
 	// Run the Zipkin receiver to "receive spans upload from a client application"
+	zexp := processor.NewTraceFanOutConnector([]consumer.TraceConsumer{tes})
 	addr := testutil.GetAvailableLocalAddress(t)
 	cfg := &zipkinreceiver.Config{
 		ReceiverSettings: configmodels.ReceiverSettings{
@@ -114,7 +117,7 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 		  "tags": {"http.path": "/api","clnt/finagle.version": "6.45.0"}
 		},
 		{
-		  "traceId": "4d1e00c0db9010db86154a4ba6e91385","parentId": "86154a4ba6e91386","id": "4d1e00c0db9010db",
+		  "traceId": "4d1e00c0db9010db86154a4ba6e91385","parentId": "86154a4ba6e91386","id": "4d1e00c0db9010dc",
 		  "kind": "SERVER","name": "put",
 		  "timestamp": 1472470996199000,"duration": 207000,
 		  "localEndpoint": {"serviceName": "frontend","ipv6": "7::80:807f"},
@@ -124,23 +127,31 @@ func TestZipkinExporter_roundtripJSON(t *testing.T) {
 		    {"timestamp": 1472470996403000,"value": "bar"}
 		  ],
 		  "tags": {"http.path": "/api","clnt/finagle.version": "6.45.0"}
-		}]
-		`, `
-		[{
+		},
+		{
 		  "traceId": "4d1e00c0db9010db86154a4ba6e91385",
 		  "parentId": "86154a4ba6e91386",
-		  "id": "4d1e00c0db9010db",
+		  "id": "4d1e00c0db9010dd",
 		  "kind": "SERVER",
 		  "name": "put",
 		  "timestamp": 1472470996199000,
 		  "duration": 207000
 		}]
 		`}
 	for i, s := range wants {
-		want := testutil.GenerateNormalizedJSON(t, s)
+		want := unmarshalZipkinSpanArrayToMap(t, s)
 		gotBytes := buf.Next(int(sizes[i]))
-		got := testutil.GenerateNormalizedJSON(t, string(gotBytes))
-		assert.Equal(t, want, got)
+		got := unmarshalZipkinSpanArrayToMap(t, string(gotBytes))
+		for id, expected := range want {
+			actual, ok := got[id]
+			assert.True(t, ok)
+			assert.Equal(t, expected.ID, actual.ID)
+			assert.Equal(t, expected.Name, actual.Name)
+			assert.Equal(t, expected.TraceID, actual.TraceID)
+			assert.Equal(t, expected.Timestamp, actual.Timestamp)
+			assert.Equal(t, expected.Duration, actual.Duration)
+			assert.Equal(t, expected.Kind, actual.Kind)
+		}
 	}
 }
 
@@ -151,7 +162,7 @@ type mockZipkinReporter struct {
 	serializer zipkinreporter.SpanSerializer
 }
 
-var _ zipkinreporter.Reporter = (*mockZipkinReporter)(nil)
+var _ (zipkinreporter.Reporter) = (*mockZipkinReporter)(nil)
 
 func (r *mockZipkinReporter) Send(span zipkinmodel.SpanModel) {
 	r.batch = append(r.batch, &span)
@@ -235,7 +246,7 @@ const zipkinSpansJSONJavaLibrary = `
 {
   "traceId": "4d1e00c0db9010db86154a4ba6e91385",
   "parentId": "86154a4ba6e91386",
-  "id": "4d1e00c0db9010db",
+  "id": "4d1e00c0db9010dc",
   "kind": "SERVER",
   "name": "put",
   "timestamp": 1472470996199000,
@@ -267,7 +278,7 @@ const zipkinSpansJSONJavaLibrary = `
 {
   "traceId": "4d1e00c0db9010db86154a4ba6e91385",
   "parentId": "86154a4ba6e91386",
-  "id": "4d1e00c0db9010db",
+  "id": "4d1e00c0db9010dd",
   "kind": "SERVER",
   "name": "put",
   "timestamp": 1472470996199000,
@@ -283,7 +294,8 @@ func TestZipkinExporter_invalidFormat(t *testing.T) {
 		Format: "foobar",
 	}
 	f := NewFactory()
-	_, err := f.CreateTraceExporter(context.Background(), component.ExporterCreateParams{Logger: zap.NewNop()}, config)
+	params := component.ExporterCreateParams{Logger: zap.NewNop()}
+	_, err := f.CreateTraceExporter(context.Background(), params, config)
 	require.Error(t, err)
 }
 
@@ -304,7 +316,7 @@ func TestZipkinExporter_roundtripProto(t *testing.T) {
 		},
 		Format: "proto",
 	}
-	zexp, err := newTraceExporter(config)
+	tes, err := newTraceExporter(config)
 	require.NoError(t, err)
 
 	// The test requires the spans from zipkinSpansJSONJavaLibrary to be sent in a single batch, use
@@ -314,6 +326,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.NewTraceFanOutConnector([]consumer.TraceConsumer{tes})
 	port := testutil.GetAvailablePort(t)
 	cfg := &zipkinreceiver.Config{
 		ReceiverSettings: configmodels.ReceiverSettings{

internal/goldendataset/pict_tracing_input_defs.go

@@ -34,6 +34,7 @@ const (
 	ResourceK8sOnPrem PICTInputResource = "K8sOnPrem"
 	ResourceK8sCloud  PICTInputResource = "K8sCloud"
 	ResourceFaas      PICTInputResource = "Faas"
+	ResourceExec      PICTInputResource = "Exec"
 )
 
 // Enumerates the number and kind of instrumentation library instances that can be generated.

internal/goldendataset/resource_generator.go

@@ -38,6 +38,8 @@ func GenerateResource(rscID PICTInputResource) *otlpresource.Resource {
 		attrs = generateCloudK8sAttributes()
 	case ResourceFaas:
 		attrs = generateFassAttributes()
+	case ResourceExec:
+		attrs = generateExecAttributes()
 	default:
 		attrs = generateEmptyAttributes()
 	}
@@ -141,3 +143,14 @@ func generateFassAttributes() map[string]interface{} {
 	attrMap[conventions.AttributeCloudZone] = "us-central1-a"
 	return attrMap
 }
+
+func generateExecAttributes() map[string]interface{} {
+	attrMap := make(map[string]interface{})
+	attrMap[conventions.AttributeProcessExecutableName] = "otelcol"
+	attrMap[conventions.AttributeProcessCommandLine] =
+		"--config=/etc/otel-collector-config.yaml --mem-ballast-size-mib=683"
+	attrMap[conventions.AttributeProcessExecutablePath] = "/usr/local/bin/otelcol"
+	attrMap[conventions.AttributeProcessID] = 2020
+	attrMap[conventions.AttributeProcessOwner] = "otel"
+	return attrMap
+}

internal/goldendataset/resource_generator_test.go

@@ -24,7 +24,7 @@ import (
 
 func TestGenerateResource(t *testing.T) {
 	resourceIds := []PICTInputResource{ResourceNil, ResourceEmpty, ResourceVMOnPrem, ResourceVMCloud, ResourceK8sOnPrem,
-		ResourceK8sCloud, ResourceFaas}
+		ResourceK8sCloud, ResourceFaas, ResourceExec}
 	for _, rscID := range resourceIds {
 		rsc := GenerateResource(rscID)
 		if rscID == ResourceNil {