Configure Jaeger receiver and exporter by flags

cmd/agent/app/flags.go

@@ -30,7 +30,8 @@ const (
 	suffixServerQueueSize     = "server-queue-size"
 	suffixServerMaxPacketSize = "server-max-packet-size"
 	suffixServerHostPort      = "server-host-port"
-	httpServerHostPort        = "http-server.host-port"
+	// HTTPServerHostPort is the flag for HTTP endpoint
+	HTTPServerHostPort = "http-server.host-port"
 )
 
 var defaultProcessors = []struct {
@@ -52,8 +53,13 @@ func AddFlags(flags *flag.FlagSet) {
 		flags.Int(prefix+suffixServerMaxPacketSize, defaultMaxPacketSize, "max packet size for the UDP server")
 		flags.String(prefix+suffixServerHostPort, ":"+strconv.Itoa(p.port), "host:port for the UDP server")
 	}
+	AddOTELFlags(flags)
+}
+
+// AddOTELFlags adds flags that are exposed by OTEL collector
+func AddOTELFlags(flags *flag.FlagSet) {
 	flags.String(
-		httpServerHostPort,
+		HTTPServerHostPort,
 		defaultHTTPServerHostPort,
 		"host:port of the http server (e.g. for /sampling point and /baggageRestrictions endpoint)")
 }
@@ -70,7 +76,7 @@ func (b *Builder) InitFromViper(v *viper.Viper) *Builder {
 		b.Processors = append(b.Processors, *p)
 	}
 
-	b.HTTPServer.HostPort = portNumToHostPort(v.GetString(httpServerHostPort))
+	b.HTTPServer.HostPort = portNumToHostPort(v.GetString(HTTPServerHostPort))
 	return b
 }
 

cmd/agent/app/reporter/grpc/flags.go

@@ -39,9 +39,14 @@ var tlsFlagsConfig = tlscfg.ClientFlagsConfig{
 
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
-	flags.String(collectorHostPort, "", "Comma-separated string representing host:port of a static list of collectors to connect to directly")
 	flags.Uint(retry, defaultMaxRetry, "Sets the maximum number of retries for a call")
 	flags.Int(discoveryMinPeers, 3, "Max number of collectors to which the agent will try to connect at any given time")
+	AddOTELFlags(flags)
+}
+
+// AddOTELFlags adds flags that are exposed by OTEL collector
+func AddOTELFlags(flags *flag.FlagSet) {
+	flags.String(collectorHostPort, "", "Comma-separated string representing host:port of a static list of collectors to connect to directly")
 	tlsFlagsConfig.AddFlags(flags)
 }
 

cmd/collector/app/builder_flags.go

