Added protocol handling for ServicePort and fixed type/value naming i…

…nside generic receiver (open-telemetry#2619)

* Fix handling of type

The name is not correctly split before it is checked. Split it first.

Signed-off-by: Oldřich Jedlička <[email protected]>

* Split parsing of Syslog, TCP Log and UDP Log parsers

The generic implementation does not allow to specify different endpoints
and endpoint-specific protocols, so split the implementation instead of
trying to make the generic implementation a complex catch-all-party.

Signed-off-by: Oldřich Jedlička <[email protected]>

* Handle port numbers together with protocols

It is legal to have the same protocol number and different protocol, so
handle it accordingly. Also interpret empty protocol as TCP, like the K8s
does.

Signed-off-by: Oldřich Jedlička <[email protected]>

* Changelog updates

Signed-off-by: Oldřich Jedlička <[email protected]>

---------

Signed-off-by: Oldřich Jedlička <[email protected]>

.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)
+		})
+	}
+}