Add nop parser for exporters (open-telemetry#3125)

* add nop parser for exporter use

* chlog

* Fix exporter tests

* another one

Author: jaronoff97

.chloggen/fix-parsing-bug.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. collector, target allocator, auto-instrumentation, opamp, github action)
+component: collector
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fixes a bug where an exporter would cause a port collision
+
+# One or more tracking issues related to the change
+issues: [3124]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

apis/v1beta1/config.go

@@ -141,6 +141,7 @@ type Config struct {
 
 // getPortsForComponentKinds gets the ports for the given ComponentKind(s).
 func (c *Config) getPortsForComponentKinds(logger logr.Logger, componentKinds ...ComponentKind) ([]corev1.ServicePort, error) {
+
 	var ports []corev1.ServicePort
 	enabledComponents := c.GetEnabledComponents()
 	for _, componentKind := range componentKinds {

apis/v1beta1/config_test.go

@@ -497,25 +497,12 @@ func TestConfig_GetExporterPorts(t *testing.T) {
 					Name: "prometheus",
 					Port: 8889,
 				},
-				{
-					Name: "otlp",
-					Port: 4317,
-				},
-				{
-					Name: "zipkin",
-					Port: 9411,
-				},
 			},
 		},
 		{
 			name: "extensions",
 			file: "testdata/otelcol-extensions.yaml",
-			want: []v1.ServicePort{
-				{
-					Name: "otlp-auth",
-					Port: 4317,
-				},
-			},
+			want: nil,
 		},
 		{
 			name: "filelog",

controllers/reconcile_test.go

@@ -99,6 +99,7 @@ func TestOpenTelemetryCollectorReconciler_Reconcile(t *testing.T) {
 	deploymentExtraPorts.Annotations = map[string]string{
 		"new-annotation": "new-value",
 	}
+	baseOTLPParams := testCollectorAssertNoErr(t, "test-otlp", "", otlpTestFile)
 	ingressParams := testCollectorAssertNoErr(t, "test-ingress", "", testFileIngress)
 	ingressParams.Spec.Ingress.Type = "ingress"
 	updatedIngressParams := testCollectorAssertNoErr(t, "test-ingress", "", testFileIngress)
@@ -220,6 +221,41 @@ func TestOpenTelemetryCollectorReconciler_Reconcile(t *testing.T) {
 				},
 			},
 		},
+
+		{
+			name: "otlp receiver collector",
+			args: args{
+				params:  baseOTLPParams,
+				updates: []v1alpha1.OpenTelemetryCollector{},
+			},
+			want: []want{
+				{
+					result: controllerruntime.Result{},
+					checks: []check[v1alpha1.OpenTelemetryCollector]{
+						func(t *testing.T, params v1alpha1.OpenTelemetryCollector) {
+							d := appsv1.StatefulSet{}
+							exists, err := populateObjectIfExists(t, &d, namespacedObjectName(naming.Collector(params.Name), params.Namespace))
+							assert.NoError(t, err)
+							assert.True(t, exists)
+							assert.Equal(t, int32(1), *d.Spec.Replicas)
+							svc := &v1.Service{}
+							exists, err = populateObjectIfExists(t, svc, namespacedObjectName(naming.Service(params.Name), params.Namespace))
+							assert.NoError(t, err)
+							assert.True(t, exists)
+							assert.Equal(t, svc.Spec.Selector, map[string]string{
+								"app.kubernetes.io/component":  "opentelemetry-collector",
+								"app.kubernetes.io/instance":   "default.test-otlp",
+								"app.kubernetes.io/managed-by": "opentelemetry-operator",
+								"app.kubernetes.io/part-of":    "opentelemetry",
+							})
+							assert.Len(t, svc.Spec.Ports, 4)
+						},
+					},
+					wantErr:     assert.NoError,
+					validateErr: assert.NoError,
+				},
+			},
+		},
 		{
 			name: "invalid mode",
 			args: args{

controllers/suite_test.go

@@ -91,6 +91,7 @@ const (
 	promFile                 = "testdata/test.yaml"
 	updatedPromFile          = "testdata/test_ta_update.yaml"
 	testFileIngress          = "testdata/ingress_testdata.yaml"
+	otlpTestFile             = "testdata/otlp_test.yaml"
 )
 
 var _ autodetect.AutoDetect = (*mockAutoDetect)(nil)

controllers/testdata/otlp_test.yaml

@@ -0,0 +1,23 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+processors:
+exporters:
+  otlp:
+    endpoint: jaeger-allinone-collector-headless.chainsaw-otlp-metrics.svc:4317
+    tls:
+      insecure: true
+  prometheus:
+    endpoint: 0.0.0.0:8889
+    resource_to_telemetry_conversion:
+      enabled: true # by default resource attributes are dropped
+service:
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [otlp]
+    metrics:
+      receivers: [otlp]
+      exporters: [prometheus]

internal/components/exporters/helpers.go

@@ -38,7 +38,7 @@ func ParserFor(name string) components.ComponentPortParser {
 		return parser
 	}
 	// We want the default for exporters to fail silently.
-	return components.NewSilentSinglePortParser(components.ComponentType(name), components.UnsetPort)
+	return components.NewNopParser(components.ComponentType(name), components.UnsetPort)
 }
 
 var (

internal/components/exporters/helpers_test.go

@@ -34,8 +34,7 @@ func TestParserForReturns(t *testing.T) {
 		"endpoint": "localhost:9000",
 	})
 	assert.NoError(t, err)
-	assert.Len(t, ports, 1)
-	assert.Equal(t, ports[0].Port, int32(9000))
+	assert.Len(t, ports, 0) // Should use the nop parser
 }
 
 func TestCanRegister(t *testing.T) {

internal/components/multi_endpoint.go

@@ -53,8 +53,8 @@ func (m *MultiPortReceiver) Ports(logger logr.Logger, name string, config interf
 			port := defaultSvc.Port
 			if ec != nil {
 				port = ec.GetPortNumOrDefault(logger, port)
-				defaultSvc.Name = naming.PortName(fmt.Sprintf("%s-%s", name, protocol), port)
 			}
+			defaultSvc.Name = naming.PortName(fmt.Sprintf("%s-%s", name, protocol), port)
 			ports = append(ports, ConstructServicePort(defaultSvc, port))
 		} else {
 			return nil, fmt.Errorf("unknown protocol set: %s", protocol)

internal/components/nop_parser.go

@@ -0,0 +1,48 @@
+// 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 components
+
+import (
+	"fmt"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+)
+
+var (
+	_ ComponentPortParser = &NopParser{}
+)
+
+// SingleEndpointParser is a special parser for a generic receiver that has an endpoint or listen_address in its
+// configuration. It doesn't self-register and should be created/used directly.
+type NopParser struct {
+	name string
+}
+
+func (n *NopParser) Ports(logger logr.Logger, name string, config interface{}) ([]corev1.ServicePort, error) {
+	return nil, nil
+}
+
+func (n *NopParser) ParserType() string {
+	return ComponentType(n.name)
+}
+
+func (n *NopParser) ParserName() string {
+	return fmt.Sprintf("__%s", n.name)
+}
+
+func NewNopParser(name string, port int32, opts ...PortBuilderOption) *NopParser {
+	return &NopParser{name: name}
+}

internal/manifests/collector/service_test.go

@@ -226,6 +226,47 @@ func TestDesiredService(t *testing.T) {
 		assert.Equal(t, expected, *actual)
 	})
 
+	t.Run("should return service with OTLP ports", func(t *testing.T) {
+		params := manifests.Params{
+			Config: config.Config{},
+			Log:    logger,
+			OtelCol: v1beta1.OpenTelemetryCollector{
+				Spec: v1beta1.OpenTelemetryCollectorSpec{Config: v1beta1.Config{
+					Receivers: v1beta1.AnyConfig{
+						Object: map[string]interface{}{
+							"otlp": map[string]interface{}{
+								"protocols": map[string]interface{}{
+									"grpc": nil,
+									"http": nil,
+								},
+							},
+						},
+					},
+					Exporters: v1beta1.AnyConfig{
+						Object: map[string]interface{}{
+							"otlp": map[string]interface{}{
+								"endpoint": "jaeger-allinone-collector-headless.chainsaw-otlp-metrics.svc:4317",
+							},
+						},
+					},
+					Service: v1beta1.Service{
+						Pipelines: map[string]*v1beta1.Pipeline{
+							"traces": {
+								Receivers: []string{"otlp"},
+								Exporters: []string{"otlp"},
+							},
+						},
+					},
+				}},
+			},
+		}
+
+		actual, err := Service(params)
+		assert.NotNil(t, actual)
+		assert.Len(t, actual.Spec.Ports, 2)
+		assert.NoError(t, err)
+	})
+
 }
 
 func TestHeadlessService(t *testing.T) {