@@ -27,13 +27,15 @@ import (
 )
 
 const (
-	collectorDynQueueSizeMemory   = "collector.queue-size-memory"
-	collectorQueueSize            = "collector.queue-size"
-	collectorNumWorkers           = "collector.num-workers"
-	collectorHTTPPort             = "collector.http-port"
-	collectorGRPCPort             = "collector.grpc-port"
-	collectorHTTPHostPort         = "collector.http-server.host-port"
-	collectorGRPCHostPort         = "collector.grpc-server.host-port"
+	collectorDynQueueSizeMemory = "collector.queue-size-memory"
+	collectorQueueSize          = "collector.queue-size"
+	collectorNumWorkers         = "collector.num-workers"
+	collectorHTTPPort           = "collector.http-port"
+	collectorGRPCPort           = "collector.grpc-port"
+	// CollectorHTTPHostPort is the flag for collector HTTP port
+	CollectorHTTPHostPort = "collector.http-server.host-port"
+	// CollectorGRPCHostPort is the flag for collector gRPC port
+	CollectorGRPCHostPort         = "collector.grpc-server.host-port"
 	collectorZipkinHTTPPort       = "collector.zipkin.http-port"
 	collectorZipkinHTTPHostPort   = "collector.zipkin.host-port"
 	collectorTags                 = "collector.tags"
@@ -79,16 +81,21 @@ type CollectorOptions struct {
 func AddFlags(flags *flag.FlagSet) {
 	flags.Int(collectorQueueSize, DefaultQueueSize, "The queue size of the collector")
 	flags.Int(collectorNumWorkers, DefaultNumWorkers, "The number of workers pulling items from the queue")
-	flags.Int(collectorHTTPPort, 0, collectorHTTPPortWarning+" see --"+collectorHTTPHostPort)
-	flags.Int(collectorGRPCPort, 0, collectorGRPCPortWarning+" see --"+collectorGRPCHostPort)
+	flags.Int(collectorHTTPPort, 0, collectorHTTPPortWarning+" see --"+CollectorHTTPHostPort)
+	flags.Int(collectorGRPCPort, 0, collectorGRPCPortWarning+" see --"+CollectorGRPCHostPort)
 	flags.Int(collectorZipkinHTTPPort, 0, collectorZipkinHTTPPortWarning+" see --"+collectorZipkinHTTPHostPort)
-	flags.String(collectorHTTPHostPort, ports.PortToHostPort(ports.CollectorHTTP), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's HTTP server")
-	flags.String(collectorGRPCHostPort, ports.PortToHostPort(ports.CollectorGRPC), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's GRPC server")
 	flags.String(collectorZipkinHTTPHostPort, ports.PortToHostPort(0), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's Zipkin server")
 	flags.Uint(collectorDynQueueSizeMemory, 0, "(experimental) The max memory size in MiB to use for the dynamic queue.")
 	flags.String(collectorTags, "", "One or more tags to be added to the Process tags of all spans passing through this collector. Ex: key1=value1,key2=${envVar:defaultValue}")
 	flags.String(collectorZipkinAllowedOrigins, "*", "Comma separated list of allowed origins for the Zipkin collector service, default accepts all")
 	flags.String(collectorZipkinAllowedHeaders, "content-type", "Comma separated list of allowed headers for the Zipkin collector service, default content-type")
+	AddOTELFlags(flags)
+}
+
+// AddOTELFlags adds flags that are exposed by OTEL collector
+func AddOTELFlags(flags *flag.FlagSet) {
+	flags.String(CollectorHTTPHostPort, ports.PortToHostPort(ports.CollectorHTTP), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's HTTP server")
+	flags.String(CollectorGRPCHostPort, ports.PortToHostPort(ports.CollectorGRPC), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's GRPC server")
 	tlsFlagsConfig.AddFlags(flags)
 }
 
@@ -97,8 +104,8 @@ func (cOpts *CollectorOptions) InitFromViper(v *viper.Viper) *CollectorOptions {
 	cOpts.DynQueueSizeMemory = v.GetUint(collectorDynQueueSizeMemory) * 1024 * 1024 // we receive in MiB and store in bytes
 	cOpts.QueueSize = v.GetInt(collectorQueueSize)
 	cOpts.NumWorkers = v.GetInt(collectorNumWorkers)
-	cOpts.CollectorHTTPHostPort = getAddressFromCLIOptions(v.GetInt(collectorHTTPPort), v.GetString(collectorHTTPHostPort))
-	cOpts.CollectorGRPCHostPort = getAddressFromCLIOptions(v.GetInt(collectorGRPCPort), v.GetString(collectorGRPCHostPort))
+	cOpts.CollectorHTTPHostPort = getAddressFromCLIOptions(v.GetInt(collectorHTTPPort), v.GetString(CollectorHTTPHostPort))
+	cOpts.CollectorGRPCHostPort = getAddressFromCLIOptions(v.GetInt(collectorGRPCPort), v.GetString(CollectorGRPCHostPort))
 	cOpts.CollectorZipkinHTTPHostPort = getAddressFromCLIOptions(v.GetInt(collectorZipkinHTTPPort), v.GetString(collectorZipkinHTTPHostPort))
 	cOpts.CollectorTags = flags.ParseJaegerTags(v.GetString(collectorTags))
 	cOpts.CollectorZipkinAllowedOrigins = v.GetString(collectorZipkinAllowedOrigins)

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

@@ -76,19 +76,24 @@ func CollectorConfig(storageType string, zipkinHostPort string, factories config
 
 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": {
+	if jaeger.Protocols == nil {
+		jaeger.Protocols = map[string]*receiver.SecureReceiverSettings{}
+	}
+	// The CreateDefaultConfig is enabling protocols from flags
+	// we do not want to override it here
+	if _, ok := jaeger.Protocols["grpc"]; !ok {
+		jaeger.Protocols["grpc"] = &receiver.SecureReceiverSettings{
 			ReceiverSettings: configmodels.ReceiverSettings{
 				Endpoint: gRPCEndpoint,
 			},
-		},
-		"thrift_http": {
+		}
+	}
+	if _, ok := jaeger.Protocols["thrift_http"]; !ok {
+		jaeger.Protocols["thrift_http"] = &receiver.SecureReceiverSettings{
 			ReceiverSettings: configmodels.ReceiverSettings{
 				Endpoint: httpThriftBinaryEndpoint,
 			},
-		},
+		}
 	}
 	recvs := map[string]configmodels.Receiver{
 		"jaeger": jaeger,
@@ -159,17 +164,19 @@ func AgentConfig(factories config.Factories) *configmodels.Config {
 
 func createAgentReceivers(factories config.Factories) configmodels.Receivers {
 	jaeger := factories.Receivers["jaeger"].CreateDefaultConfig().(*jaegerreceiver.Config)
-	jaeger.Protocols = map[string]*receiver.SecureReceiverSettings{
-		"thrift_compact": {
+	if _, ok := jaeger.Protocols["thrift_compact"]; !ok {
+		jaeger.Protocols["thrift_compact"] = &receiver.SecureReceiverSettings{
 			ReceiverSettings: configmodels.ReceiverSettings{
 				Endpoint: udpThriftCompactEndpoint,
 			},
-		},
-		"thrift_binary": {
+		}
+	}
+	if _, ok := jaeger.Protocols["thrift_binary"]; !ok {
+		jaeger.Protocols["thrift_binary"] = &receiver.SecureReceiverSettings{
 			ReceiverSettings: configmodels.ReceiverSettings{
 				Endpoint: udpThriftBinaryEndpoint,
 			},
-		},
+		}
 	}
 	recvs := configmodels.Receivers{
 		"jaeger": jaeger,

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

@@ -20,16 +20,19 @@ import (
 	"testing"
 
 	"github.com/open-telemetry/opentelemetry-collector/config"
+	"github.com/open-telemetry/opentelemetry-collector/config/configgrpc"
 	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
 	"github.com/open-telemetry/opentelemetry-collector/exporter/jaegerexporter"
 	"github.com/open-telemetry/opentelemetry-collector/processor/resourceprocessor"
+	"github.com/open-telemetry/opentelemetry-collector/receiver"
 	"github.com/open-telemetry/opentelemetry-collector/receiver/jaegerreceiver"
+	"github.com/open-telemetry/opentelemetry-collector/receiver/zipkinreceiver"
 	"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"
 	"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/grpcplugin"
@@ -118,7 +121,7 @@ func TestDefaultCollectorConfig(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(test.storageType, func(t *testing.T) {
-			v, _ := jConfig.Viperize(grpc.AddFlags)
+			v, _ := jConfig.Viperize(app.AddComponentFlags)
 			factories := Components(v)
 			for key, val := range test.config {
 				v.Set(key, val)
@@ -162,6 +165,103 @@ func TestDefaultCollectorConfig(t *testing.T) {
 	}
 }
 
+func TestCreateCollectorReceivers(t *testing.T) {
+	tests := []struct {
+		name           string
+		args           []string
+		zipkinHostPort string
+		receivers      configmodels.Receivers
+	}{
+		{
+			name:           "defaultWithoutZipkin",
+			args:           []string{},
+			zipkinHostPort: ":0",
+			receivers: configmodels.Receivers{
+				"jaeger": &jaegerreceiver.Config{
+					TypeVal: "jaeger",
+					NameVal: "jaeger",
+					Protocols: map[string]*receiver.SecureReceiverSettings{
+						"grpc": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: gRPCEndpoint,
+							},
+						},
+						"thrift_http": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: httpThriftBinaryEndpoint,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "configurationViaFlags",
+			args: []string{
+				"--collector.grpc-server.host-port=host:11",
+				"--collector.grpc.tls.cert=cacert.crt",
+				"--collector.grpc.tls.key=keycert.crt",
+				"--collector.http-server.host-port=host2:22",
+				"--reporter.grpc.host-port=coll:33",
+				"--reporter.grpc.tls.enabled=true",
+				"--reporter.grpc.tls.ca=cacert.pem",
+				"--reporter.grpc.tls.cert=cert.pem",
+				"--reporter.grpc.tls.key=key.key",
+			},
+			zipkinHostPort: "localhost:55",
+			receivers: configmodels.Receivers{
+				"jaeger": &jaegerreceiver.Config{
+					TypeVal: "jaeger",
+					NameVal: "jaeger",
+					RemoteSampling: &jaegerreceiver.RemoteSamplingConfig{
+						GRPCSettings: configgrpc.GRPCSettings{
+							Endpoint: "coll:33",
+							TLSConfig: configgrpc.TLSConfig{
+								UseSecure:  true,
+								CaCert:     "cacert.pem",
+								ClientCert: "cert.pem",
+								ClientKey:  "key.key",
+							},
+						},
+					},
+					Protocols: map[string]*receiver.SecureReceiverSettings{
+						"grpc": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: "host:11",
+							},
+							TLSCredentials: &receiver.TLSCredentials{
+								CertFile: "cacert.crt",
+								KeyFile:  "keycert.crt",
+							},
+						},
+						"thrift_http": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: "host2:22",
+							},
+						},
+					},
+				},
+				"zipkin": &zipkinreceiver.Config{
+					ReceiverSettings: configmodels.ReceiverSettings{
+						NameVal:  "zipkin",
+						TypeVal:  "zipkin",
+						Endpoint: "localhost:55",
+					},
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			v, c := jConfig.Viperize(app.AddComponentFlags)
+			require.NoError(t, c.ParseFlags(test.args))
+			factories := Components(v)
+			recvs := createCollectorReceivers(test.zipkinHostPort, factories)
+			assert.Equal(t, test.receivers, recvs)
+		})
+	}
+}
+
 func TestDefaultAgentConfig(t *testing.T) {
 	tests := []struct {
 		config  map[string]interface{}
@@ -196,7 +296,7 @@ func TestDefaultAgentConfig(t *testing.T) {
 	}
 	for _, test := range tests {
 		t.Run(fmt.Sprintf("%v", test.config), func(t *testing.T) {
-			v, _ := jConfig.Viperize(grpc.AddFlags)
+			v, _ := jConfig.Viperize(app.AddComponentFlags)
 			for key, val := range test.config {
 				v.Set(key, val)
 			}
@@ -221,6 +321,65 @@ func TestDefaultAgentConfig(t *testing.T) {
 	}
 }
 
+func TestCreateAgentReceivers(t *testing.T) {
+	tests := []struct {
+		args      []string
+		receivers configmodels.Receivers
+	}{
+		{
+			args: []string{""},
+			receivers: configmodels.Receivers{
+				"jaeger": &jaegerreceiver.Config{
+					TypeVal: "jaeger",
+					NameVal: "jaeger",
+					Protocols: map[string]*receiver.SecureReceiverSettings{
+						"thrift_compact": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: udpThriftCompactEndpoint,
+							},
+						},
+						"thrift_binary": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: udpThriftBinaryEndpoint,
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			args: []string{"--processor.jaeger-binary.server-host-port=host:1", "--processor.jaeger-compact.server-host-port=host:2"},
+			receivers: configmodels.Receivers{
+				"jaeger": &jaegerreceiver.Config{
+					TypeVal: "jaeger",
+					NameVal: "jaeger",
+					Protocols: map[string]*receiver.SecureReceiverSettings{
+						"thrift_binary": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: "host:1",
+							},
+						},
+						"thrift_compact": {
+							ReceiverSettings: configmodels.ReceiverSettings{
+								Endpoint: "host:2",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%v", test.args), func(t *testing.T) {
+			v, c := jConfig.Viperize(app.AddComponentFlags)
+			require.NoError(t, c.ParseFlags(test.args))
+			factories := Components(v)
+			recvs := createAgentReceivers(factories)
+			assert.Equal(t, test.receivers, recvs)
+		})
+	}
+}
+
 func TestDefaultIngesterConfig(t *testing.T) {
 	tests := []struct {
 		storageType string