Create service for extensions

.chloggen/service-extension.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# 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: support for creating a service for extensions when ports are specified.
+
+# One or more tracking issues related to the change
+issues: [3460]
+
+# (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

@@ -206,7 +206,12 @@ func (c *Config) getPortsForComponentKinds(logger logr.Logger, componentKinds ..
 		case KindProcessor:
 			continue
 		case KindExtension:
-			continue
+			retriever = extensions.ParserFor
+			if c.Extensions == nil {
+				cfg = AnyConfig{}
+			} else {
+				cfg = *c.Extensions
+			}
 		}
 		for componentName := range enabledComponents[componentKind] {
 			// TODO: Clean up the naming here and make it simpler to use a retriever.
@@ -318,10 +323,18 @@ func (c *Config) GetExporterPorts(logger logr.Logger) ([]corev1.ServicePort, err
 	return c.getPortsForComponentKinds(logger, KindExporter)
 }
 
-func (c *Config) GetAllPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
+func (c *Config) GetExtensionPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
+	return c.getPortsForComponentKinds(logger, KindExtension)
+}
+
+func (c *Config) GetReceiverAndExporterPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
 	return c.getPortsForComponentKinds(logger, KindReceiver, KindExporter)
 }
 
+func (c *Config) GetAllPorts(logger logr.Logger) ([]corev1.ServicePort, error) {
+	return c.getPortsForComponentKinds(logger, KindReceiver, KindExporter, KindExtension)
+}
+
 func (c *Config) GetEnvironmentVariables(logger logr.Logger) ([]corev1.EnvVar, error) {
 	return c.getEnvironmentVariablesForComponentKinds(logger, KindReceiver)
 }

internal/components/extensions/helpers.go

@@ -55,6 +55,9 @@ var (
 				return components.ParseSingleEndpointSilent(logger, name, defaultPort, &config.SingleEndpointConfig)
 			}).
 			MustBuild(),
+		components.NewSinglePortParserBuilder("jaeger_query", 16686).
+			WithTargetPort(16686).
+			MustBuild(),
 	}
 )
 

internal/manifests/collector/collector.go

@@ -53,6 +53,7 @@ func Build(params manifests.Params) ([]client.Object, error) {
 		manifests.Factory(Service),
 		manifests.Factory(HeadlessService),
 		manifests.Factory(MonitoringService),
+		manifests.Factory(ExtensionService),
 		manifests.Factory(Ingress),
 	}...)
 

internal/manifests/collector/service.go

@@ -42,10 +42,11 @@ const (
 	BaseServiceType ServiceType = iota
 	HeadlessServiceType
 	MonitoringServiceType
+	ExtensionServiceType
 )
 
 func (s ServiceType) String() string {
-	return [...]string{"base", "headless", "monitoring"}[s]
+	return [...]string{"base", "headless", "monitoring", "extension"}[s]
 }
 
 func HeadlessService(params manifests.Params) (*corev1.Service, error) {
@@ -108,6 +109,39 @@ func MonitoringService(params manifests.Params) (*corev1.Service, error) {
 	}, nil
 }
 
+func ExtensionService(params manifests.Params) (*corev1.Service, error) {
+	name := naming.ExtensionService(params.OtelCol.Name)
+	labels := manifestutils.Labels(params.OtelCol.ObjectMeta, name, params.OtelCol.Spec.Image, ComponentOpenTelemetryCollector, []string{})
+	labels[serviceTypeLabel] = ExtensionServiceType.String()
+
+	annotations, err := manifestutils.Annotations(params.OtelCol, params.Config.AnnotationsFilter())
+	if err != nil {
+		return nil, err
+	}
+
+	ports, err := params.OtelCol.Spec.Config.GetExtensionPorts(params.Log)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(ports) == 0 {
+		return nil, nil
+	}
+
+	return &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        name,
+			Namespace:   params.OtelCol.Namespace,
+			Labels:      labels,
+			Annotations: annotations,
+		},
+		Spec: corev1.ServiceSpec{
+			Ports:    ports,
+			Selector: manifestutils.SelectorLabels(params.OtelCol.ObjectMeta, ComponentOpenTelemetryCollector),
+		},
+	}, nil
+}
+
 func Service(params manifests.Params) (*corev1.Service, error) {
 	name := naming.Service(params.OtelCol.Name)
 	labels := manifestutils.Labels(params.OtelCol.ObjectMeta, name, params.OtelCol.Spec.Image, ComponentOpenTelemetryCollector, []string{})
@@ -118,7 +152,7 @@ func Service(params manifests.Params) (*corev1.Service, error) {
 		return nil, err
 	}
 
-	ports, err := params.OtelCol.Spec.Config.GetAllPorts(params.Log)
+	ports, err := params.OtelCol.Spec.Config.GetReceiverAndExporterPorts(params.Log)
 	if err != nil {
 		return nil, err
 	}

internal/manifests/collector/service_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/open-telemetry/opentelemetry-operator/internal/config"
 	"github.com/open-telemetry/opentelemetry-operator/internal/manifests"
 	"github.com/open-telemetry/opentelemetry-operator/internal/manifests/manifestutils"
+	"github.com/open-telemetry/opentelemetry-operator/internal/naming"
 )
 
 func TestExtractPortNumbersAndNames(t *testing.T) {
@@ -321,6 +322,206 @@ func TestMonitoringService(t *testing.T) {
 	})
 }
 
