Introduce simplified parsers

internal/components/component.go

@@ -0,0 +1,124 @@
+// 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 (
+	"encoding/json"
+	"errors"
+	"regexp"
+	"strconv"
+	"strings"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+var (
+	PortNotFoundErr = errors.New("port should not be empty")
+)
+
+type PortRetriever interface {
+	GetPortNum() (int32, error)
+	GetPortNumOrDefault(logr.Logger, int32) int32
+}
+
+type PortBuilderOption func(portBuilder *corev1.ServicePort)
+
+func WithTargetPort(targetPort int32) PortBuilderOption {
+	return func(servicePort *corev1.ServicePort) {
+		servicePort.TargetPort = intstr.FromInt32(targetPort)
+	}
+}
+
+func WithAppProtocol(proto *string) PortBuilderOption {
+	return func(servicePort *corev1.ServicePort) {
+		servicePort.AppProtocol = proto
+	}
+}
+
+func WithProtocol(proto corev1.Protocol) PortBuilderOption {
+	return func(servicePort *corev1.ServicePort) {
+		servicePort.Protocol = proto
+	}
+}
+
+func ComponentType(name string) string {
+	// components have a name like:
+	// - mycomponent/custom
+	// - mycomponent
+	// we extract the "mycomponent" part and see if we have a parser for the component
+	if strings.Contains(name, "/") {
+		return name[:strings.Index(name, "/")]
+	}
+
+	return name
+}
+
+func PortFromEndpoint(endpoint string) (int32, error) {
+	var err error
+	var port int64
+
+	r := regexp.MustCompile(":[0-9]+")
+
+	if r.MatchString(endpoint) {
+		port, err = strconv.ParseInt(strings.Replace(r.FindString(endpoint), ":", "", -1), 10, 32)
+
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	if port == 0 {
+		return 0, PortNotFoundErr
+	}
+
+	return int32(port), err
+}
+
+type ComponentPortParser interface {
+	// Ports returns the service ports parsed based on the exporter's configuration
+	Ports(logger logr.Logger, config interface{}) ([]corev1.ServicePort, error)
+
+	// ParserType returns the name of this parser
+	ParserType() string
+
+	// ParserName is an internal name for the parser
+	ParserName() string
+}
+
+func LoadMap[T any](m interface{}, in T) error {
+	// Convert map to JSON bytes
+	yamlData, err := json.Marshal(m)
+	if err != nil {
+		return err
+	}
+	// Unmarshal YAML into the provided struct
+	if err := json.Unmarshal(yamlData, in); err != nil {
+		return err
+	}
+	return nil
+}
+
+func ConstructServicePort(current *corev1.ServicePort, port int32) corev1.ServicePort {
+	return corev1.ServicePort{
+		Name:        current.Name,
+		Port:        port,
+		TargetPort:  current.TargetPort,
+		NodePort:    current.NodePort,
+		AppProtocol: current.AppProtocol,
+		Protocol:    current.Protocol,
+	}
+}

internal/components/component_test.go

@@ -0,0 +1,67 @@
+// 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_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/open-telemetry/opentelemetry-operator/internal/components"
+)
+
+func TestComponentType(t *testing.T) {
+	for _, tt := range []struct {
+		desc     string
+		name     string
+		expected string
+	}{
+		{"regular case", "myreceiver", "myreceiver"},
+		{"named instance", "myreceiver/custom", "myreceiver"},
+	} {
+		t.Run(tt.desc, func(t *testing.T) {
+			// test and verify
+			assert.Equal(t, tt.expected, components.ComponentType(tt.name))
+		})
+	}
+}
+
+func TestReceiverParsePortFromEndpoint(t *testing.T) {
+	for _, tt := range []struct {
+		desc          string
+		endpoint      string
+		expected      int
+		errorExpected bool
+	}{
+		{"regular case", "http://localhost:1234", 1234, false},
+		{"absolute with path", "http://localhost:1234/server-status?auto", 1234, false},
+		{"no protocol", "0.0.0.0:1234", 1234, false},
+		{"just port", ":1234", 1234, false},
+		{"no port at all", "http://localhost", 0, true},
+		{"overflow", "0.0.0.0:2147483648", 0, true},
+	} {
+		t.Run(tt.desc, func(t *testing.T) {
+			// test
+			val, err := components.PortFromEndpoint(tt.endpoint)
+			if tt.errorExpected {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			assert.EqualValues(t, tt.expected, val, "wrong port from endpoint %s: %d", tt.endpoint, val)
+		})
+	}
+}

internal/components/receivers/helpers.go

@@ -0,0 +1,46 @@
+// 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 receivers
+
+import "github.com/open-telemetry/opentelemetry-operator/internal/components"
+
+// registry holds a record of all known receiver parsers.
+var registry = make(map[string]components.ComponentPortParser)
+
+// Register adds a new parser builder to the list of known builders.
+func Register(name string, p components.ComponentPortParser) {
+	registry[name] = p
+}
+
+// IsRegistered checks whether a parser is registered with the given name.
+func IsRegistered(name string) bool {
+	_, ok := registry[name]
+	return ok
+}
+
+// BuilderFor returns a parser builder for the given exporter name.
+func BuilderFor(name string) components.ComponentPortParser {
+	if parser, ok := registry[components.ComponentType(name)]; ok {
+		return parser
+	}
+	return NewSinglePortParser(components.ComponentType(name), unsetPort)
+}
+
+func init() {
+	parsers := append(scraperReceivers, append(singleEndpointConfigs, multiPortReceivers...)...)
+	for _, parser := range parsers {
+		Register(parser.ParserType(), parser)
+	}
+}

internal/components/receivers/multi_endpoint.go

@@ -0,0 +1,148 @@
+// 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 receivers
+
+import (
+	"fmt"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+
+	"github.com/open-telemetry/opentelemetry-operator/internal/components"
+	"github.com/open-telemetry/opentelemetry-operator/internal/naming"
+)
+
+var (
+	_                  components.ComponentPortParser = &MultiPortReceiver{}
+	multiPortReceivers                                = []components.ComponentPortParser{
+		NewMultiPortReceiver("otlp",
+			WithPortMapping(
+				"grpc",
+				4317,
+				components.WithAppProtocol(&grpc),
+				components.WithTargetPort(4317),
+			), WithPortMapping(
+				"http",
+				4318,
+				components.WithAppProtocol(&http),
+				components.WithTargetPort(4318),
+			),
+		),
+		NewMultiPortReceiver("skywalking",
+			WithPortMapping(grpc, 11800,
+				components.WithTargetPort(11800),
+				components.WithAppProtocol(&grpc),
+			),
+			WithPortMapping(http, 12800,
+				components.WithTargetPort(12800),
+				components.WithAppProtocol(&http),
+			)),
+		NewMultiPortReceiver("jaeger",
+			WithPortMapping(grpc, 14250,
+				components.WithProtocol(corev1.ProtocolTCP),
+				components.WithAppProtocol(&grpc),
+			),
+			WithPortMapping("thrift_http", 14268,
+				components.WithProtocol(corev1.ProtocolTCP),
+				components.WithAppProtocol(&http),
+			),
+			WithPortMapping("thrift_compact", 6831,
+				components.WithProtocol(corev1.ProtocolUDP),
+			),
+			WithPortMapping("thrift_binary", 6832,
+				components.WithProtocol(corev1.ProtocolUDP),
+			),
+		),
+		NewMultiPortReceiver("loki",
+			WithPortMapping(grpc, 9095,
+				components.WithTargetPort(9095),
+				components.WithAppProtocol(&grpc),
+			),
+			WithPortMapping(http, 3100,
+				components.WithTargetPort(3100),
+				components.WithAppProtocol(&http),
+			),
+		),
+	}
+)
+
+// MultiProtocolEndpointConfig represents the minimal struct for a given YAML configuration input containing a map to
+// a struct with either endpoint or listen_address.
+type MultiProtocolEndpointConfig struct {
+	Protocols map[string]*SingleEndpointConfig `json:"protocols"`
+}
+
+// MultiPortOption allows the setting of options for a MultiPortReceiver.
+type MultiPortOption func(parser *MultiPortReceiver)
+
+// MultiPortReceiver is a special parser for components with endpoints for each protocol.
+type MultiPortReceiver struct {
+	name string
+
+	portMappings map[string]*corev1.ServicePort
+}
+
+func (m *MultiPortReceiver) Ports(logger logr.Logger, config interface{}) ([]corev1.ServicePort, error) {
+	multiProtoEndpointCfg := &MultiProtocolEndpointConfig{}
+	if err := components.LoadMap[*MultiProtocolEndpointConfig](config, multiProtoEndpointCfg); err != nil {
+		return nil, err
+	}
+	var ports []corev1.ServicePort
+	for protocol, ec := range multiProtoEndpointCfg.Protocols {
+		if defaultSvc, ok := m.portMappings[protocol]; ok {
+			port := defaultSvc.Port
+			if ec != nil {
+				port = ec.GetPortNumOrDefault(logger, port)
+				defaultSvc.Name = naming.PortName(fmt.Sprintf("%s-%s", m.name, protocol), port)
+			}
+			ports = append(ports, components.ConstructServicePort(defaultSvc, port))
+		} else {
+			return nil, fmt.Errorf("unknown protocol set: %s", protocol)
+		}
+	}
+	return ports, nil
+}
+
+func (m *MultiPortReceiver) ParserType() string {
+	return components.ComponentType(m.name)
+}
+
+func (m *MultiPortReceiver) ParserName() string {
+	return fmt.Sprintf("__%s", m.name)
+}
+
+func NewMultiPortReceiver(name string, opts ...MultiPortOption) *MultiPortReceiver {
+	multiReceiver := &MultiPortReceiver{
+		name:         name,
+		portMappings: map[string]*corev1.ServicePort{},
+	}
+	for _, opt := range opts {
+		opt(multiReceiver)
+	}
+	return multiReceiver
+}
+
+func WithPortMapping(name string, port int32, opts ...components.PortBuilderOption) MultiPortOption {
+	return func(parser *MultiPortReceiver) {
+		servicePort := &corev1.ServicePort{
+			Name: naming.PortName(fmt.Sprintf("%s-%s", parser.name, name), port),
+			Port: port,
+		}
+		for _, opt := range opts {
+			opt(servicePort)
+		}
+		parser.portMappings[name] = servicePort
+	}
+}