Automate creation of Otel collector Service

pkg/api/v1beta3/dynakube/telemetryservice/props.go

@@ -1,5 +1,7 @@
 package telemetryservice
 
+import "github.com/Dynatrace/dynatrace-operator/pkg/consts"
+
 type Protocol string
 
 const (
@@ -35,6 +37,10 @@ func (spec *Spec) GetProtocols() []Protocol {
 	return protocols
 }
 
+func (ts *TelemetryService) DefaultName(dynakubeName string) string {
+	return dynakubeName + consts.TelemetryControllerSuffix
+}
+
 func (ts *TelemetryService) IsEnabled() bool {
 	return ts.Spec != nil
 }

pkg/api/validation/dynakube/telemetryservice.go

@@ -2,17 +2,22 @@ package validation
 
 import (
 	"context"
+	"fmt"
 	"slices"
 	"strings"
 
 	"github.com/Dynatrace/dynatrace-operator/pkg/api/v1beta3/dynakube"
 	"github.com/Dynatrace/dynatrace-operator/pkg/api/v1beta3/dynakube/telemetryservice"
+	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 const (
 	errorTelemetryServiceNotEnoughProtocols  = `DynaKube's specification enables the TelemetryService feature, at least one Protocol has to be specified.`
 	errorTelemetryServiceUnknownProtocols    = `DynaKube's specification enables the TelemetryService feature, unsupported protocols found on the Protocols list.`
 	errorTelemetryServiceDuplicatedProtocols = `DynaKube's specification enables the TelemetryService feature, duplicated protocols found on the Protocols list.`
+	errorTelemetryServiceNoDNS1053Label      = `DynaKube's specification enables the TelemetryService feature, the telemetry service name violates DNS-1035.
+    [The length limit for the name is %d. Additionally a DNS-1035 name must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character (e.g. 'my-name',  or 'abc-123', regex used for validation is '[a-z]([-a-z0-9]*[a-z0-9])?')]
+	`
 )
 
 func emptyTelemetryServiceProtocolsList(_ context.Context, _ *Validator, dk *dynakube.DynaKube) string {
@@ -82,3 +87,25 @@ func duplicatedTelemetryServiceProtocols(_ context.Context, _ *Validator, dk *dy
 
 	return ""
 }
+
+func invalidTelemetryServiceName(_ context.Context, _ *Validator, dk *dynakube.DynaKube) string {
+	if !dk.TelemetryService().IsEnabled() {
+		return ""
+	}
+
+	var errs []string
+
+	if dk.Spec.TelemetryService.ServiceName != "" {
+		errs = validation.IsDNS1035Label(dk.Spec.TelemetryService.ServiceName)
+	}
+
+	if len(errs) == 0 {
+		return ""
+	}
+
+	return invalidTelemetryServiceNameErrorMessage()
+}
+
+func invalidTelemetryServiceNameErrorMessage() string {
+	return fmt.Sprintf(errorTelemetryServiceNoDNS1053Label, validation.DNS1035LabelMaxLength)
+}

pkg/api/validation/dynakube/telemetryservice_test.go

@@ -129,4 +129,32 @@ func TestTelemetryServiceProtocols(t *testing.T) {
 				},
 			})
 	})
+
+	t.Run(`service name too long`, func(t *testing.T) {
+		assertDenied(t,
+			[]string{invalidTelemetryServiceNameErrorMessage()},
+			&dynakube.DynaKube{
+				ObjectMeta: defaultDynakubeObjectMeta,
+				Spec: dynakube.DynaKubeSpec{
+					APIURL: testApiUrl,
+					TelemetryService: &telemetryservice.Spec{
+						ServiceName: "a123456789012345678901234567890123456789012345678901234567890123",
+					},
+				},
+			})
+	})
+
+	t.Run(`service name violates DNS-1035`, func(t *testing.T) {
+		assertDenied(t,
+			[]string{invalidTelemetryServiceNameErrorMessage()},
+			&dynakube.DynaKube{
+				ObjectMeta: defaultDynakubeObjectMeta,
+				Spec: dynakube.DynaKubeSpec{
+					APIURL: testApiUrl,
+					TelemetryService: &telemetryservice.Spec{
+						ServiceName: "0123",
+					},
+				},
+			})
+	})
 }

pkg/api/validation/dynakube/validation.go