+func TestExtensionService(t *testing.T) {
+	testCases := []struct {
+		name          string
+		params        manifests.Params
+		expectedPorts []v1.ServicePort
+	}{
+		{
+			name: "when the extension has http endpoint",
+			params: manifests.Params{
+				Config: config.Config{},
+				Log:    logger,
+				OtelCol: v1beta1.OpenTelemetryCollector{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1beta1.OpenTelemetryCollectorSpec{
+						Config: v1beta1.Config{
+							Service: v1beta1.Service{
+								Extensions: []string{"jaeger_query"},
+							},
+							Extensions: &v1beta1.AnyConfig{
+								Object: map[string]interface{}{
+									"jaeger_query": map[string]interface{}{
+										"http": map[string]interface{}{
+											"endpoint": "0.0.0.0:16686",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedPorts: []v1.ServicePort{
+				{
+					Name: "jaeger-query",
+					Port: 16686,
+					TargetPort: intstr.IntOrString{
+						IntVal: 16686,
+					},
+				},
+			},
+		},
+		{
+			name: "when the extension has grpc endpoint",
+			params: manifests.Params{
+				Config: config.Config{},
+				Log:    logger,
+				OtelCol: v1beta1.OpenTelemetryCollector{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1beta1.OpenTelemetryCollectorSpec{
+						Config: v1beta1.Config{
+							Service: v1beta1.Service{
+								Extensions: []string{"jaeger_query"},
+							},
+							Extensions: &v1beta1.AnyConfig{
+								Object: map[string]interface{}{
+									"jaeger_query": map[string]interface{}{
+										"http": map[string]interface{}{
+											"endpoint": "0.0.0.0:16686",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedPorts: []v1.ServicePort{
+				{
+					Name: "jaeger-query",
+					Port: 16686,
+					TargetPort: intstr.IntOrString{
+						IntVal: 16686,
+					},
+				},
+			},
+		},
+		{
+			name: "when the extension has both http and grpc endpoint",
+			params: manifests.Params{
+				Config: config.Config{},
+				Log:    logger,
+				OtelCol: v1beta1.OpenTelemetryCollector{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1beta1.OpenTelemetryCollectorSpec{
+						Config: v1beta1.Config{
+							Service: v1beta1.Service{
+								Extensions: []string{"jaeger_query"},
+							},
+							Extensions: &v1beta1.AnyConfig{
+								Object: map[string]interface{}{
+									"jaeger_query": map[string]interface{}{
+										"http": map[string]interface{}{
+											"endpoint": "0.0.0.0:16686",
+										},
+										"grpc": map[string]interface{}{
+											"endpoint": "0.0.0.0:16686",
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedPorts: []v1.ServicePort{
+				{
+					Name: "jaeger-query",
+					Port: 16686,
+					TargetPort: intstr.IntOrString{
+						IntVal: 16686,
+					},
+				},
+			},
+		},
+		{
+			name: "when the extension has no extensions defined",
+			params: manifests.Params{
+				Config: config.Config{},
+				Log:    logger,
+				OtelCol: v1beta1.OpenTelemetryCollector{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1beta1.OpenTelemetryCollectorSpec{
+						Config: v1beta1.Config{
+							Service: v1beta1.Service{
+								Extensions: []string{"jaeger_query"},
+							},
+							Extensions: &v1beta1.AnyConfig{
+								Object: map[string]interface{}{},
+							},
+						},
+					},
+				},
+			},
+			expectedPorts: []v1.ServicePort{},
+		},
+		{
+			name: "when the extension has no endpoint defined",
+			params: manifests.Params{
+				Config: config.Config{},
+				Log:    logger,
+				OtelCol: v1beta1.OpenTelemetryCollector{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test",
+					},
+					Spec: v1beta1.OpenTelemetryCollectorSpec{
+						Config: v1beta1.Config{
+							Service: v1beta1.Service{
+								Extensions: []string{"jaeger_query"},
+							},
+							Extensions: &v1beta1.AnyConfig{
+								Object: map[string]interface{}{
+									"jaeger_query": map[string]interface{}{},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedPorts: []v1.ServicePort{
+				{
+					Name: "jaeger-query",
+					Port: 16686,
+					TargetPort: intstr.IntOrString{
+						IntVal: 16686,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			actual, err := ExtensionService(tc.params)
+			assert.NoError(t, err)
+
+			if len(tc.expectedPorts) > 0 {
+				assert.NotNil(t, actual)
+				assert.Equal(t, actual.Name, naming.ExtensionService(tc.params.OtelCol.Name))
+				// ports assertion
+				assert.Equal(t, len(tc.expectedPorts), len(actual.Spec.Ports))
+				assert.Equal(t, tc.expectedPorts[0].Name, actual.Spec.Ports[0].Name)
+				assert.Equal(t, tc.expectedPorts[0].Port, actual.Spec.Ports[0].Port)
+				assert.Equal(t, tc.expectedPorts[0].TargetPort.IntVal, actual.Spec.Ports[0].TargetPort.IntVal)
+			} else {
+				// no ports, no service
+				assert.Nil(t, actual)
+			}
+		})
+	}
+}
+
 func service(name string, ports []v1beta1.PortsSpec) v1.Service {
 	return serviceWithInternalTrafficPolicy(name, ports, v1.ServiceInternalTrafficPolicyCluster)
 }

internal/naming/main.go

@@ -116,6 +116,11 @@ func MonitoringService(otelcol string) string {
 	return DNSName(Truncate("%s-monitoring", 63, Service(otelcol)))
 }
 
+// ExtensionService builds the name for the extension service based on the instance.
+func ExtensionService(otelcol string) string {
+	return DNSName(Truncate("%s-extension", 63, Service(otelcol)))
+}
+
 // Service builds the service name based on the instance.
 func Service(otelcol string) string {
 	return DNSName(Truncate("%s-collector", 63, otelcol))