Skip to content

Commit

Permalink
Configure Jaeger receiver and exporter by flags (#2241)
Browse files Browse the repository at this point in the history 
* Configure Jaeger receiver and exporter by flags

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Add config tests

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Refactor flags

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* improve test coverage

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* make scoped test

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Fix lint

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Move sampling to agent in test

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Use host port

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Order

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Add comment

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

* Bind to all ports

Signed-off-by: Pavol Loffay <ploffay@redhat.com>
  • Loading branch information
@pavolloffay
pavolloffay authored May 15, 2020
1 parent 0fe3e75 commit 75a1d93
Show file tree
Hide file tree
Showing 23 changed files with 623 additions and 89 deletions.
12 changes: 9 additions & 3 deletions cmd/agent/app/flags.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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)")
}
Expand All @@ -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
}

Expand Down
7 changes: 6 additions & 1 deletion cmd/agent/app/reporter/grpc/flags.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}

Expand Down
35 changes: 21 additions & 14 deletions cmd/collector/app/builder_flags.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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(collectorZipkinHTTPHostPort, ports.PortToHostPort(0), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's Zipkin server")
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)
}

Expand All @@ -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)
Expand Down
31 changes: 19 additions & 12 deletions cmd/opentelemetry-collector/app/defaults/default_config.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
168 changes: 165 additions & 3 deletions cmd/opentelemetry-collector/app/defaults/default_config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -162,6 +165,87 @@ 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",
},
zipkinHostPort: "localhost:55",
receivers: configmodels.Receivers{
"jaeger": &jaegerreceiver.Config{
TypeVal: "jaeger",
NameVal: "jaeger",
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{}
Expand Down Expand Up @@ -196,7 +280,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)
}
Expand All @@ -221,6 +305,84 @@ 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",
"--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",
},
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{
"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
Expand Down
1 change: 0 additions & 1 deletion cmd/opentelemetry-collector/app/exporter/jaegerexporter/.nocover
View file

This file was deleted.

Loading

0 comments on commit 75a1d93

Please sign in to comment.