Added protocol handling for ServicePort and fixed type/value naming inside generic receiver

.chloggen/fix-protocol-handling-in-serviceports.yaml

@@ -0,0 +1,18 @@
+# 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. operator, target allocator, github action)
+component: operator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fixed handling of protocol in exposed ports."
+
+# One or more tracking issues related to the change
+issues: [2619]
+
+# (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: |
+  Make distinction not only on the port number, but also on protocol. This fix allows to have multiple exposed
+  ServicePorts with the same port number, but different protocols.

.chloggen/fix-syslog-tcplog-udplog.yaml

@@ -0,0 +1,18 @@
+# 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. operator, target allocator, github action)
+component: operator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Fixed handling of exposed port protocol in syslog, tcplog and udplog receivers."
+
+# One or more tracking issues related to the change
+issues: [767,2619]
+
+# (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: |
+  Please note that the operator currently exposes just one port (tcp or udp) of syslog receiver due to the current
+  receiver implementation (patches are welcome).

internal/manifests/collector/parser/receiver/receiver.go

@@ -104,30 +104,12 @@ func isScraperReceiver(name string) bool {
 
 func singlePortFromConfigEndpoint(logger logr.Logger, name string, config map[interface{}]interface{}) *v1.ServicePort {
 	var endpoint interface{}
+	var receiverType = receiverType(name)
 	switch {
-	// syslog receiver contains the endpoint
-	// that needs to be exposed one level down inside config
-	// i.e. either in tcp or udp section with field key
-	// as `listen_address`
-	case name == "syslog":
-		var c map[interface{}]interface{}
-		if udp, isUDP := config["udp"]; isUDP && udp != nil {
-			c = udp.(map[interface{}]interface{})
-			endpoint = getAddressFromConfig(logger, name, listenAddressKey, c)
-		} else if tcp, isTCP := config["tcp"]; isTCP && tcp != nil {
-			c = tcp.(map[interface{}]interface{})
-			endpoint = getAddressFromConfig(logger, name, listenAddressKey, c)
-		}
-
-	// tcplog and udplog receivers hold the endpoint
-	// value in `listen_address` field
-	case name == "tcplog" || name == "udplog":
-		endpoint = getAddressFromConfig(logger, name, listenAddressKey, config)
-
 	// ignore the receiver as it holds the field key endpoint, and it
 	// is a scraper, we only expose endpoint through k8s service objects for
 	// receivers that aren't scrapers.
-	case isScraperReceiver(name):
+	case isScraperReceiver(receiverType):
 		return nil
 
 	default:

internal/manifests/collector/parser/receiver/receiver_syslog.go

@@ -0,0 +1,98 @@
+// 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 receiver
+
+import (
+	"fmt"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser"
+	"github.com/open-telemetry/opentelemetry-operator/internal/naming"
+)
+
+var _ parser.ComponentPortParser = &SyslogReceiverParser{}
+
+const parserNameSyslog = "__syslog"
+
+// SyslogReceiverParser parses the configuration for TCP log receivers.
+type SyslogReceiverParser struct {
+	config map[interface{}]interface{}
+	logger logr.Logger
+	name   string
+}
+
+// NewSyslogReceiverParser builds a new parser for TCP log receivers.
+func NewSyslogReceiverParser(logger logr.Logger, name string, config map[interface{}]interface{}) parser.ComponentPortParser {
+	return &SyslogReceiverParser{
+		logger: logger,
+		name:   name,
+		config: config,
+	}
+}
+
+func (o *SyslogReceiverParser) Ports() ([]corev1.ServicePort, error) {
+	var endpoint interface{}
+	var endpointName string
+	var protocol corev1.Protocol
+	var c map[interface{}]interface{}
+
+	// syslog receiver contains the endpoint
+	// that needs to be exposed one level down inside config
+	// i.e. either in tcp or udp section with field key
+	// as `listen_address`
+	if tcp, isTCP := o.config["tcp"]; isTCP && tcp != nil {
+		c = tcp.(map[interface{}]interface{})
+		endpointName = "tcp"
+		endpoint = getAddressFromConfig(o.logger, o.name, listenAddressKey, c)
+		protocol = corev1.ProtocolTCP
+	} else if udp, isUDP := o.config["udp"]; isUDP && udp != nil {
+		c = udp.(map[interface{}]interface{})
+		endpointName = "udp"
+		endpoint = getAddressFromConfig(o.logger, o.name, listenAddressKey, c)
+		protocol = corev1.ProtocolUDP
+	}
+
+	switch e := endpoint.(type) {
+	case nil:
+		break
+	case string:
+		port, err := portFromEndpoint(e)
+		if err != nil {
+			o.logger.WithValues(listenAddressKey, e).Error(err, fmt.Sprintf("couldn't parse the %s endpoint's port", endpointName))
+			return nil, nil
+		}
+
+		return []corev1.ServicePort{{
+			Port:     port,
+			Name:     naming.PortName(o.name, port),
+			Protocol: protocol,
+		}}, nil
+	default:
+		o.logger.WithValues(listenAddressKey, endpoint).Error(fmt.Errorf("unrecognized type %T of %s endpoint", endpoint, endpointName),
+			"receiver's endpoint isn't a string")
+	}
+
+	return []corev1.ServicePort{}, nil
+}
+
+func (o *SyslogReceiverParser) ParserName() string {
+	return parserNameSyslog
+}
+
+func init() {
+	Register("syslog", NewSyslogReceiverParser)
+}

internal/manifests/collector/parser/receiver/receiver_syslog_test.go

@@ -0,0 +1,64 @@
+// 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 receiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestSyslogSelfRegisters(t *testing.T) {
+	// verify
+	assert.True(t, IsRegistered("syslog"))
+}
+
+func TestSyslogIsFoundByName(t *testing.T) {
+	// test
+	p, err := For(logger, "syslog", map[interface{}]interface{}{})
+	assert.NoError(t, err)
+
+	// verify
+	assert.Equal(t, "__syslog", p.ParserName())
+}
+
+func TestSyslogConfiguration(t *testing.T) {
+	for _, tt := range []struct {
+		desc     string
+		config   map[interface{}]interface{}
+		expected []corev1.ServicePort
+	}{
+		{"Empty configuration", map[interface{}]interface{}{}, []corev1.ServicePort{}},
+		{"UDP port configuration",
+			map[interface{}]interface{}{"udp": map[interface{}]interface{}{"listen_address": "0.0.0.0:1234"}},
+			[]corev1.ServicePort{{Name: "syslog", Port: 1234, Protocol: corev1.ProtocolUDP}}},
+		{"TCP port configuration",
+			map[interface{}]interface{}{"tcp": map[interface{}]interface{}{"listen_address": "0.0.0.0:1234"}},
+			[]corev1.ServicePort{{Name: "syslog", Port: 1234, Protocol: corev1.ProtocolTCP}}},
+	} {
+		t.Run(tt.desc, func(t *testing.T) {
+			// prepare
+			builder := NewSyslogReceiverParser(logger, "syslog", tt.config)
+
+			// test
+			ports, err := builder.Ports()
+
+			// verify
+			assert.NoError(t, err)
+			assert.Equal(t, ports, tt.expected)
+		})
+	}
+}

internal/manifests/collector/parser/receiver/receiver_tcplog.go

@@ -0,0 +1,79 @@
+// 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 receiver
+
+import (
+	"fmt"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/collector/parser"
+	"github.com/open-telemetry/opentelemetry-operator/internal/naming"
+)
+
+var _ parser.ComponentPortParser = &TcpLogReceiverParser{}
+
+const parserNameTcpLog = "__tcplog"
+
+// TcpLogReceiverParser parses the configuration for TCP log receivers.
+type TcpLogReceiverParser struct {
+	config map[interface{}]interface{}
+	logger logr.Logger
+	name   string
+}
+
+// NewTcpLogReceiverParser builds a new parser for TCP log receivers.
+func NewTcpLogReceiverParser(logger logr.Logger, name string, config map[interface{}]interface{}) parser.ComponentPortParser {
+	return &TcpLogReceiverParser{
+		logger: logger,
+		name:   name,
+		config: config,
+	}
+}
+
+func (o *TcpLogReceiverParser) Ports() ([]corev1.ServicePort, error) {
+	// tcplog receiver hold the endpoint value in `listen_address` field
+	var endpoint = getAddressFromConfig(o.logger, o.name, listenAddressKey, o.config)
+
+	switch e := endpoint.(type) {
+	case nil:
+		break
+	case string:
+		port, err := portFromEndpoint(e)
+		if err != nil {
+			o.logger.WithValues(listenAddressKey, e).Error(err, "couldn't parse the endpoint's port")
+			return nil, nil
+		}
+
+		return []corev1.ServicePort{{
+			Port:     port,
+			Name:     naming.PortName(o.name, port),
+			Protocol: corev1.ProtocolTCP,
+		}}, nil
+	default:
+		o.logger.WithValues(listenAddressKey, endpoint).Error(fmt.Errorf("unrecognized type %T", endpoint), "receiver's endpoint isn't a string")
+	}
+
+	return []corev1.ServicePort{}, nil
+}
+
+func (o *TcpLogReceiverParser) ParserName() string {
+	return parserNameTcpLog
+}
+
+func init() {
+	Register("tcplog", NewTcpLogReceiverParser)
+}

internal/manifests/collector/parser/receiver/receiver_tcplog_test.go

@@ -0,0 +1,61 @@
+// 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 receiver
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestTcpLogSelfRegisters(t *testing.T) {
+	// verify
+	assert.True(t, IsRegistered("tcplog"))
+}
+
+func TestTcpLogIsFoundByName(t *testing.T) {
+	// test
+	p, err := For(logger, "tcplog", map[interface{}]interface{}{})
+	assert.NoError(t, err)
+
+	// verify
+	assert.Equal(t, "__tcplog", p.ParserName())
+}
+
+func TestTcpLogConfiguration(t *testing.T) {
+	for _, tt := range []struct {
+		desc     string
+		config   map[interface{}]interface{}
+		expected []corev1.ServicePort
+	}{
+		{"Empty configuration", map[interface{}]interface{}{}, []corev1.ServicePort{}},
+		{"TCP port configuration",
+			map[interface{}]interface{}{"listen_address": "0.0.0.0:1234"},
+			[]corev1.ServicePort{{Name: "tcplog", Port: 1234, Protocol: corev1.ProtocolTCP}}},
+	} {
+		t.Run(tt.desc, func(t *testing.T) {
+			// prepare
+			builder := NewTcpLogReceiverParser(logger, "tcplog", tt.config)
+
+			// test
+			ports, err := builder.Ports()
+
+			// verify
+			assert.NoError(t, err)
+			assert.Equal(t, ports, tt.expected)
+		})
+	}
+}