@@ -58,6 +58,7 @@ var (
 		emptyTelemetryServiceProtocolsList,
 		unknownTelemetryServiceProtocols,
 		duplicatedTelemetryServiceProtocols,
+		invalidTelemetryServiceName,
 		extensionsWithoutK8SMonitoring,
 	}
 	validatorWarningFuncs = []validatorFunc{

pkg/consts/otelc.go

@@ -4,4 +4,6 @@ const (
 	OtelcTokenSecretKey         = "otelc.token"
 	OtelcTokenSecretValuePrefix = "dt0x01"
 	OtelCollectorComPort        = 14599
+
+	TelemetryControllerSuffix = "-telemetry"
 )

pkg/controllers/dynakube/extension/service.go

@@ -66,17 +66,19 @@ func (r *reconciler) buildService() (*corev1.Service, error) {
 	// TODO: add proper version later on
 	appLabels := labels.NewAppLabels(labels.ExtensionComponentLabel, r.dk.Name, labels.ExtensionComponentLabel, "")
 
-	svcPort := corev1.ServicePort{
-		Name:       r.dk.ExtensionsPortName(),
-		Port:       consts.OtelCollectorComPort,
-		Protocol:   corev1.ProtocolTCP,
-		TargetPort: intstr.IntOrString{Type: intstr.String, StrVal: consts.ExtensionsCollectorTargetPortName},
+	svcPorts := []corev1.ServicePort{
+		{
+			Name:       r.dk.ExtensionsPortName(),
+			Port:       consts.OtelCollectorComPort,
+			Protocol:   corev1.ProtocolTCP,
+			TargetPort: intstr.IntOrString{Type: intstr.String, StrVal: consts.ExtensionsCollectorTargetPortName},
+		},
 	}
 
 	return service.Build(r.dk,
 		r.dk.ExtensionsServiceName(),
 		appLabels.BuildMatchLabels(),
-		svcPort,
+		svcPorts,
 		service.SetLabels(coreLabels.BuildLabels()),
 		service.SetType(corev1.ServiceTypeClusterIP),
 	)

pkg/controllers/dynakube/otelc/reconciler.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/Dynatrace/dynatrace-operator/pkg/api/v1beta3/dynakube"
 	"github.com/Dynatrace/dynatrace-operator/pkg/controllers"
+	"github.com/Dynatrace/dynatrace-operator/pkg/controllers/dynakube/otelc/service"
 	"github.com/Dynatrace/dynatrace-operator/pkg/controllers/dynakube/otelc/statefulset"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -14,6 +15,7 @@ type Reconciler struct {
 	apiReader             client.Reader
 	dk                    *dynakube.DynaKube
 	statefulsetReconciler controllers.Reconciler
+	serviceReconciler     *service.Reconciler
 }
 
 type ReconcilerBuilder func(client client.Client, apiReader client.Reader, dk *dynakube.DynaKube) controllers.Reconciler
@@ -24,11 +26,17 @@ func NewReconciler(client client.Client, apiReader client.Reader, dk *dynakube.D
 		apiReader:             apiReader,
 		dk:                    dk,
 		statefulsetReconciler: statefulset.NewReconciler(client, apiReader, dk),
+		serviceReconciler:     service.NewReconciler(client, apiReader, dk),
 	}
 }
 
 func (r *Reconciler) Reconcile(ctx context.Context) error {
-	err := r.statefulsetReconciler.Reconcile(ctx)
+	err := r.serviceReconciler.Reconcile(ctx)
+	if err != nil {
+		return err
+	}
+
+	err = r.statefulsetReconciler.Reconcile(ctx)
 	if err != nil {
 		log.Info("failed to reconcile Dynatrace OTELc statefulset")
 

pkg/controllers/dynakube/otelc/service/conditions.go

@@ -0,0 +1,5 @@
+package service
+
+const (
+	serviceConditionType = "OTELCService"
+)

pkg/controllers/dynakube/otelc/service/config.go

@@ -0,0 +1,7 @@
+package service
+
+import "github.com/Dynatrace/dynatrace-operator/pkg/logd"
+
+var (
+	log = logd.Get().WithName("otelc-service")
+)

pkg/controllers/dynakube/otelc/service/reconciler.go

@@ -0,0 +1,188 @@
+package service
+
+import (
+	"github.com/Dynatrace/dynatrace-operator/pkg/api/v1beta3/dynakube"
+	"github.com/Dynatrace/dynatrace-operator/pkg/api/v1beta3/dynakube/telemetryservice"
+	"github.com/Dynatrace/dynatrace-operator/pkg/util/conditions"
+	"github.com/Dynatrace/dynatrace-operator/pkg/util/kubeobjects/labels"
+	"github.com/Dynatrace/dynatrace-operator/pkg/util/kubeobjects/service"
+	"golang.org/x/net/context"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+const (
+	portNameZipkin              = "zipkin"
+	portNameOtlpGrpc            = "otlp-grpc"
+	portNameOtlpHttp            = "otlp-http"
+	portNameJaegerGrpc          = "jaeger-grpc"
+	portNameJaegerThriftBinary  = "jaeger-thrift-binary"
+	portNameJaegerThriftCompact = "jaeger-thrift-compact"
+	portNameJaegerThriftHttp    = "jaeger-thrift-http"
+	portNameStatsd              = "statsd"
+)
+
+type Reconciler struct {
+	client    client.Client
+	apiReader client.Reader
+	dk        *dynakube.DynaKube
+}
+
+type ReconcilerBuilder func(client client.Client, apiReader client.Reader, dk *dynakube.DynaKube) *Reconciler
+
+func NewReconciler(client client.Client, apiReader client.Reader, dk *dynakube.DynaKube) *Reconciler {
+	return &Reconciler{
+		client:    client,
+		dk:        dk,
+		apiReader: apiReader,
+	}
+}
+
+func (r *Reconciler) Reconcile(ctx context.Context) error {
+	if !r.dk.TelemetryService().IsEnabled() {
+		return r.removeServiceOnce(ctx)
+	}
+
+	if r.dk.Spec.TelemetryService.ServiceName != "" {
+		return r.removeServiceOnce(ctx)
+	}
+
+	return r.createOrUpdateService(ctx)
+}
+
+func (r *Reconciler) removeServiceOnce(ctx context.Context) error {
+	if meta.FindStatusCondition(*r.dk.Conditions(), serviceConditionType) == nil {
+		return nil
+	}
+	defer meta.RemoveStatusCondition(r.dk.Conditions(), serviceConditionType)
+
+	svc, err := r.buildService()
+	if err != nil {
+		log.Error(err, "could not build service during cleanup")
+
+		return err
+	}
+
+	err = service.Query(r.client, r.apiReader, log).Delete(ctx, svc)
+	if err != nil {
+		log.Error(err, "failed to clean up extension service")
+
+		return nil
+	}
+
+	return nil
+}
+
+func (r *Reconciler) createOrUpdateService(ctx context.Context) error {
+	newService, err := r.buildService()
+	if err != nil {
+		conditions.SetServiceGenFailed(r.dk.Conditions(), serviceConditionType, err)
+
+		return err
+	}
+
+	_, err = service.Query(r.client, r.apiReader, log).CreateOrUpdate(ctx, newService)
+	if err != nil {
+		log.Info("failed to create/update otelc service")
+		conditions.SetKubeApiError(r.dk.Conditions(), serviceConditionType, err)
+
+		return err
+	}
+
+	conditions.SetServiceCreated(r.dk.Conditions(), serviceConditionType, r.dk.TelemetryService().DefaultName(r.dk.Name))
+
+	return nil
+}
+
+func (r *Reconciler) buildService() (*corev1.Service, error) {
+	coreLabels := labels.NewCoreLabels(r.dk.Name, labels.OtelCComponentLabel)
+	// TODO: add proper version later on
+	appLabels := labels.NewAppLabels(labels.OtelCComponentLabel, r.dk.Name, labels.OtelCComponentLabel, "")
+
+	var svcPorts []corev1.ServicePort
+	if r.dk.TelemetryService().IsEnabled() && r.dk.Spec.TelemetryService.ServiceName == "" {
+		svcPorts = buildServicePortList(r.dk.TelemetryService().GetProtocols())
+	}
+
+	return service.Build(r.dk,
+		r.dk.TelemetryService().DefaultName(r.dk.Name),
+		appLabels.BuildMatchLabels(),
+		svcPorts,
+		service.SetLabels(coreLabels.BuildLabels()),
+		service.SetType(corev1.ServiceTypeClusterIP),
+	)
+}
+
+func buildServicePortList(protocols []telemetryservice.Protocol) []corev1.ServicePort {
+	if len(protocols) == 0 {
+		return nil
+	}
+
+	svcPorts := make([]corev1.ServicePort, 0)
+
+	for _, protocol := range protocols {
+		switch protocol {
+		case telemetryservice.ZipkinProtocol:
+			svcPorts = append(svcPorts, corev1.ServicePort{
+				Name:       portNameZipkin,
+				Port:       9411,
+				Protocol:   corev1.ProtocolTCP,
+				TargetPort: intstr.FromInt32(9411),
+			})
+		case telemetryservice.OtlpProtocol:
+			svcPorts = append(svcPorts,
+				corev1.ServicePort{
+					Name:       portNameOtlpGrpc,
+					Port:       4317,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(4317),
+				},
+				corev1.ServicePort{
+					Name:       portNameOtlpHttp,
+					Port:       4318,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(4318),
+				})
+		case telemetryservice.JaegerProtocol:
+			svcPorts = append(svcPorts,
+				corev1.ServicePort{
+					Name:       portNameJaegerGrpc,
+					Port:       14250,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(14250),
+				},
+				corev1.ServicePort{
+					Name:       portNameJaegerThriftBinary,
+					Port:       6832,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(6832),
+				},
+				corev1.ServicePort{
+					Name:       portNameJaegerThriftCompact,
+					Port:       6831,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(6831),
+				},
+				corev1.ServicePort{
+					Name:       portNameJaegerThriftHttp,
+					Port:       14268,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(14268),
+				})
+		case telemetryservice.StatsdProtocol:
+			svcPorts = append(svcPorts,
+				corev1.ServicePort{
+					Name:       portNameStatsd,
+					Port:       8125,
+					Protocol:   corev1.ProtocolTCP,
+					TargetPort: intstr.FromInt32(8125),
+				})
+		default:
+			log.Info("unknown telemetry service protocol ignored", "protocol", protocol)
+		}
+	}
+
+	return svcPorts
+}