Merge branch 'master' into kafka-otel

Signed-off-by: Joe Elliott <number101010@gmail.com>

cmd/opentelemetry/app/exporter/badgerexporter/factory.go

@@ -117,6 +117,16 @@ func (f Factory) CreateMetricsExporter(
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
 
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}
+
 func (f Factory) createStorageFactory(params component.ExporterCreateParams, cfg *Config) (storage.Factory, error) {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()

cmd/opentelemetry/app/exporter/cassandraexporter/factory.go

@@ -86,3 +86,13 @@ func (Factory) CreateMetricsExporter(
 ) (component.MetricsExporter, error) {
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
+
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}

cmd/opentelemetry/app/exporter/elasticsearchexporter/factory.go

@@ -90,3 +90,13 @@ func (Factory) CreateMetricsExporter(
 ) (component.MetricsExporter, error) {
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
+
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}

cmd/opentelemetry/app/exporter/elasticsearchexporter/spanstore.go

@@ -253,7 +253,7 @@ func bulkItemsToTraces(bulkItems []bulkItem) pdata.Traces {
 		if !spanData.InstrumentationLibrary.IsNil() {
 			spanData.InstrumentationLibrary.CopyTo(ispans.InstrumentationLibrary())
 		}
-		ispans.Spans().Append(&spanData.Span)
+		ispans.Spans().Append(spanData.Span)
 	}
 	return traces
 }

cmd/opentelemetry/app/exporter/grpcpluginexporter/factory.go

@@ -85,3 +85,13 @@ func (f Factory) CreateMetricsExporter(
 ) (component.MetricsExporter, error) {
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
+
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}

cmd/opentelemetry/app/exporter/jaegerexporter/jaeger_exporter.go

@@ -77,3 +77,13 @@ func (f Factory) CreateMetricsExporter(
 ) (component.MetricsExporter, error) {
 	return f.Wrapped.CreateMetricsExporter(ctx, params, cfg)
 }
+
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return f.Wrapped.CreateLogsExporter(ctx, params, cfg)
+}

cmd/opentelemetry/app/exporter/kafkaexporter/kafka_exporter.go

@@ -68,3 +68,13 @@ func (f Factory) CreateMetricsExporter(
 ) (component.MetricsExporter, error) {
 	return f.Wrapped.CreateMetricsExporter(ctx, params, cfg)
 }
+
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}

cmd/opentelemetry/app/exporter/memoryexporter/factory.go

@@ -102,6 +102,16 @@ func (Factory) CreateMetricsExporter(
 	return nil, configerror.ErrDataTypeIsNotSupported
 }
 
+// CreateLogsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateLogsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.LogsExporter, error) {
+	return nil, configerror.ErrDataTypeIsNotSupported
+}
+
 func (f Factory) createStorageFactory(params component.ExporterCreateParams, cfg configmodels.Exporter) (storage.Factory, error) {
 	config, ok := cfg.(*Config)
 	if !ok {

cmd/opentelemetry/app/processor/resourceprocessor/resource_processor.go

@@ -72,8 +72,8 @@ func (f Factory) GetTags() map[string]string {
 func (f Factory) CreateTraceProcessor(
 	ctx context.Context,
 	params component.ProcessorCreateParams,
-	nextConsumer consumer.TraceConsumer,
 	cfg configmodels.Processor,
+	nextConsumer consumer.TraceConsumer,
 ) (component.TraceProcessor, error) {
 	c := cfg.(*resourceprocessor.Config)
 	attributeKeys := map[string]bool{}
@@ -91,16 +91,28 @@ func (f Factory) CreateTraceProcessor(
 			})
 		}
 	}
-	return f.Wrapped.CreateTraceProcessor(ctx, params, nextConsumer, cfg)
+	return f.Wrapped.CreateTraceProcessor(ctx, params, cfg, nextConsumer)
 }
 
 // CreateMetricsProcessor creates a resource processor.
 // This function implements component.ProcessorFactoryOld.
 func (f Factory) CreateMetricsProcessor(
 	ctx context.Context,
 	params component.ProcessorCreateParams,
-	nextConsumer consumer.MetricsConsumer,
 	cfg configmodels.Processor,
+	nextConsumer consumer.MetricsConsumer,
 ) (component.MetricsProcessor, error) {
-	return f.Wrapped.CreateMetricsProcessor(ctx, params, nextConsumer, cfg)
+	return f.Wrapped.CreateMetricsProcessor(ctx, params, cfg, nextConsumer)
+}
+
+// CreateLogsProcessor creates a processor based on the config.
+// If the processor type does not support logs or if the config is not valid
+// error will be returned instead.
+func (f Factory) CreateLogsProcessor(
+	ctx context.Context,
+	params component.ProcessorCreateParams,
+	cfg configmodels.Processor,
+	nextConsumer consumer.LogsConsumer,
+) (component.LogsProcessor, error) {
+	return f.Wrapped.CreateLogsProcessor(ctx, params, cfg, nextConsumer)
 }

