Enable batch and queued retry processors by default (#2330)

* Enable batch and queued retry processors by default Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Add godoc Signed-off-by: Pavol Loffay <ploffay@redhat.com> * Fix review comments Signed-off-by: Pavol Loffay <ploffay@redhat.com>
jaegertracing · Jul 8, 2020 · c549252 · c549252
1 parent 97d2319
commit c549252
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 27 deletions.
diff --git a/cmd/opentelemetry/app/defaultconfig/default_config.go b/cmd/opentelemetry/app/defaultconfig/default_config.go
@@ -23,6 +23,8 @@ import (
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/config/configprotocol"
+	"go.opentelemetry.io/collector/processor/batchprocessor"
+	"go.opentelemetry.io/collector/processor/queuedprocessor"
 	"go.opentelemetry.io/collector/processor/resourceprocessor"
 	"go.opentelemetry.io/collector/receiver/jaegerreceiver"
 	"go.opentelemetry.io/collector/receiver/zipkinreceiver"
@@ -71,11 +73,7 @@ func (c ComponentSettings) CreateDefaultConfig() (*configmodels.Config, error) {
 		return nil, err
 	}
 	receivers := createReceivers(c.ComponentType, c.ZipkinHostPort, c.Factories)
-	processors := configmodels.Processors{}
-	resProcessor := c.Factories.Processors["resource"].CreateDefaultConfig().(*resourceprocessor.Config)
-	if len(resProcessor.Labels) > 0 {
-		processors[resProcessor.Name()] = resProcessor
-	}
+	processors, processorNames := createProcessors(c.Factories)
 	hc := c.Factories.Extensions["health_check"].CreateDefaultConfig()
 	return &configmodels.Config{
 		Receivers:  receivers,
@@ -88,14 +86,31 @@ func (c ComponentSettings) CreateDefaultConfig() (*configmodels.Config, error) {
 				string(configmodels.TracesDataType): {
 					InputType:  configmodels.TracesDataType,
 					Receivers:  receiverNames(receivers),
-					Processors: processorNames(processors),
+					Processors: processorNames,
 					Exporters:  exporterNames(exporters),
 				},
 			},
 		},
 	}, nil
 }
 
+func createProcessors(factories config.Factories) (configmodels.Processors, []string) {
+	processors := configmodels.Processors{}
+	var names []string
+	resource := factories.Processors["resource"].CreateDefaultConfig().(*resourceprocessor.Config)
+	if len(resource.Labels) > 0 {
+		processors[resource.Name()] = resource
+		names = append(names, resource.Name())
+	}
+	batch := factories.Processors["batch"].CreateDefaultConfig().(*batchprocessor.Config)
+	processors[batch.Name()] = batch
+	names = append(names, batch.Name())
+	queuedRetry := factories.Processors["queued_retry"].CreateDefaultConfig().(*queuedprocessor.Config)
+	processors[queuedRetry.Name()] = queuedRetry
+	names = append(names, queuedRetry.Name())
+	return processors, names
+}
+
 func createReceivers(component ComponentType, zipkinHostPort string, factories config.Factories) configmodels.Receivers {
 	if component == Ingester {
 		kafkaReceiver := factories.Receivers[kafkareceiver.TypeStr].CreateDefaultConfig().(*kafkareceiver.Config)
@@ -188,14 +203,6 @@ func receiverNames(receivers configmodels.Receivers) []string {
 	return names
 }
 
-func processorNames(processors configmodels.Processors) []string {
-	var names []string
-	for _, v := range processors {
-		names = append(names, v.Name())
-	}
-	return names
-}
-
 func exporterNames(exporters configmodels.Exporters) []string {
 	var names []string
 	for _, v := range exporters {

diff --git a/cmd/opentelemetry/app/defaultconfig/default_config_test.go b/cmd/opentelemetry/app/defaultconfig/default_config_test.go
@@ -51,9 +51,10 @@ func TestService(t *testing.T) {
 				Extensions: []string{"health_check"},
 				Pipelines: configmodels.Pipelines{
 					"traces": &configmodels.Pipeline{
-						InputType: configmodels.TracesDataType,
-						Receivers: []string{"otlp", "jaeger"},
-						Exporters: []string{"jaeger"},
+						InputType:  configmodels.TracesDataType,
+						Receivers:  []string{"otlp", "jaeger"},
+						Processors: []string{"batch", "queued_retry"},
+						Exporters:  []string{"jaeger"},
 					},
 				},
 			},
@@ -70,7 +71,7 @@ func TestService(t *testing.T) {
 					"traces": &configmodels.Pipeline{
 						InputType:  configmodels.TracesDataType,
 						Receivers:  []string{"otlp", "jaeger"},
-						Processors: []string{"resource"},
+						Processors: []string{"resource", "batch", "queued_retry"},
 						Exporters:  []string{elasticsearchexporter.TypeStr, kafkaexporter.TypeStr, memoryexporter.TypeStr},
 					},
 				},
@@ -85,9 +86,10 @@ func TestService(t *testing.T) {
 				Extensions: []string{"health_check"},
 				Pipelines: configmodels.Pipelines{
 					"traces": &configmodels.Pipeline{
-						InputType: configmodels.TracesDataType,
-						Receivers: []string{kafkareceiver.TypeStr},
-						Exporters: []string{elasticsearchexporter.TypeStr},
+						InputType:  configmodels.TracesDataType,
+						Receivers:  []string{kafkareceiver.TypeStr},
+						Processors: []string{"batch", "queued_retry"},
+						Exporters:  []string{elasticsearchexporter.TypeStr},
 					},
 				},
 			},
@@ -101,9 +103,10 @@ func TestService(t *testing.T) {
 				Extensions: []string{"health_check"},
 				Pipelines: configmodels.Pipelines{
 					"traces": &configmodels.Pipeline{
-						InputType: configmodels.TracesDataType,
-						Receivers: []string{kafkareceiver.TypeStr},
-						Exporters: []string{cassandraexporter.TypeStr, elasticsearchexporter.TypeStr, grpcpluginexporter.TypeStr},
+						InputType:  configmodels.TracesDataType,
+						Receivers:  []string{kafkareceiver.TypeStr},
+						Processors: []string{"batch", "queued_retry"},
+						Exporters:  []string{cassandraexporter.TypeStr, elasticsearchexporter.TypeStr, grpcpluginexporter.TypeStr},
 					},
 				},
 			},
@@ -118,9 +121,10 @@ func TestService(t *testing.T) {
 				Extensions: []string{"health_check"},
 				Pipelines: configmodels.Pipelines{
 					"traces": &configmodels.Pipeline{
-						InputType: configmodels.TracesDataType,
-						Receivers: []string{"otlp", "jaeger", "zipkin"},
-						Exporters: []string{elasticsearchexporter.TypeStr},
+						InputType:  configmodels.TracesDataType,
+						Receivers:  []string{"otlp", "jaeger", "zipkin"},
+						Processors: []string{"batch", "queued_retry"},
+						Exporters:  []string{elasticsearchexporter.TypeStr},
 					},
 				},
 			},