Fix default internal telemetry endpoint for collector

#3730

Author: shuheiktgw

apis/v1beta1/collector_webhook_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/go-logr/logr"
+	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
@@ -158,7 +159,7 @@ func TestCollectorDefaultingWebhook(t *testing.T) {
 					Mode:            v1beta1.ModeDeployment,
 					UpgradeStrategy: v1beta1.UpgradeStrategyAutomatic,
 					Config: func() v1beta1.Config {
-						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"endpoint":"0.0.0.0:4317"},"http":{"endpoint":"0.0.0.0:4318"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"address":"0.0.0.0:8888"}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
+						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"endpoint":"0.0.0.0:4317"},"http":{"endpoint":"0.0.0.0:4318"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"readers":[{"pull":{"exporter":{"prometheus":{"host":"0.0.0.0","port":8888}}}}]}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
 						var cfg v1beta1.Config
 						require.NoError(t, yaml.Unmarshal([]byte(input), &cfg))
 						return cfg
@@ -171,7 +172,7 @@ func TestCollectorDefaultingWebhook(t *testing.T) {
 			otelcol: v1beta1.OpenTelemetryCollector{
 				Spec: v1beta1.OpenTelemetryCollectorSpec{
 					Config: func() v1beta1.Config {
-						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"headers":{"example":"another"}},"http":{"endpoint":"0.0.0.0:4000"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"address":"1.2.3.4:7654"}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
+						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"headers":{"example":"another"}},"http":{"endpoint":"0.0.0.0:4000"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"readers":[{"pull":{"exporter":{"prometheus":{"host":"localhost","port":9999}}}}]}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
 						var cfg v1beta1.Config
 						require.NoError(t, yaml.Unmarshal([]byte(input), &cfg))
 						return cfg
@@ -190,7 +191,7 @@ func TestCollectorDefaultingWebhook(t *testing.T) {
 					Mode:            v1beta1.ModeDeployment,
 					UpgradeStrategy: v1beta1.UpgradeStrategyAutomatic,
 					Config: func() v1beta1.Config {
-						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"endpoint":"0.0.0.0:4317","headers":{"example":"another"}},"http":{"endpoint":"0.0.0.0:4000"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"address":"1.2.3.4:7654"}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
+						const input = `{"receivers":{"otlp":{"protocols":{"grpc":{"endpoint":"0.0.0.0:4317","headers":{"example":"another"}},"http":{"endpoint":"0.0.0.0:4000"}}}},"exporters":{"debug":null},"service":{"telemetry":{"metrics":{"readers":[{"pull":{"exporter":{"prometheus":{"host":"localhost","port":9999}}}}]}},"pipelines":{"traces":{"receivers":["otlp"],"exporters":["debug"]}}}}`
 						var cfg v1beta1.Config
 						require.NoError(t, yaml.Unmarshal([]byte(input), &cfg))
 						return cfg
@@ -547,7 +548,9 @@ func TestCollectorDefaultingWebhook(t *testing.T) {
 				assert.NoError(t, test.expected.Spec.Config.Service.ApplyDefaults(logr.Discard()), "could not apply defaults")
 			}
 			assert.NoError(t, err)
-			assert.Equal(t, test.expected, test.otelcol)
+			if diff := cmp.Diff(test.expected, test.otelcol); diff != "" {
+				t.Errorf("v1beta1.OpenTelemetryCollector mismatch (-want +got):\n%s", diff)
+			}
 		})
 	}
 }

apis/v1beta1/config.go

@@ -474,23 +474,81 @@ func (s *Service) ApplyDefaults(logger logr.Logger) error {
 		return err
 	}
 
-	tm := &AnyConfig{
-		Object: map[string]interface{}{
-			"metrics": map[string]interface{}{
-				"address": fmt.Sprintf("%s:%d", telemetryAddr, telemetryPort),
-			},
-		},
+	if s.Telemetry == nil {
+		s.Telemetry = &AnyConfig{
+			Object: map[string]any{},
+		}
 	}
 
-	if s.Telemetry == nil {
-		s.Telemetry = tm
+	metrics, ok := s.Telemetry.Object["metrics"]
+	if !ok {
+		metrics = map[string]any{}
+		s.Telemetry.Object["metrics"] = metrics
+	}
+	metricsMap, ok := metrics.(map[string]any)
+	if !ok {
+		logger.Info("metrics field is not a map")
+		return nil
+	}
+
+	readers, ok := metricsMap["readers"]
+	if !ok {
+		readers = []any{}
+		metricsMap["readers"] = readers
+	}
+	readersSlice, ok := readers.([]any)
+	if !ok {
+		logger.Info("readers field is not a slice")
 		return nil
 	}
-	// NOTE: Merge without overwrite. If a telemetry endpoint is specified, the defaulting
-	// respects the configuration and returns an equal value.
-	if err := mergo.Merge(s.Telemetry, tm); err != nil {
-		return fmt.Errorf("telemetry config merge failed: %w", err)
+
+	var found bool
+	for _, reader := range readersSlice {
+		readerMap, ok := reader.(map[string]any)
+		if !ok {
+			logger.Info("reader field is not a map")
+			return nil
+		}
+
+		pull, ok := readerMap["pull"]
+		if !ok {
+			continue
+		}
+		pullMap, ok := pull.(map[string]any)
+		if !ok {
+			logger.Info("pull field is not a map")
+			return nil
+		}
+
+		exporter, ok := pullMap["exporter"]
+		if !ok {
+			continue
+		}
+		exporterMap, ok := exporter.(map[string]any)
+		if !ok {
+			logger.Info("exporter field is not a map")
+			return nil
+		}
+
+		if _, ok := exporterMap["prometheus"]; ok {
+			found = true
+			break
+		}
+	}
+	if !found {
+		readersSlice = append(readersSlice, map[string]any{
+			"pull": map[string]any{
+				"exporter": map[string]any{
+					"prometheus": map[string]any{
+						"host": telemetryAddr,
+						"port": int(telemetryPort),
+					},
+				},
+			},
+		})
+		metricsMap["readers"] = readersSlice
 	}
+
 	return nil
 }
 

pkg/collector/upgrade/v0_111_0_test.go

@@ -34,9 +34,20 @@ func Test0_111_0Upgrade(t *testing.T) {
 	defaultCollectorWithConfig := defaultCollector.DeepCopy()
 
 	defaultCollectorWithConfig.Spec.Config.Service.Telemetry = &v1beta1.AnyConfig{
-		Object: map[string]interface{}{
-			"metrics": map[string]interface{}{
-				"address": "1.2.3.4:8888",
+		Object: map[string]any{
+			"metrics": map[string]any{
+				"readers": []any{
+					map[string]any{
+						"pull": map[string]any{
+							"exporter": map[string]any{
+								"prometheus": map[string]any{
+									"host": "1.2.3.4",
+									"port": 8888,
+								},
+							},
+						},
+					},
+				},
 			},
 		},
 	}
@@ -59,7 +70,18 @@ func Test0_111_0Upgrade(t *testing.T) {
 				col.Spec.Config.Service.Telemetry = &v1beta1.AnyConfig{
 					Object: map[string]interface{}{
 						"metrics": map[string]interface{}{
-							"address": "0.0.0.0:8888",
+							"readers": []any{
+								map[string]any{
+									"pull": map[string]any{
+										"exporter": map[string]any{
+											"prometheus": map[string]any{
+												"host": "0.0.0.0",
+												"port": 8888,
+											},
+										},
+									},
+								},
+							},
 						},
 					},
 				}