Check exposed svc ports

pkg/collector/adapters/config_to_ports.go

@@ -51,14 +51,22 @@ func ConfigToReceiverPorts(logger logr.Logger, config map[interface{}]interface{
 	if !ok {
 		return nil, ErrNoReceivers
 	}
-
+	recEnabled, err := GetEnabledReceivers(logger, config)
+	if err != nil {
+		return nil, err
+	}
 	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)

pkg/collector/adapters/config_to_ports_test.go

@@ -62,7 +62,18 @@ func TestExtractPortsFromConfig(t *testing.T) {
         endpoint: 0.0.0.0:55555
   zipkin:
   zipkin/2:
-        endpoint: 0.0.0.0:33333
+    endpoint: 0.0.0.0:33333
+service:
+  pipelines:
+    metrics:
+      receivers: [examplereceiver, examplereceiver/settings]
+      exporters: [logging]
+    metrics/1:
+      receivers: [jaeger, jaeger/custom]
+      exporters: [logging]
+    metrics/2:
+      receivers: [otlp, otlp/2, zipkin]
+      exporters: [logging]
 `
 
 	// prepare
@@ -73,7 +84,7 @@ func TestExtractPortsFromConfig(t *testing.T) {
 	// test
 	ports, err := adapters.ConfigToReceiverPorts(logger, config)
 	assert.NoError(t, err)
-	assert.Len(t, ports, 12)
+	assert.Len(t, ports, 11)
 
 	// verify
 	expectedPorts := map[int32]bool{}
@@ -87,7 +98,6 @@ func TestExtractPortsFromConfig(t *testing.T) {
 	expectedPorts[int32(55681)] = false
 	expectedPorts[int32(55555)] = false
 	expectedPorts[int32(9411)] = false
-	expectedPorts[int32(33333)] = false
 
 	expectedNames := map[string]bool{}
 	expectedNames["examplereceiver"] = false
@@ -101,7 +111,6 @@ func TestExtractPortsFromConfig(t *testing.T) {
 	expectedNames["otlp-http-legacy"] = false
 	expectedNames["otlp-2-grpc"] = false
 	expectedNames["zipkin"] = false
-	expectedNames["zipkin-2"] = false
 
 	expectedAppProtocols := map[string]string{}
 	expectedAppProtocols["otlp-grpc"] = "grpc"
@@ -111,7 +120,6 @@ func TestExtractPortsFromConfig(t *testing.T) {
 	expectedAppProtocols["jaeger-grpc"] = "grpc"
 	expectedAppProtocols["otlp-2-grpc"] = "grpc"
 	expectedAppProtocols["zipkin"] = "http"
-	expectedAppProtocols["zipkin-2"] = "http"
 
 	// make sure we only have the ports in the set
 	for _, port := range ports {
@@ -175,11 +183,11 @@ func TestInvalidReceivers(t *testing.T) {
 	}{
 		{
 			"receiver isn't a map",
-			"receivers:\n  some-receiver: string",
+			"receivers:\n  some-receiver: string\nservice:\n  pipelines:\n    metrics:\n      receivers: [some-receiver]",
 		},
 		{
 			"receiver's endpoint isn't string",
-			"receivers:\n  some-receiver:\n    endpoint: 123",
+			"receivers:\n  some-receiver:\n    endpoint: 123\nservice:\n  pipelines:\n    metrics:\n      receivers: [some-receiver]",
 		},
 	} {
 		t.Run(tt.desc, func(t *testing.T) {
@@ -212,7 +220,14 @@ func TestParserFailed(t *testing.T) {
 
 	config := map[interface{}]interface{}{
 		"receivers": map[interface{}]interface{}{
-			"mock": map[interface{}]interface{}{},
+			"mock": map[string]interface{}{},
+		},
+		"service": map[interface{}]interface{}{
+			"pipelines": map[interface{}]interface{}{
+				"metrics": map[interface{}]interface{}{
+					"receivers": []interface{}{"mock"},
+				},
+			},
 		},
 	}
 

pkg/collector/adapters/config_validate.go

@@ -0,0 +1,118 @@
+// Copyright The OpenTelemetry 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 adapters
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/go-logr/logr"
+)
+
+var (
+	errNoPipeline = errors.New("no pipeline available as part of the configuration")
+)
+
+//Following Otel Doc: Configuring a receiver does not enable it. The receivers are enabled via pipelines within the service section.
+//ConfigValidate returns all receivers, setting them as true for enabled and false for non-configured services in pipeline set.
+func GetEnabledReceivers(logger logr.Logger, config map[interface{}]interface{}) (map[interface{}]bool, error) {
+	cfgReceivers, ok := config["receivers"]
+	if !ok {
+		return nil, ErrNoReceivers
+	}
+	receivers, ok := cfgReceivers.(map[interface{}]interface{})
+	if !ok {
+		return nil, ErrReceiversNotAMap
+	}
+	availableReceivers := map[interface{}]bool{}
+
+	for recvID := range receivers {
+
+		//Safe Cast
+		receiverID, ok := recvID.(string)
+		if !ok {
+			return nil, fmt.Errorf("ReceiverID is not a string: %v", receiverID)
+		}
+		//Getting all receivers present in the receivers section and setting them to false.
+		availableReceivers[receiverID] = false
+	}
+
+	cfgService, ok := config["service"].(map[interface{}]interface{})
+	if !ok {
+		return nil, errNoService
+	}
+
+	pipeline, ok := cfgService["pipelines"].(map[interface{}]interface{})
+	if !ok {
+		return nil, errNoPipeline
+	}
+	availablePipelines := map[string]bool{}
+
+	for pipID := range pipeline {
+		//Safe Cast
+		pipelineID, ok := pipID.(string)
+		if !ok {
+			return nil, fmt.Errorf("PipelineID is not a string: %v", pipelineID)
+		}
+		//Getting all the available pipelines.
+		availablePipelines[pipelineID] = true
+	}
+
+	if len(pipeline) > 0 {
+		for pipelineID, pipelineCfg := range pipeline {
+			//Safe Cast
+			pipelineV, ok := pipelineID.(string)
+			if !ok {
+				return nil, fmt.Errorf("PipelineID is not a string: %v", pipelineV)
+			}
+			//Condition will get information if there are multiple configured pipelines.
+			if len(pipelineV) > 0 {
+				pipelineDesc, ok := pipelineCfg.(map[interface{}]interface{})
+				if !ok {
+					return nil, fmt.Errorf("pipeline was not properly configured")
+				}
+				for pipSpecID, pipSpecCfg := range pipelineDesc {
+					if pipSpecID.(string) == "receivers" {
+						receiversList, ok := pipSpecCfg.([]interface{})
+						if !ok {
+							return nil, fmt.Errorf("no receivers on pipeline configuration %q", receiversList...)
+						}
+						// If receiversList is empty means that we haven't any enabled Receiver.
+						if len(receiversList) == 0 {
+							availableReceivers = nil
+						} else {
+							// All enabled receivers will be set as true
+							for _, recKey := range receiversList {
+								//Safe Cast
+								receiverKey, ok := recKey.(string)
+								if !ok {
+									return nil, fmt.Errorf("ReceiverKey is not a string: %v", receiverKey)
+								}
+								availableReceivers[receiverKey] = true
+							}
+						}
+						//Removing all non-enabled receivers
+						for recID, recKey := range availableReceivers {
+							if !(recKey) {
+								delete(availableReceivers, recID)
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+	return availableReceivers, nil
+}

pkg/collector/adapters/config_validate_test.go

@@ -0,0 +1,111 @@
+// Copyright The OpenTelemetry 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 adapters
+
+import (
+	"testing"
+
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var logger = logf.Log.WithName("unit-tests")
+
+func TestConfigValidate(t *testing.T) {
+	// prepare
+
+	// First Test - Exporters
+	configStr := `
+receivers:
+  httpd/mtls:
+    protocols:
+      http:
+        endpoint: mysite.local:55690
+  jaeger:
+    protocols:
+      grpc:
+  prometheus:
+    protocols:
+      grpc:
+
+processors:
+
+exporters:
+  logging:
+
+service:
+  pipelines:
+    metrics:
+      receivers: [httpd/mtls, jaeger]
+      exporters: [logging]
+    metrics/1:
+      receivers: [httpd/mtls, jaeger]
+      exporters: [logging]
+`
+	// // prepare
+	config, err := ConfigFromString(configStr)
+	require.NoError(t, err)
+	require.NotEmpty(t, config)
+
+	// test
+	check, err := GetEnabledReceivers(logger, config)
+	assert.NoError(t, err)
+	require.NotEmpty(t, check)
+}
+
+func TestEmptyEnabledReceivers(t *testing.T) {
+	// prepare
+
+	// First Test - Exporters
+	configStr := `
+receivers:
+  httpd/mtls:
+    protocols:
+      http:
+        endpoint: mysite.local:55690
+  jaeger:
+    protocols:
+      grpc:
+  prometheus:
+    protocols:
+      grpc:
+
+processors:
+
+exporters:
+  logging:
+
+service:
+  pipelines:
+    metrics:
+      receivers: []
+      exporters: []
+    metrics/1:
+      receivers: []
+      exporters: []
+`
+	// // prepare
+	config, err := ConfigFromString(configStr)
+	require.NoError(t, err)
+	require.NotEmpty(t, config)
+
+	// test
+	check, err := GetEnabledReceivers(logger, config)
+	assert.NoError(t, err)
+	require.Empty(t, check)
+	//require.NotEmpty(t, check)
+}

pkg/collector/reconcile/configmap_test.go

@@ -62,7 +62,7 @@ exporters:
 service:
   pipelines:
     metrics:
-      receivers: [prometheus]
+      receivers: [prometheus, jaeger]
       processors: []
       exporters: [logging]`,
 		}
@@ -112,6 +112,7 @@ service:
       processors: []
       receivers:
       - prometheus
+      - jaeger
 `,
 		}
 

pkg/collector/testdata/test.yaml

@@ -17,6 +17,6 @@ exporters:
 service:
   pipelines:
     metrics:
-      receivers: [prometheus]
+      receivers: [prometheus, jaeger]
       processors: []
       exporters: [logging]