Create Jaeger OTEL agent component

.gitignore

@@ -32,5 +32,6 @@ cmd/docs/*.1
 cmd/docs/*.yaml
 crossdock/crossdock-*
 run-crossdock.log
-cmd/opentelemetry-collector/opentelemetry-collector-*
+cmd/opentelemetry-collector/cmd/collector/opentelemetry-collector-*
+cmd/opentelemetry-collector/cmd/agent/opentelemetry-agent-*
 __pycache__

Makefile

@@ -259,9 +259,17 @@ endif
 .PHONY: build-otel-collector
 build-otel-collector: elasticsearch-mappings
 ifeq ($(GOARCH), s390x)
-	cd ${OTEL_COLLECTOR_DIR} && $(GOBUILD) -o ./opentelemetry-collector-$(GOOS)-$(GOARCH) $(BUILD_INFO) main.go
+	cd ${OTEL_COLLECTOR_DIR}/cmd/collector && $(GOBUILD) -o ./opentelemetry-collector-$(GOOS)-$(GOARCH) $(BUILD_INFO) main.go
 else
-	cd ${OTEL_COLLECTOR_DIR} && $(GOBUILD) -o ./opentelemetry-collector-$(GOOS) $(BUILD_INFO) main.go
+	cd ${OTEL_COLLECTOR_DIR}/cmd/collector && $(GOBUILD) -o ./opentelemetry-collector-$(GOOS) $(BUILD_INFO) main.go
+endif
+
+.PHONY: build-otel-agent
+build-otel-agent:
+ifeq ($(GOARCH), s390x)
+	cd ${OTEL_COLLECTOR_DIR}/cmd/agent && $(GOBUILD) -o ./opentelemetry-agent-$(GOOS)-$(GOARCH) $(BUILD_INFO) main.go
+else
+	cd ${OTEL_COLLECTOR_DIR}/cmd/agent && $(GOBUILD) -o ./opentelemetry-agent-$(GOOS) $(BUILD_INFO) main.go
 endif
 
 .PHONY: build-ingester
@@ -292,7 +300,7 @@ build-binaries-s390x:
 	GOOS=linux GOARCH=s390x $(MAKE) build-platform-binaries
 
 .PHONY: build-platform-binaries
-build-platform-binaries: build-agent build-collector build-query build-ingester build-all-in-one build-examples build-tracegen build-otel-collector
+build-platform-binaries: build-agent build-collector build-query build-ingester build-all-in-one build-examples build-tracegen build-otel-collector build-otel-agent
 
 .PHONY: build-all-platforms
 build-all-platforms: build-binaries-linux build-binaries-windows build-binaries-darwin build-binaries-s390x
@@ -310,10 +318,12 @@ docker-images-elastic:
 
 .PHONY: docker-images-jaeger-backend
 docker-images-jaeger-backend:
-	for component in agent collector query ingester opentelemetry-collector ; do \
+	for component in agent collector query ingester ; do \
 		docker build -t $(DOCKER_NAMESPACE)/jaeger-$$component:${DOCKER_TAG} cmd/$$component ; \
 		echo "Finished building $$component ==============" ; \
 	done
+	docker build -t $(DOCKER_NAMESPACE)/jaeger-opentelemetry-collector:${DOCKER_TAG} -f ${OTEL_COLLECTOR_DIR}/cmd/collector/Dockerfile cmd/opentelemetry-collector/cmd/collector
+	docker build -t $(DOCKER_NAMESPACE)/jaeger-opentelemetry-agent:${DOCKER_TAG} -f ${OTEL_COLLECTOR_DIR}/cmd/agent/Dockerfile cmd/opentelemetry-collector/cmd/agent
 
 .PHONY: docker-images-tracegen
 docker-images-tracegen:

cmd/opentelemetry-collector/app/defaults/default_config.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector/config"
 	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+	"github.com/open-telemetry/opentelemetry-collector/extension/healthcheckextension"
 	"github.com/open-telemetry/opentelemetry-collector/receiver"
 	"github.com/open-telemetry/opentelemetry-collector/receiver/jaegerreceiver"
 	"github.com/open-telemetry/opentelemetry-collector/receiver/zipkinreceiver"
@@ -30,9 +31,17 @@ import (
 	"github.com/jaegertracing/jaeger/ports"
 )
 
-// Config creates default configuration.
+const (
+	gRPCEndpoint             = "localhost:14250"
+	httpThriftBinaryEndpoint = "localhost:14268"
+	udpThriftCompactEndpoint = "localhost:6831"
+	udpThriftBinaryEndpoint  = "localhost:6832"
+	httpSamplingEndpoint     = "localhost:5778"
+)
+
+// CollectorConfig creates default collector configuration.
 // It enables default Jaeger receivers, processors and exporters.
-func Config(storageType string, zipkinHostPort string, factories config.Factories) (*configmodels.Config, error) {
+func CollectorConfig(storageType string, zipkinHostPort string, factories config.Factories) (*configmodels.Config, error) {
 	exporters, err := createExporters(storageType, factories)
 	if err != nil {
 		return nil, err
@@ -41,7 +50,7 @@ func Config(storageType string, zipkinHostPort string, factories config.Factorie
 	for _, v := range exporters {
 		expTypes = append(expTypes, string(v.Type()))
 	}
-	receivers := createReceivers(zipkinHostPort, factories)
+	receivers := createCollectorReceivers(zipkinHostPort, factories)
 	recTypes := []string{}
 	for _, v := range receivers {
 		recTypes = append(recTypes, string(v.Type()))
@@ -64,29 +73,19 @@ func Config(storageType string, zipkinHostPort string, factories config.Factorie
 	}, nil
 }
 
-func createReceivers(zipkinHostPort string, factories config.Factories) configmodels.Receivers {
+func createCollectorReceivers(zipkinHostPort string, factories config.Factories) configmodels.Receivers {
 	jaeger := factories.Receivers["jaeger"].CreateDefaultConfig().(*jaegerreceiver.Config)
 	// TODO load and serve sampling strategies
 	// TODO bind sampling strategies file
 	jaeger.Protocols = map[string]*receiver.SecureReceiverSettings{
 		"grpc": {
 			ReceiverSettings: configmodels.ReceiverSettings{
-				Endpoint: "localhost:14250",
+				Endpoint: gRPCEndpoint,
 			},
 		},
 		"thrift_http": {
 			ReceiverSettings: configmodels.ReceiverSettings{
-				Endpoint: "localhost:14268",
-			},
-		},
-		"thrift_compact": {
-			ReceiverSettings: configmodels.ReceiverSettings{
-				Endpoint: "localhost:6831",
-			},
-		},
-		"thrift_binary": {
-			ReceiverSettings: configmodels.ReceiverSettings{
-				Endpoint: "localhost:6832",
+				Endpoint: httpThriftBinaryEndpoint,
 			},
 		},
 	}
@@ -120,3 +119,45 @@ func createExporters(storageTypes string, factories config.Factories) (configmod
 	}
 	return exporters, nil
 }
+
+// AgentConfig creates default agent configuration.
+// It enables Jaeger receiver with UDP endpoints and Jaeger exporter.
+func AgentConfig(factories config.Factories) *configmodels.Config {
+	jaegerExporter := factories.Exporters["jaeger"]
+	hc := factories.Extensions["health_check"].CreateDefaultConfig().(*healthcheckextension.Config)
+	return &configmodels.Config{
+		Receivers:  createAgentReceivers(factories),
+		Exporters:  configmodels.Exporters{"jaeger": jaegerExporter.CreateDefaultConfig()},
+		Extensions: configmodels.Extensions{"health_check": hc},
+		Service: configmodels.Service{
+			Extensions: []string{"health_check"},
+			Pipelines: map[string]*configmodels.Pipeline{
+				"traces": {
+					InputType: configmodels.TracesDataType,
+					Receivers: []string{"jaeger"},
+					Exporters: []string{"jaeger"},
+				},
+			},
+		},
+	}
+}
+
+func createAgentReceivers(factories config.Factories) configmodels.Receivers {
+	jaeger := factories.Receivers["jaeger"].CreateDefaultConfig().(*jaegerreceiver.Config)
+	jaeger.Protocols = map[string]*receiver.SecureReceiverSettings{
+		"thrift_compact": {
+			ReceiverSettings: configmodels.ReceiverSettings{
+				Endpoint: udpThriftCompactEndpoint,
+			},
+		},
+		"thrift_binary": {
+			ReceiverSettings: configmodels.ReceiverSettings{
+				Endpoint: udpThriftBinaryEndpoint,
+			},
+		},
+	}
+	recvs := map[string]configmodels.Receiver{
+		"jaeger": jaeger,
+	}
+	return recvs
+}

cmd/opentelemetry-collector/app/defaults/default_config_test.go

@@ -20,18 +20,22 @@ import (
 
 	"github.com/open-telemetry/opentelemetry-collector/config"
 	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+	"github.com/open-telemetry/opentelemetry-collector/exporter/jaegerexporter"
+	"github.com/open-telemetry/opentelemetry-collector/receiver/jaegerreceiver"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 
+	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/cassandra"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/elasticsearch"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/kafka"
+	jConfig "github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/ports"
 )
 
-func TestDefaultConfig(t *testing.T) {
+func TestDefaultCollectorConfig(t *testing.T) {
 	factories := Components(viper.New())
 	disabledHostPort := ports.PortToHostPort(0)
 	tests := []struct {
@@ -108,7 +112,7 @@ func TestDefaultConfig(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.storageType, func(t *testing.T) {
-			cfg, err := Config(test.storageType, test.zipkinHostPort, factories)
+			cfg, err := CollectorConfig(test.storageType, test.zipkinHostPort, factories)
 			if test.err != "" {
 				require.Nil(t, cfg)
 				assert.EqualError(t, err, test.err)
@@ -135,3 +139,24 @@ func TestDefaultConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestDefaultAgentConfig(t *testing.T) {
+	v, _ := jConfig.Viperize(grpc.AddFlags)
+	factories := Components(v)
+	cfg := AgentConfig(factories)
+	assert.Equal(t, configmodels.Service{
+		Extensions: []string{"health_check"},
+		Pipelines: configmodels.Pipelines{
+			"traces": &configmodels.Pipeline{
+				InputType: configmodels.TracesDataType,
+				Receivers: []string{"jaeger"},
+				Exporters: []string{"jaeger"},
+			},
+		},
+	}, cfg.Service)
+	assert.Equal(t, 0, len(cfg.Processors))
+	assert.Equal(t, 1, len(cfg.Receivers))
+	assert.IsType(t, &jaegerreceiver.Config{}, cfg.Receivers["jaeger"])
+	assert.Equal(t, 1, len(cfg.Exporters))
+	assert.IsType(t, &jaegerexporter.Config{}, cfg.Exporters["jaeger"])
+}

cmd/opentelemetry-collector/app/defaults/defaults.go

@@ -18,13 +18,15 @@ import (
 	"flag"
 
 	"github.com/open-telemetry/opentelemetry-collector/config"
+	otelJaegerEexporter "github.com/open-telemetry/opentelemetry-collector/exporter/jaegerexporter"
 	otelJaegerreceiver "github.com/open-telemetry/opentelemetry-collector/receiver/jaegerreceiver"
 	"github.com/open-telemetry/opentelemetry-collector/service/defaultcomponents"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/cassandra"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/elasticsearch"
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/jaegerexporter"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/kafka"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/receiver/jaegerreceiver"
 	storageCassandra "github.com/jaegertracing/jaeger/plugin/storage/cassandra"
@@ -59,9 +61,14 @@ func Components(v *viper.Viper) config.Factories {
 	factories.Exporters[cassandraExp.Type()] = cassandraExp
 	factories.Exporters[esExp.Type()] = esExp
 
-	otelJRec := factories.Receivers["jaeger"].(*otelJaegerreceiver.Factory)
+	jaegerRec := factories.Receivers["jaeger"].(*otelJaegerreceiver.Factory)
 	factories.Receivers["jaeger"] = &jaegerreceiver.Factory{
-		Wrapped: otelJRec,
+		Wrapped: jaegerRec,
+		Viper:   v,
+	}
+	jaegerExp := factories.Exporters["jaeger"].(*otelJaegerEexporter.Factory)
+	factories.Exporters["jaeger"] = &jaegerexporter.Factory{
+		Wrapped: jaegerExp,
 		Viper:   v,
 	}
 	return factories

cmd/opentelemetry-collector/app/defaults/defaults_test.go

@@ -17,17 +17,23 @@ package defaults
 import (
 	"testing"
 
-	"github.com/magiconair/properties/assert"
 	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+	"github.com/stretchr/testify/assert"
 
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/cassandra"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/elasticsearch"
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/jaegerexporter"
 	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/kafka"
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/receiver/jaegerreceiver"
 	jConfig "github.com/jaegertracing/jaeger/pkg/config"
 )
 
 func TestComponents(t *testing.T) {
-	v, _ := jConfig.Viperize(kafka.DefaultOptions().AddFlags, cassandra.DefaultOptions().AddFlags, elasticsearch.DefaultOptions().AddFlags)
+	v, _ := jConfig.Viperize(
+		kafka.DefaultOptions().AddFlags,
+		cassandra.DefaultOptions().AddFlags,
+		elasticsearch.DefaultOptions().AddFlags,
+	)
 	factories := Components(v)
 	assert.Equal(t, configmodels.Type("jaeger_kafka"), factories.Exporters[kafka.TypeStr].Type())
 	assert.Equal(t, configmodels.Type("jaeger_cassandra"), factories.Exporters[cassandra.TypeStr].Type())
@@ -43,4 +49,6 @@ func TestComponents(t *testing.T) {
 	esFactory := factories.Exporters[elasticsearch.TypeStr]
 	ec := esFactory.CreateDefaultConfig().(*elasticsearch.Config)
 	assert.Equal(t, []string{"http://127.0.0.1:9200"}, ec.GetPrimary().Servers)
+	assert.IsType(t, &jaegerreceiver.Factory{}, factories.Receivers["jaeger"])
+	assert.IsType(t, &jaegerexporter.Factory{}, factories.Exporters["jaeger"])
 }

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

@@ -0,0 +1,77 @@
+// Copyright (c) 2020 The Jaeger 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 jaegerexporter
+
+import (
+	"context"
+
+	"github.com/open-telemetry/opentelemetry-collector/component"
+	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+	"github.com/open-telemetry/opentelemetry-collector/exporter/jaegerexporter"
+	"github.com/spf13/viper"
+
+	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
+)
+
+// Factory wraps jaegerexporter.Factory and makes the default config configurable via viper.
+// For instance this enables using flags as default values in the config object.
+type Factory struct {
+	// Wrapped is Jaeger receiver.
+	Wrapped *jaegerexporter.Factory
+	// Viper is used to get configuration values for default configuration
+	Viper *viper.Viper
+}
+
+var _ component.ExporterFactory = (*Factory)(nil)
+
+// Type returns the type of the exporter.
+func (f Factory) Type() configmodels.Type {
+	return f.Wrapped.Type()
+}
+
+// CreateDefaultConfig returns default configuration of Factory.
+// This function implements OTEL component.ExporterFactoryBase interface.
+func (f Factory) CreateDefaultConfig() configmodels.Exporter {
+	repCfg := grpc.ConnBuilder{}
+	repCfg.InitFromViper(f.Viper)
+	cfg := f.Wrapped.CreateDefaultConfig().(*jaegerexporter.Config)
+	if len(repCfg.CollectorHostPorts) > 0 {
+		cfg.Endpoint = repCfg.CollectorHostPorts[0]
+	}
+	cfg.GRPCSettings.UseSecure = repCfg.TLS.Enabled
+	cfg.GRPCSettings.CertPemFile = repCfg.TLS.CertPath
+	cfg.GRPCSettings.ServerNameOverride = repCfg.TLS.ServerName
+	return cfg
+}
+
+// CreateTraceExporter creates Jaeger trace exporter.
+// This function implements OTEL component.ExporterFactory interface.
+func (f Factory) CreateTraceExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.TraceExporter, error) {
+	return f.Wrapped.CreateTraceExporter(ctx, params, cfg)
+}
+
+// CreateMetricsExporter creates a metrics exporter based on provided config.
+// This function implements component.ExporterFactory.
+func (f Factory) CreateMetricsExporter(
+	ctx context.Context,
+	params component.ExporterCreateParams,
+	cfg configmodels.Exporter,
+) (component.MetricsExporter, error) {
+	return f.Wrapped.CreateMetricsExporter(ctx, params, cfg)
+}