cmd/opentelemetry/app/processor/resourceprocessor/resource_processor_test.go

@@ -56,7 +56,7 @@ func TestDefaultValueFromViper(t *testing.T) {
 	}
 
 	cfg := f.CreateDefaultConfig().(*resourceprocessor.Config)
-	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, &componenttest.ExampleExporterConsumer{}, cfg)
+	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, cfg, &componenttest.ExampleExporterConsumer{})
 	require.NoError(t, err)
 	assert.NotNil(t, p)
 
@@ -81,7 +81,7 @@ func TestLegacyJaegerTagsOnly(t *testing.T) {
 	}
 
 	cfg := f.CreateDefaultConfig().(*resourceprocessor.Config)
-	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, &componenttest.ExampleExporterConsumer{}, cfg)
+	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, cfg, &componenttest.ExampleExporterConsumer{})
 	require.NoError(t, err)
 	assert.NotNil(t, p)
 
@@ -116,7 +116,7 @@ func TestLoadConfigAndFlags(t *testing.T) {
 	require.NotNil(t, colConfig)
 
 	cfg := colConfig.Processors[string(f.Type())].(*resourceprocessor.Config)
-	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, &componenttest.ExampleExporterConsumer{}, cfg)
+	p, err := f.CreateTraceProcessor(context.Background(), component.ProcessorCreateParams{Logger: zap.NewNop()}, cfg, &componenttest.ExampleExporterConsumer{})
 	require.NoError(t, err)
 	assert.NotNil(t, p)
 	assert.Equal(t, []processorhelper.ActionKeyValue{
@@ -136,9 +136,9 @@ func TestCreateMetricsProcessor(t *testing.T) {
 	f := &Factory{
 		Wrapped: resourceprocessor.NewFactory(),
 	}
-	mReceiver, err := f.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, &componenttest.ExampleExporterConsumer{}, &resourceprocessor.Config{
+	mReceiver, err := f.CreateMetricsProcessor(context.Background(), component.ProcessorCreateParams{}, &resourceprocessor.Config{
 		AttributesActions: []processorhelper.ActionKeyValue{{Key: "foo", Value: "val", Action: processorhelper.UPSERT}},
-	})
+	}, &componenttest.ExampleExporterConsumer{})
 	require.Nil(t, err)
 	assert.NotNil(t, mReceiver)
 }

cmd/opentelemetry/app/receiver/jaegerreceiver/jaeger_receiver.go

@@ -168,3 +168,15 @@ func (f *Factory) CreateMetricsReceiver(
 ) (component.MetricsReceiver, error) {
 	return f.Wrapped.CreateMetricsReceiver(ctx, params, cfg, nextConsumer)
 }
+
+// CreateLogsReceiver creates a receiver based on the config.
+// If the receiver type does not support logs or if the config is not valid
+// error will be returned instead.
+func (f Factory) CreateLogsReceiver(
+	ctx context.Context,
+	params component.ReceiverCreateParams,
+	cfg configmodels.Receiver,
+	nextConsumer consumer.LogsConsumer,
+) (component.LogsReceiver, error) {
+	return f.Wrapped.CreateLogsReceiver(ctx, params, cfg, nextConsumer)
+}

cmd/opentelemetry/app/receiver/zipkinreceiver/zipkin_recevier.go → cmd/opentelemetry/app/receiver/zipkinreceiver/zipkin_receiver.go

@@ -72,3 +72,15 @@ func (f Factory) CreateMetricsReceiver(
 ) (component.MetricsReceiver, error) {
 	return f.Wrapped.CreateMetricsReceiver(ctx, params, cfg, consumer)
 }
+
+// CreateLogsReceiver creates a receiver based on the config.
+// If the receiver type does not support logs or if the config is not valid
+// error will be returned instead.
+func (f Factory) CreateLogsReceiver(
+	ctx context.Context,
+	params component.ReceiverCreateParams,
+	cfg configmodels.Receiver,
+	nextConsumer consumer.LogsConsumer,
+) (component.LogsReceiver, error) {
+	return f.Wrapped.CreateLogsReceiver(ctx, params, cfg, nextConsumer)
+}

cmd/opentelemetry/cmd/agent/main.go

@@ -45,7 +45,7 @@ func main() {
 	}
 
 	ver := version.Get()
-	info := service.ApplicationStartInfo{
+	info := component.ApplicationStartInfo{
 		ExeName:  "jaeger-opentelemetry-agent",
 		LongName: "Jaeger OpenTelemetry Agent",
 		Version:  ver.GitVersion,
@@ -97,6 +97,6 @@ func main() {
 		handleErr(fmt.Errorf("could not load Jaeger configuration file %w", err))
 	}
 
-	err = svc.Start()
+	err = svc.Run()
 	handleErr(err)
 }

cmd/opentelemetry/cmd/all-in-one/main.go

@@ -66,7 +66,7 @@ func main() {
 	}
 
 	ver := version.Get()
-	info := service.ApplicationStartInfo{
+	info := component.ApplicationStartInfo{
 		ExeName:  "jaeger-opentelemetry-all-in-one",
 		LongName: "Jaeger OpenTelemetry all-in-one",
 		Version:  ver.GitVersion,
@@ -136,7 +136,7 @@ func main() {
 	}
 
 	go func() {
-		err = svc.Start()
+		err = svc.Run()
 		handleErr(err)
 	}()
 

cmd/opentelemetry/cmd/collector/main.go

@@ -47,7 +47,7 @@ func main() {
 	}
 
 	ver := version.Get()
-	info := service.ApplicationStartInfo{
+	info := component.ApplicationStartInfo{
 		ExeName:  "jaeger-opentelemetry-collector",
 		LongName: "Jaeger OpenTelemetry Collector",
 		Version:  ver.GitVersion,
@@ -115,6 +115,6 @@ func main() {
 		handleErr(fmt.Errorf("could not load Jaeger configuration file %w", err))
 	}
 
-	err = svc.Start()
+	err = svc.Run()
 	handleErr(err)
 }

cmd/opentelemetry/cmd/ingester/main.go

@@ -47,7 +47,7 @@ func main() {
 	}
 
 	ver := version.Get()
-	info := service.ApplicationStartInfo{
+	info := component.ApplicationStartInfo{
 		ExeName:  "jaeger-opentelemetry-ingester",
 		LongName: "Jaeger OpenTelemetry Ingester",
 		Version:  ver.GitVersion,
@@ -114,6 +114,6 @@ func main() {
 		handleErr(fmt.Errorf("could not load Jaeger configuration file %w", err))
 	}
 
-	err = svc.Start()
+	err = svc.Run()
 	handleErr(err)
 }

cmd/query/app/server.go

@@ -74,7 +74,7 @@ func NewServer(logger *zap.Logger, querySvc *querysvc.QueryService, options *Que
 		tracer:             tracer,
 		grpcServer:         grpcServer,
 		httpServer:         createHTTPServer(querySvc, options, tracer, logger),
-		separatePorts:      (grpcPort != httpPort),
+		separatePorts:      grpcPort != httpPort,
 		unavailableChannel: make(chan healthcheck.Status),
 	}, nil
 }

cmd/query/app/server_test.go

@@ -24,6 +24,7 @@ import (
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest/observer"
 
@@ -73,28 +74,44 @@ func TestServerBadHostPort(t *testing.T) {
 }
 
 func TestServerInUseHostPort(t *testing.T) {
-
-	for _, hostPort := range [2]string{":8080", ":8081"} {
-		conn, err := net.Listen("tcp", hostPort)
-		assert.NoError(t, err)
-
-		server, err := NewServer(zap.NewNop(), &querysvc.QueryService{},
-			&QueryOptions{HTTPHostPort: "127.0.0.1:8080", GRPCHostPort: "127.0.0.1:8081", BearerTokenPropagation: true},
-			opentracing.NoopTracer{})
-		assert.NoError(t, err)
-
-		err = server.Start()
-		assert.NotNil(t, err)
-		conn.Close()
-		if server.grpcConn != nil {
-			server.grpcConn.Close()
-		}
-		if server.httpConn != nil {
-			server.httpConn.Close()
-		}
-
+	const availableHostPort = "127.0.0.1:0"
+	conn, err := net.Listen("tcp", availableHostPort)
+	require.NoError(t, err)
+	defer func() { require.NoError(t, conn.Close()) }()
+
+	testCases := []struct {
+		name         string
+		httpHostPort string
+		grpcHostPort string
+	}{
+		{"HTTP host port clash", conn.Addr().String(), availableHostPort},
+		{"GRPC host port clash", availableHostPort, conn.Addr().String()},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			server, err := NewServer(
+				zap.NewNop(),
+				&querysvc.QueryService{},
+				&QueryOptions{
+					HTTPHostPort:           tc.httpHostPort,
+					GRPCHostPort:           tc.grpcHostPort,
+					BearerTokenPropagation: true,
+				},
+				opentracing.NoopTracer{},
+			)
+			assert.NoError(t, err)
+
+			err = server.Start()
+			assert.Error(t, err)
+
+			if server.grpcConn != nil {
+				server.grpcConn.Close()
+			}
+			if server.httpConn != nil {
+				server.httpConn.Close()
+			}
+		})
 	}
-
 }
 
 func TestServer(t *testing.T) {