[chore] Create a single interface for the Receiver and Exporter parse…

…rs (open-telemetry#2287)

* Create a single interface for the Receiver and Exporter parsers

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

* Fix lint

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

* Use ComponentType

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

* Apply changes requested in code review

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

* Simplify getting the enabled components

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

* Fix lint

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

---------

Signed-off-by: Israel Blancas <iblancasa@gmail.com>

apis/v1alpha1/collector_webhook_test.go

@@ -277,6 +277,7 @@ func TestOTELColDefaultingWebhook(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test
 		t.Run(test.name, func(t *testing.T) {
 			cvw := &CollectorWebhook{
 				logger: logr.Discard(),
@@ -786,6 +787,7 @@ func TestOTELColValidatingWebhook(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test
 		t.Run(test.name, func(t *testing.T) {
 			cvw := &CollectorWebhook{
 				logger: logr.Discard(),

apis/v1alpha1/instrumentation_webhook_test.go

@@ -116,6 +116,7 @@ func TestInstrumentationValidatingWebhook(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test
 		t.Run(test.name, func(t *testing.T) {
 			ctx := context.Background()
 			if test.err == "" {

apis/v1alpha1/opampbridge_webhook_test.go

@@ -108,6 +108,7 @@ func TestOpAMPBridgeDefaultingWebhook(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test
 		t.Run(test.name, func(t *testing.T) {
 			webhook := &OpAMPBridgeWebhook{
 				logger: logr.Discard(),
@@ -294,6 +295,7 @@ func TestOpAMPBridgeValidatingWebhook(t *testing.T) {
 	}
 
 	for _, test := range tests {
+		test := test
 		t.Run(test.name, func(t *testing.T) {
 			webhook := &OpAMPBridgeWebhook{
 				logger: logr.Discard(),

controllers/reconcile_test.go

@@ -518,6 +518,7 @@ func TestOpenTelemetryCollectorReconciler_Reconcile(t *testing.T) {
 		},
 	}
 	for _, tt := range tests {
+		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			testContext := context.Background()
 			nsn := types.NamespacedName{Name: tt.args.params.OtelCol.Name, Namespace: tt.args.params.OtelCol.Namespace}
@@ -557,6 +558,7 @@ func TestOpenTelemetryCollectorReconciler_Reconcile(t *testing.T) {
 			}
 			// run the next set of checks
 			for pid, updateParam := range tt.args.updates {
+				updateParam := updateParam
 				existing := v1alpha1.OpenTelemetryCollector{}
 				found, err := populateObjectIfExists(t, &existing, nsn)
 				assert.True(t, found)
@@ -676,6 +678,7 @@ func TestOpAMPBridgeReconciler_Reconcile(t *testing.T) {
 		},
 	}
 	for _, tt := range tests {
+		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			testContext := context.Background()
 			nsn := types.NamespacedName{Name: tt.args.params.OpAMPBridge.Name, Namespace: tt.args.params.OpAMPBridge.Namespace}
@@ -710,6 +713,7 @@ func TestOpAMPBridgeReconciler_Reconcile(t *testing.T) {
 			}
 			// run the next set of checks
 			for pid, updateParam := range tt.args.updates {
+				updateParam := updateParam
 				existing := v1alpha1.OpAMPBridge{}
 				found, err := populateObjectIfExists(t, &existing, nsn)
 				assert.True(t, found)

internal/manifests/collector/adapters/config_to_ports.go

@@ -15,7 +15,7 @@
 package adapters
 
 import (
-	"errors"
+	"fmt"
 	"net"
 	"sort"
 	"strconv"
@@ -24,26 +24,24 @@ import (
 	"github.com/mitchellh/mapstructure"
 	corev1 "k8s.io/api/core/v1"
 
+	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser"
 	exporterParser "github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser/exporter"
 	receiverParser "github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser/receiver"
 )
 
-var (
-	// ErrNoExporters indicates that there are no exporters in the configuration.
-	ErrNoExporters = errors.New("no exporters available as part of the configuration")
+type ComponentType int
 
-	// ErrNoReceivers indicates that there are no receivers in the configuration.
-	ErrNoReceivers = errors.New("no receivers available as part of the configuration")
-
-	// ErrReceiversNotAMap indicates that the receivers property isn't a map of values.
-	ErrReceiversNotAMap = errors.New("receivers property in the configuration doesn't contain valid receivers")
-
-	// ErrExportersNotAMap indicates that the exporters property isn't a map of values.
-	ErrExportersNotAMap = errors.New("exporters property in the configuration doesn't contain valid exporters")
+const (
+	ComponentTypeReceiver ComponentType = iota
+	ComponentTypeExporter
 )
 
-// ConfigToExporterPorts converts the incoming configuration object into a set of service ports required by the exporters.
-func ConfigToExporterPorts(logger logr.Logger, config map[interface{}]interface{}) ([]corev1.ServicePort, error) {
+func (c ComponentType) String() string {
+	return [...]string{"receiver", "exporter"}[c]
+}
+
+// ConfigToComponentPorts converts the incoming configuration object into a set of service ports required by the exporters.
+func ConfigToComponentPorts(logger logr.Logger, cType ComponentType, config map[interface{}]interface{}) ([]corev1.ServicePort, error) {
 	// now, we gather which ports we might need to open
 	// for that, we get all the exporters and check their `endpoint` properties,
 	// extracting the port from it. The port name has to be a "DNS_LABEL", so, we try to make it follow the pattern:
@@ -52,119 +50,64 @@ func ConfigToExporterPorts(logger logr.Logger, config map[interface{}]interface{
 	// the exporter-qualifier is what comes after the slash in the exporter name, but typically nil
 	// examples:
 	// ```yaml
-	// exporters:
-	//   exampleexporter:
+	// components:
+	//   componentexample:
 	//     endpoint: 0.0.0.0:12345
-	//   exampleexporter/settings:
+	//   componentexample/settings:
 	//     endpoint: 0.0.0.0:12346
-	// in this case, we have two ports, named: "exampleexporter" and "exampleexporter-settings"
-	exportersProperty, ok := config["exporters"]
+	// in this case, we have 2 ports, named: "componentexample" and "componentexample-settings"
+	componentsProperty, ok := config[fmt.Sprintf("%ss", cType.String())]
 	if !ok {
-		return nil, ErrNoExporters
-	}
-	expEnabled := GetEnabledExporters(logger, config)
-	if expEnabled == nil {
-		return nil, ErrExportersNotAMap
+		return nil, fmt.Errorf("no %ss available as part of the configuration", cType)
 	}
-	exporters, ok := exportersProperty.(map[interface{}]interface{})
+
+	components, ok := componentsProperty.(map[interface{}]interface{})
 	if !ok {
-		return nil, ErrExportersNotAMap
+		return nil, fmt.Errorf("%ss doesn't contain valid components", cType.String())
+	}
+
+	compEnabled := getEnabledComponents(config, cType)
+
+	if compEnabled == nil {
+		return nil, fmt.Errorf("no enabled %ss available as part of the configuration", cType)
 	}
 
 	ports := []corev1.ServicePort{}
-	for key, val := range exporters {
-		// This check will pass only the enabled exporters,
+	for key, val := range components {
+		// This check will pass only the enabled components,
 		// then only the related ports will be opened.
-		if !expEnabled[key] {
+		if !compEnabled[key] {
 			continue
 		}
 		exporter, ok := val.(map[interface{}]interface{})
 		if !ok {
-			logger.V(2).Info("exporter doesn't seem to be a map of properties", "exporter", key)
+			logger.V(2).Info("component doesn't seem to be a map of properties", cType.String(), key)
 			exporter = map[interface{}]interface{}{}
 		}
 
-		exprtName := key.(string)
-		exprtParser, err := exporterParser.For(logger, exprtName, exporter)
-		if err != nil {
-			logger.V(2).Info("no parser found for '%s'", exprtName)
-			continue
+		cmptName := key.(string)
+		var cmptParser parser.ComponentPortParser
+		var err error
+		switch cType {
+		case ComponentTypeExporter:
+			cmptParser, err = exporterParser.For(logger, cmptName, exporter)
+		case ComponentTypeReceiver:
+			cmptParser, err = receiverParser.For(logger, cmptName, exporter)
 		}
 
-		exprtPorts, err := exprtParser.Ports()
 		if err != nil {
-			logger.Error(err, "parser for '%s' has returned an error: %w", exprtName, err)
+			logger.V(2).Info("no parser found for '%s'", cmptName)
 			continue
 		}
 
-		if len(exprtPorts) > 0 {
-			ports = append(ports, exprtPorts...)
-		}
-	}
-
-	sort.Slice(ports, func(i, j int) bool {
-		return ports[i].Name < ports[j].Name
-	})
-
-	return ports, nil
-}
-
-// ConfigToReceiverPorts converts the incoming configuration object into a set of service ports required by the receivers.
-func ConfigToReceiverPorts(logger logr.Logger, config map[interface{}]interface{}) ([]corev1.ServicePort, error) {
-	// now, we gather which ports we might need to open
-	// for that, we get all the receivers and check their `endpoint` properties,
-	// extracting the port from it. The port name has to be a "DNS_LABEL", so, we try to make it follow the pattern:
-	// ${instance.Name}-${receiver.name}-${receiver.qualifier}
-	// the receiver-name is typically the node name from the receivers map
-	// the receiver-qualifier is what comes after the slash in the receiver name, but typically nil
-	// examples:
-	// ```yaml
-	// receivers:
-	//   examplereceiver:
-	//     endpoint: 0.0.0.0:12345
-	//   examplereceiver/settings:
-	//     endpoint: 0.0.0.0:12346
-	// in this case, we have two ports, named: "examplereceiver" and "examplereceiver-settings"
-	receiversProperty, ok := config["receivers"]
-	if !ok {
-		return nil, ErrNoReceivers
-	}
-	recEnabled := GetEnabledReceivers(logger, config)
-	if recEnabled == nil {
-		return nil, ErrReceiversNotAMap
-	}
-	receivers, ok := receiversProperty.(map[interface{}]interface{})
-	if !ok {
-		return nil, ErrReceiversNotAMap
-	}
-
-	ports := []corev1.ServicePort{}
-	for key, val := range receivers {
-		// This check will pass only the enabled receivers,
-		// then only the related ports will be opened.
-		if !recEnabled[key] {
-			continue
-		}
-		receiver, ok := val.(map[interface{}]interface{})
-		if !ok {
-			logger.Info("receiver doesn't seem to be a map of properties", "receiver", key)
-			receiver = map[interface{}]interface{}{}
-		}
-
-		rcvrName := key.(string)
-		rcvrParser := receiverParser.For(logger, rcvrName, receiver)
-
-		rcvrPorts, err := rcvrParser.Ports()
+		exprtPorts, err := cmptParser.Ports()
 		if err != nil {
-			// should we break the process and return an error, or just ignore this faulty parser
-			// and let the other parsers add their ports to the service? right now, the best
-			// option seems to be to log the failures and move on, instead of failing them all
-			logger.Error(err, "parser for '%s' has returned an error: %w", rcvrName, err)
+			logger.Error(err, "parser for '%s' has returned an error: %w", cmptName, err)
 			continue
 		}
 
-		if len(rcvrPorts) > 0 {
-			ports = append(ports, rcvrPorts...)
+		if len(exprtPorts) > 0 {
+			ports = append(ports, exprtPorts...)
 		}
 	}
 
@@ -176,12 +119,12 @@ func ConfigToReceiverPorts(logger logr.Logger, config map[interface{}]interface{
 }
 
 func ConfigToPorts(logger logr.Logger, config map[interface{}]interface{}) []corev1.ServicePort {
-	ports, err := ConfigToReceiverPorts(logger, config)
+	ports, err := ConfigToComponentPorts(logger, ComponentTypeReceiver, config)
 	if err != nil {
 		logger.Error(err, "there was a problem while getting the ports from the receivers")
 	}
 
-	exporterPorts, err := ConfigToExporterPorts(logger, config)
+	exporterPorts, err := ConfigToComponentPorts(logger, ComponentTypeExporter, config)
 	if err != nil {
 		logger.Error(err, "there was a problem while getting the ports from the exporters")
 	}

internal/manifests/collector/adapters/config_to_ports_test.go

@@ -26,6 +26,7 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/adapters"
+	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser"
 	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser/receiver"
 )
 
@@ -83,7 +84,7 @@ func TestExtractPortsFromConfig(t *testing.T) {
 	require.NotEmpty(t, config)
 
 	// test
-	ports, err := adapters.ConfigToReceiverPorts(logger, config)
+	ports, err := adapters.ConfigToComponentPorts(logger, adapters.ComponentTypeReceiver, config)
 	assert.NoError(t, err)
 	assert.Len(t, ports, 10)
 
@@ -116,12 +117,12 @@ func TestNoPortsParsed(t *testing.T) {
 		configStr string
 	}{
 		{
-			expected:  adapters.ErrNoReceivers,
+			expected:  errors.New("no receivers available as part of the configuration"),
 			desc:      "empty",
 			configStr: "",
 		},
 		{
-			expected:  adapters.ErrReceiversNotAMap,
+			expected:  errors.New("receivers doesn't contain valid components"),
 			desc:      "not a map",
 			configStr: "receivers: some-string",
 		},
@@ -132,7 +133,7 @@ func TestNoPortsParsed(t *testing.T) {
 			require.NoError(t, err)
 
 			// test
-			ports, err := adapters.ConfigToReceiverPorts(logger, config)
+			ports, err := adapters.ConfigToComponentPorts(logger, adapters.ComponentTypeReceiver, config)
 
 			// verify
 			assert.Nil(t, ports)
@@ -161,7 +162,7 @@ func TestInvalidReceivers(t *testing.T) {
 			require.NoError(t, err)
 
 			// test
-			ports, err := adapters.ConfigToReceiverPorts(logger, config)
+			ports, err := adapters.ConfigToComponentPorts(logger, adapters.ComponentTypeReceiver, config)
 
 			// verify
 			assert.NoError(t, err)
@@ -179,7 +180,7 @@ func TestParserFailed(t *testing.T) {
 			return nil, errors.New("mocked error")
 		},
 	}
-	receiver.Register("mock", func(logger logr.Logger, name string, config map[interface{}]interface{}) receiver.ReceiverParser {
+	receiver.Register("mock", func(logger logr.Logger, name string, config map[interface{}]interface{}) parser.ComponentPortParser {
 		return mockParser
 	})
 
@@ -197,7 +198,7 @@ func TestParserFailed(t *testing.T) {
 	}
 
 	// test
-	ports, err := adapters.ConfigToReceiverPorts(logger, config)
+	ports, err := adapters.ConfigToComponentPorts(logger, adapters.ComponentTypeReceiver, config)
 
 	// verify
 	assert.Len(t, ports, 0)

internal/manifests/collector/adapters/config_validate.go

@@ -14,22 +14,13 @@
 
 package adapters
 
-import (
-	"github.com/go-logr/logr"
-)
+import "fmt"
 
 // Following Otel Doc: Configuring a receiver does not enable it. The receivers are enabled via pipelines within the service section.
-// GetEnabledReceivers returns all enabled receivers as a true flag set. If it can't find any receiver, it will return a nil interface.
-func GetEnabledReceivers(_ logr.Logger, config map[interface{}]interface{}) map[interface{}]bool {
-	return getEnabledComponents(config, "receivers")
-}
-
-func GetEnabledExporters(_ logr.Logger, config map[interface{}]interface{}) map[interface{}]bool {
-	return getEnabledComponents(config, "exporters")
-}
-
-func getEnabledComponents(config map[interface{}]interface{}, componentType string) map[interface{}]bool {
-	cfgComponents, ok := config[componentType]
+// getEnabledComponents returns all enabled components as a true flag set. If it can't find any receiver, it will return a nil interface.
+func getEnabledComponents(config map[interface{}]interface{}, componentType ComponentType) map[interface{}]bool {
+	componentTypePlural := fmt.Sprintf("%ss", componentType)
+	cfgComponents, ok := config[componentTypePlural]
 	if !ok {
 		return nil
 	}
@@ -85,7 +76,7 @@ func getEnabledComponents(config map[interface{}]interface{}, componentType stri
 					return nil
 				}
 				for pipSpecID, pipSpecCfg := range pipelineDesc {
-					if pipSpecID.(string) == componentType {
+					if pipSpecID.(string) == componentTypePlural {
 						receiversList, ok := pipSpecCfg.([]interface{})
 						if !ok {
 							continue