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 := ConfigValidate(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_to_probe.go

@@ -23,7 +23,7 @@ import (
 )
 
 var (
-	errNoService    = errors.New("no service available as part of the configuration")
+	ErrNoService    = errors.New("no service available as part of the configuration")
 	errNoExtensions = errors.New("no extensions available as part of the configuration")
 
 	errServiceNotAMap    = errors.New("service property in the configuration doesn't contain valid services")
@@ -51,7 +51,7 @@ const (
 func ConfigToContainerProbe(config map[interface{}]interface{}) (*corev1.Probe, error) {
 	serviceProperty, ok := config["service"]
 	if !ok {
-		return nil, errNoService
+		return nil, ErrNoService
 	}
 	service, ok := serviceProperty.(map[interface{}]interface{})
 	if !ok {

pkg/collector/adapters/config_to_probe_test.go

@@ -170,7 +170,7 @@ service: [hi]`,
 			desc: "NoService",
 			config: `extensions:
   health_check:`,
-			expectedErr: errNoService,
+			expectedErr: ErrNoService,
 		},
 	}
 

pkg/collector/adapters/config_validate.go

@@ -0,0 +1,86 @@
+// 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 ConfigValidate(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 {
+		//Getting all receivers present in the receivers section and setting them to false.
+		availableReceivers[recvID.(string)] = 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 {
+		//Getting all the available pipelines.
+		availablePipelines[pipID.(string)] = true
+	}
+
+	if len(pipeline) > 0 {
+		for pipelineID, pipelineCfg := range pipeline {
+			//Condition will get information if there are multiple configured pipelines.
+			if len(pipelineID.(string)) > 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...)
+						}
+						// All enabled receivers will be set as true
+						for _, recKey := range receiversList {
+							availableReceivers[recKey.(string)] = true
+						}
+					}
+				}
+			}
+		}
+	}
+	return availableReceivers, nil
+}

pkg/collector/adapters/config_validate_test.go

@@ -0,0 +1,68 @@
+// 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 := ConfigValidate(logger, config)
+	assert.NoError(t, err)
+	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]