Merge hardcoded/default configuration with OTEL config file

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

@@ -53,7 +53,7 @@ func Config(storageType string, zipkinHostPort string, factories config.Factorie
 		Extensions: configmodels.Extensions{"health_check": hc},
 		Service: configmodels.Service{
 			Extensions: []string{"health_check"},
-			Pipelines: map[string]*configmodels.Pipeline{
+			Pipelines: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: recTypes,

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

@@ -38,14 +38,14 @@ func TestDefaultConfig(t *testing.T) {
 		storageType    string
 		zipkinHostPort string
 		exporterTypes  []string
-		pipeline       map[string]*configmodels.Pipeline
+		pipeline       configmodels.Pipelines
 		err            string
 	}{
 		{
 			storageType:    "elasticsearch",
 			zipkinHostPort: disabledHostPort,
 			exporterTypes:  []string{elasticsearch.TypeStr},
-			pipeline: map[string]*configmodels.Pipeline{
+			pipeline: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: []string{"jaeger"},
@@ -57,7 +57,7 @@ func TestDefaultConfig(t *testing.T) {
 			storageType:    "cassandra",
 			zipkinHostPort: disabledHostPort,
 			exporterTypes:  []string{cassandra.TypeStr},
-			pipeline: map[string]*configmodels.Pipeline{
+			pipeline: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: []string{"jaeger"},
@@ -69,7 +69,7 @@ func TestDefaultConfig(t *testing.T) {
 			storageType:    "kafka",
 			zipkinHostPort: disabledHostPort,
 			exporterTypes:  []string{kafka.TypeStr},
-			pipeline: map[string]*configmodels.Pipeline{
+			pipeline: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: []string{"jaeger"},
@@ -81,7 +81,7 @@ func TestDefaultConfig(t *testing.T) {
 			storageType:    "cassandra,elasticsearch",
 			zipkinHostPort: disabledHostPort,
 			exporterTypes:  []string{cassandra.TypeStr, elasticsearch.TypeStr},
-			pipeline: map[string]*configmodels.Pipeline{
+			pipeline: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: []string{"jaeger"},
@@ -93,7 +93,7 @@ func TestDefaultConfig(t *testing.T) {
 			storageType:    "cassandra",
 			zipkinHostPort: ":9411",
 			exporterTypes:  []string{cassandra.TypeStr},
-			pipeline: map[string]*configmodels.Pipeline{
+			pipeline: configmodels.Pipelines{
 				"traces": {
 					InputType: configmodels.TracesDataType,
 					Receivers: []string{"jaeger", "zipkin"},

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

@@ -0,0 +1,34 @@
+// 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 defaults
+
+import (
+	"github.com/imdario/mergo"
+	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+)
+
+// MergeConfigs merges two configs.
+// The src is merged into dst.
+func MergeConfigs(dst, src *configmodels.Config) error {
+	if src == nil {
+		return nil
+	}
+	err := mergo.Merge(dst, src,
+		mergo.WithOverride)
+	if err != nil {
+		return err
+	}
+	return nil
+}

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

@@ -0,0 +1,176 @@
+// 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 defaults
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/open-telemetry/opentelemetry-collector/config"
+	"github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+	"github.com/open-telemetry/opentelemetry-collector/processor/attributesprocessor"
+	"github.com/open-telemetry/opentelemetry-collector/processor/batchprocessor"
+	"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"
+
+	"github.com/jaegertracing/jaeger/cmd/opentelemetry-collector/app/exporter/elasticsearch"
+	jConfig "github.com/jaegertracing/jaeger/pkg/config"
+)
+
+func TestMergeConfigs_nil(t *testing.T) {
+	cfg := &configmodels.Config{
+		Receivers: configmodels.Receivers{
+			"jaeger": &jaegerreceiver.Config{
+				RemoteSampling: &jaegerreceiver.RemoteSamplingConfig{StrategyFile: "file.json"},
+			},
+		},
+	}
+	err := MergeConfigs(cfg, nil)
+	require.NoError(t, err)
+	assert.Equal(t, cfg, cfg)
+}
+
+func TestMergeConfigs(t *testing.T) {
+	cfg := &configmodels.Config{
+		Receivers: configmodels.Receivers{
+			"jaeger": &jaegerreceiver.Config{
+				Protocols: map[string]*receiver.SecureReceiverSettings{
+					"grpc":           {ReceiverSettings: configmodels.ReceiverSettings{Endpoint: "def"}},
+					"thrift_compact": {ReceiverSettings: configmodels.ReceiverSettings{Endpoint: "def"}},
+				},
+			},
+		},
+		Processors: configmodels.Processors{
+			"batch": &batchprocessor.Config{
+				SendBatchSize: uint32(160),
+			},
+		},
+		Service: configmodels.Service{
+			Extensions: []string{"def", "def2"},
+			Pipelines: configmodels.Pipelines{
+				"traces": &configmodels.Pipeline{
+					Receivers:  []string{"jaeger"},
+					Processors: []string{"batch"},
+				},
+			},
+		},
+	}
+	overrideCfg := &configmodels.Config{
+		Receivers: configmodels.Receivers{
+			"jaeger": &jaegerreceiver.Config{
+				Protocols: map[string]*receiver.SecureReceiverSettings{
+					"grpc": {ReceiverSettings: configmodels.ReceiverSettings{Endpoint: "master_jaeger_url"}},
+				},
+			},
+			"zipkin": &zipkinreceiver.Config{
+				ReceiverSettings: configmodels.ReceiverSettings{
+					Endpoint: "master_zipkin_url",
+				},
+			},
+		},
+		Processors: configmodels.Processors{
+			"attributes": &attributesprocessor.Config{
+				Actions: []attributesprocessor.ActionKeyValue{{Key: "foo"}},
+			},
+		},
+		Service: configmodels.Service{
+			Extensions: []string{"def", "master1", "master2"},
+			Pipelines: configmodels.Pipelines{
+				"traces": &configmodels.Pipeline{
+					Receivers:  []string{"jaeger", "zipkin"},
+					Processors: []string{"attributes"},
+				},
+				"traces/2": &configmodels.Pipeline{
+					Processors: []string{"example"},
+				},
+			},
+		},
+	}
+	expected := &configmodels.Config{
+		Receivers: configmodels.Receivers{
+			"jaeger": &jaegerreceiver.Config{
+				Protocols: map[string]*receiver.SecureReceiverSettings{
+					"grpc":           {ReceiverSettings: configmodels.ReceiverSettings{Endpoint: "master_jaeger_url"}},
+					"thrift_compact": {ReceiverSettings: configmodels.ReceiverSettings{Endpoint: "def"}},
+				},
+			},
+			"zipkin": &zipkinreceiver.Config{
+				ReceiverSettings: configmodels.ReceiverSettings{
+					Endpoint: "master_zipkin_url",
+				},
+			},
+		},
+		Processors: configmodels.Processors{
+			"batch": &batchprocessor.Config{
+				SendBatchSize: uint32(160),
+			},
+			"attributes": &attributesprocessor.Config{
+				Actions: []attributesprocessor.ActionKeyValue{{Key: "foo"}},
+			},
+		},
+		Service: configmodels.Service{
+			Extensions: []string{"def", "master1", "master2"},
+			Pipelines: configmodels.Pipelines{
+				"traces": &configmodels.Pipeline{
+					Receivers:  []string{"jaeger", "zipkin"},
+					Processors: []string{"attributes"},
+				},
+				"traces/2": &configmodels.Pipeline{
+					Processors: []string{"example"},
+				},
+			},
+		},
+	}
+	err := MergeConfigs(cfg, overrideCfg)
+	require.NoError(t, err)
+	assert.Equal(t, expected, cfg)
+}
+
+func TestMergeConfigFiles(t *testing.T) {
+	testFiles := []string{"emptyoverride", "addprocessor", "multiplecomponents"}
+	v, _ := jConfig.Viperize(elasticsearch.DefaultOptions().AddFlags)
+	cmpts := Components(v)
+	for _, f := range testFiles {
+		t.Run(f, func(t *testing.T) {
+			cfg, err := loadConfig(cmpts, fmt.Sprintf("testdata/%s.yaml", f))
+			require.NoError(t, err)
+			override, err := loadConfig(cmpts, fmt.Sprintf("testdata/%s-override.yaml", f))
+			require.NoError(t, err)
+			merged, err := loadConfig(cmpts, fmt.Sprintf("testdata/%s-merged.yaml", f))
+			require.NoError(t, err)
+			err = MergeConfigs(cfg, override)
+			require.NoError(t, err)
+			assert.Equal(t, merged, cfg)
+		})
+	}
+}
+
+func loadConfig(factories config.Factories, file string) (*configmodels.Config, error) {
+	// config.Load fails to load an empty config
+	if file == "testdata/emptyoverride-override.yaml" {
+		return &configmodels.Config{}, nil
+	}
+	v := viper.New()
+	v.SetConfigFile(file)
+	err := v.ReadInConfig()
+	if err != nil {
+		return nil, fmt.Errorf("error loading config file %q: %v", file, err)
+	}
+	return config.Load(v, factories)
+}

cmd/opentelemetry-collector/app/defaults/testdata/addprocessor-merged.yaml

@@ -0,0 +1,21 @@
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_compact:
+      thrift_binary:
+
+processors:
+  queued_retry: {}
+  batch:
+    timeout: 5s
+
+exporters:
+  jaeger_elasticsearch:
+
+service:
+  pipelines:
+    traces:
+      receivers: [jaeger]
+      processors: [batch]
+      exporters: [jaeger_elasticsearch]

cmd/opentelemetry-collector/app/defaults/testdata/addprocessor-override.yaml

@@ -0,0 +1,24 @@
+# TODO OTEL config.Load() fails if there are no receiver/exporters
+# https://github.com/open-telemetry/opentelemetry-collector/issues/888
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+exporters:
+  jaeger_elasticsearch:
+    # TODO These two properties have to be overridden here if they were overridden in addprocessor.yaml
+    # If we do not override them here the default values from viper will override the values from "default/src" config.
+    # However in reality the default config is created from viper, hence when OTEL configuration is created it
+    # uses viper to create default values and then it sets the values from config.
+    # The viper is not used in tests hence we have to override it here.
+    #num_shards: 1
+    #num_replicas: 0
+
+processors:
+  batch:
+    timeout: 5s
+
+service:
+  pipelines:
+    traces:
+      processors: [batch]

cmd/opentelemetry-collector/app/defaults/testdata/addprocessor.yaml

@@ -0,0 +1,19 @@
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_compact:
+      thrift_binary:
+
+processors:
+  queued_retry: {}
+
+exporters:
+  jaeger_elasticsearch:
+
+service:
+  pipelines:
+    traces:
+      receivers: [jaeger]
+      processors: [queued_retry]
+      exporters: [jaeger_elasticsearch]

cmd/opentelemetry-collector/app/defaults/testdata/emptyoverride-merged.yaml

@@ -0,0 +1,22 @@
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_compact:
+      thrift_binary:
+
+processors:
+  queued_retry: {}
+
+exporters:
+  jaeger_elasticsearch:
+    server_urls: http://localhost:9200
+    num_shards: 1
+    num_replicas: 0
+
+service:
+  pipelines:
+    traces:
+      receivers: [jaeger]
+      processors: [queued_retry]
+      exporters: [jaeger_elasticsearch]

cmd/opentelemetry-collector/app/defaults/testdata/emptyoverride.yaml

@@ -0,0 +1,22 @@
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+      thrift_compact:
+      thrift_binary:
+
+processors:
+  queued_retry: {}
+
+exporters:
+  jaeger_elasticsearch:
+    server_urls: http://localhost:9200
+    num_shards: 1
+    num_replicas: 0
+
+service:
+  pipelines:
+    traces:
+      receivers: [jaeger]
+      processors: [queued_retry]
+      exporters: [jaeger_elasticsearch]