Target Allocator implementation (Part 3 - Collector Enhancement for Prometheus Config Update)

go.mod

@@ -5,6 +5,8 @@ go 1.16
 require (
 	github.com/Masterminds/semver/v3 v3.1.1
 	github.com/go-logr/logr v0.4.0
+	github.com/mitchellh/mapstructure v1.4.1
+	github.com/prometheus/prometheus v1.8.2-0.20210621150501-ff58416a0b02
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.0
 	gopkg.in/yaml.v2 v2.4.0

pkg/collector/container.go

@@ -68,6 +68,15 @@ func Container(cfg config.Config, logger logr.Logger, otelcol v1alpha1.OpenTelem
 		envVars = []corev1.EnvVar{}
 	}
 
+	envVars = append(envVars, corev1.EnvVar{
+		Name: "POD_NAME",
+		ValueFrom: &corev1.EnvVarSource{
+			FieldRef: &corev1.ObjectFieldSelector{
+				FieldPath: "metadata.name",
+			},
+		},
+	})
+
 	return corev1.Container{
 		Name:            naming.Container(),
 		Image:           image,

pkg/collector/container_test.go

@@ -142,12 +142,12 @@ func TestContainerEnvVarsOverridden(t *testing.T) {
 	c := Container(cfg, logger, otelcol)
 
 	// verify
-	assert.Len(t, c.Env, 1)
+	assert.Len(t, c.Env, 2)
 	assert.Equal(t, "foo", c.Env[0].Name)
 	assert.Equal(t, "bar", c.Env[0].Value)
 }
 
-func TestContainerEmptyEnvVarsByDefault(t *testing.T) {
+func TestContainerDefaultEnvVars(t *testing.T) {
 	otelcol := v1alpha1.OpenTelemetryCollector{
 		Spec: v1alpha1.OpenTelemetryCollectorSpec{},
 	}
@@ -158,7 +158,8 @@ func TestContainerEmptyEnvVarsByDefault(t *testing.T) {
 	c := Container(cfg, logger, otelcol)
 
 	// verify
-	assert.Empty(t, c.Env)
+	assert.Len(t, c.Env, 1)
+	assert.Equal(t, c.Env[0].Name, "POD_NAME")
 }
 
 func TestContainerResourceRequirements(t *testing.T) {

pkg/collector/reconcile/config_replace.go

@@ -0,0 +1,84 @@
+// 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 reconcile
+
+import (
+	"fmt"
+
+	"github.com/mitchellh/mapstructure"
+	promconfig "github.com/prometheus/prometheus/config"
+	"github.com/prometheus/prometheus/discovery"
+	"github.com/prometheus/prometheus/discovery/http"
+	_ "github.com/prometheus/prometheus/discovery/install"
+	"gopkg.in/yaml.v2"
+
+	"github.com/open-telemetry/opentelemetry-operator/pkg/collector/adapters"
+	"github.com/open-telemetry/opentelemetry-operator/pkg/naming"
+	ta "github.com/open-telemetry/opentelemetry-operator/pkg/targetallocator/adapters"
+)
+
+type Config struct {
+	PromConfig *promconfig.Config `yaml:"config"`
+}
+
+func ReplaceConfig(params Params) (string, error) {
+	if !params.Instance.Spec.TargetAllocator.Enabled {
+		return params.Instance.Spec.Config, nil
+	}
+	config, err := adapters.ConfigFromString(params.Instance.Spec.Config)
+	if err != nil {
+		return "", err
+	}
+
+	promCfgMap, err := ta.ConfigToPromConfig(params.Instance.Spec.Config)
+	if err != nil {
+		return "", err
+	}
+
+	// yaml marshaling/unsmarshaling is preferred because of the problems associated with the conversion of map to a struct using mapstructure
+	promCfg, err := yaml.Marshal(map[string]interface{}{
+		"config": promCfgMap,
+	})
+	if err != nil {
+		return "", err
+	}
+
+	var cfg Config
+	if err = yaml.UnmarshalStrict(promCfg, &cfg); err != nil {
+		return "", fmt.Errorf("error unmarshaling YAML: %w", err)
+	}
+
+	for i := range cfg.PromConfig.ScrapeConfigs {
+		cfg.PromConfig.ScrapeConfigs[i].ServiceDiscoveryConfigs = discovery.Configs{
+			&http.SDConfig{
+				URL: fmt.Sprintf("http://%s:80/jobs/%s/targets?collector_id=$POD_NAME", naming.TAService(params.Instance), cfg.PromConfig.ScrapeConfigs[i].JobName),
+			},
+		}
+	}
+
+	updPromCfgMap := make(map[string]interface{})
+	if err := mapstructure.Decode(cfg, &updPromCfgMap); err != nil {
+		return "", err
+	}
+
+	// type corecion checks are handled in the ConfigToPromConfig method above
+	config["receivers"].(map[interface{}]interface{})["prometheus"].(map[interface{}]interface{})["config"] = updPromCfgMap["PromConfig"]
+
+	out, err := yaml.Marshal(config)
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
+}

pkg/collector/reconcile/config_replace_test.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 reconcile
+
+import (
+	"testing"
+
+	"github.com/prometheus/prometheus/discovery/http"
+	"github.com/stretchr/testify/assert"
+	"gopkg.in/yaml.v2"
+
+	ta "github.com/open-telemetry/opentelemetry-operator/pkg/targetallocator/adapters"
+)
+
+func TestPrometheusParser(t *testing.T) {
+	param, err := newParams("test/test-img", "../testdata/http_sd_config_test.yaml")
+	assert.NoError(t, err)
+
+	t.Run("should update config with http_sd_config", func(t *testing.T) {
+		actualConfig, err := ReplaceConfig(param)
+		assert.NoError(t, err)
+
+		// prepare
+		var cfg Config
+		promCfgMap, err := ta.ConfigToPromConfig(actualConfig)
+		assert.NoError(t, err)
+
+		promCfg, err := yaml.Marshal(map[string]interface{}{
+			"config": promCfgMap,
+		})
+		assert.NoError(t, err)
+
+		err = yaml.UnmarshalStrict(promCfg, &cfg)
+		assert.NoError(t, err)
+
+		// test
+		expectedMap := map[string]bool{
+			"prometheus": false,
+			"service-x":  false,
+		}
+		for _, scrapeConfig := range cfg.PromConfig.ScrapeConfigs {
+			assert.Len(t, scrapeConfig.ServiceDiscoveryConfigs, 1)
+			assert.Equal(t, scrapeConfig.ServiceDiscoveryConfigs[0].Name(), "http")
+			assert.Equal(t, scrapeConfig.ServiceDiscoveryConfigs[0].(*http.SDConfig).URL, "http://test-targetallocator:80/jobs/"+scrapeConfig.JobName+"/targets?collector_id=$POD_NAME")
+			expectedMap[scrapeConfig.JobName] = true
+		}
+		for k := range expectedMap {
+			assert.True(t, expectedMap[k], k)
+		}
+	})
+
+	t.Run("should not update config with http_sd_config", func(t *testing.T) {
+		param.Instance.Spec.TargetAllocator.Enabled = false
+		actualConfig, err := ReplaceConfig(param)
+		assert.NoError(t, err)
+
+		// prepare
+		var cfg Config
+		promCfgMap, err := ta.ConfigToPromConfig(actualConfig)
+		assert.NoError(t, err)
+
+		promCfg, err := yaml.Marshal(map[string]interface{}{
+			"config": promCfgMap,
+		})
+		assert.NoError(t, err)
+
+		err = yaml.UnmarshalStrict(promCfg, &cfg)
+		assert.NoError(t, err)
+
+		// test
+		expectedMap := map[string]bool{
+			"prometheus": false,
+			"service-x":  false,
+		}
+		for _, scrapeConfig := range cfg.PromConfig.ScrapeConfigs {
+			assert.Len(t, scrapeConfig.ServiceDiscoveryConfigs, 2)
+			assert.Equal(t, scrapeConfig.ServiceDiscoveryConfigs[0].Name(), "file")
+			assert.Equal(t, scrapeConfig.ServiceDiscoveryConfigs[1].Name(), "static")
+			expectedMap[scrapeConfig.JobName] = true
+		}
+		for k := range expectedMap {
+			assert.True(t, expectedMap[k], k)
+		}
+	})
+
+}

pkg/collector/reconcile/configmap.go

@@ -66,6 +66,10 @@ func desiredConfigMap(_ context.Context, params Params) corev1.ConfigMap {
 	name := naming.ConfigMap(params.Instance)
 	labels := collector.Labels(params.Instance)
 	labels["app.kubernetes.io/name"] = name
+	config, err := ReplaceConfig(params)
+	if err != nil {
+		params.Log.V(2).Info("failed to parse config: ", err)
+	}
 
 	return corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -75,7 +79,7 @@ func desiredConfigMap(_ context.Context, params Params) corev1.ConfigMap {
 			Annotations: params.Instance.Annotations,
 		},
 		Data: map[string]string{
-			"collector.yaml": params.Instance.Spec.Config,
+			"collector.yaml": config,
 		},
 	}
 }

pkg/collector/reconcile/configmap_test.go

@@ -51,7 +51,7 @@ receivers:
   prometheus:
     config:
       scrape_configs:
-        job_name: otel-collector
+      - job_name: otel-collector
         scrape_interval: 10s
         static_configs:
           - targets: [ '0.0.0.0:8888', '0.0.0.0:9999' ]
@@ -75,14 +75,64 @@ service:
 
 	})
 
+	t.Run("should return expected collector config map with http_sd_config", func(t *testing.T) {
+		expectedLables["app.kubernetes.io/component"] = "opentelemetry-collector"
+		expectedLables["app.kubernetes.io/name"] = "test-collector"
+
+		expectedData := map[string]string{
+			"collector.yaml": `exporters:
+  logging: null
+processors: null
+receivers:
+  jaeger:
+    protocols:
+      grpc: null
+  prometheus:
+    config:
+      global:
+        scrape_interval: 1m
+        scrape_timeout: 10s
+        evaluation_interval: 1m
+      scrape_configs:
+      - job_name: otel-collector
+        honor_timestamps: true
+        scrape_interval: 10s
+        scrape_timeout: 10s
+        metrics_path: /metrics
+        scheme: http
+        follow_redirects: true
+        http_sd_configs:
+        - follow_redirects: false
+          url: http://test-targetallocator:80/jobs/otel-collector/targets?collector_id=$POD_NAME
+service:
+  pipelines:
+    metrics:
+      exporters:
+      - logging
+      processors: []
+      receivers:
+      - prometheus
+`,
+		}
+
+		param := params()
+		param.Instance.Spec.TargetAllocator.Enabled = true
+		actual := desiredConfigMap(context.Background(), param)
+
+		assert.Equal(t, "test-collector", actual.Name)
+		assert.Equal(t, expectedLables, actual.Labels)
+		assert.Equal(t, expectedData, actual.Data)
+
+	})
+
 	t.Run("should return expected target allocator config map", func(t *testing.T) {
 		expectedLables["app.kubernetes.io/component"] = "opentelemetry-targetallocator"
 		expectedLables["app.kubernetes.io/name"] = "test-targetallocator"
 
 		expectedData := map[string]string{
 			"targetallocator.yaml": `config:
   scrape_configs:
-    job_name: otel-collector
+  - job_name: otel-collector
     scrape_interval: 10s
     static_configs:
     - targets:

pkg/collector/reconcile/deployment_test.go

@@ -121,7 +121,7 @@ func TestExpectedDeployments(t *testing.T) {
 		createObjectIfNotExists(t, "test-targetallocator", &expectedTADeploy)
 		orgUID := expectedTADeploy.OwnerReferences[0].UID
 
-		updatedParam, err := newParams("test/test-img")
+		updatedParam, err := newParams("test/test-img", "")
 		assert.NoError(t, err)
 		updatedDeploy := targetallocator.Deployment(updatedParam.Config, logger, param.Instance)
 		*updatedDeploy.Spec.Replicas = int32(3)
@@ -144,7 +144,7 @@ func TestExpectedDeployments(t *testing.T) {
 		createObjectIfNotExists(t, "test-targetallocator", &expectedTADeploy)
 		orgUID := expectedTADeploy.OwnerReferences[0].UID
 
-		updatedParam, err := newParams("test/test-img")
+		updatedParam, err := newParams("test/test-img", "")
 		assert.NoError(t, err)
 		updatedDeploy := targetallocator.Deployment(updatedParam.Config, logger, updatedParam.Instance)
 

pkg/collector/reconcile/suite_test.go

@@ -83,7 +83,7 @@ func TestMain(m *testing.M) {
 
 func params() Params {
 	replicas := int32(2)
-	configYAML, err := ioutil.ReadFile("test.yaml")
+	configYAML, err := ioutil.ReadFile("../testdata/test.yaml")
 	if err != nil {
 		fmt.Printf("Error getting yaml file: %v", err)
 	}
@@ -120,9 +120,16 @@ func params() Params {
 	}
 }
 
-func newParams(containerImage string) (Params, error) {
+func newParams(containerImage string, file string) (Params, error) {
 	replicas := int32(1)
-	configYAML, err := ioutil.ReadFile("test.yaml")
+	var configYAML []byte
+	var err error
+
+	if file == "" {
+		configYAML, err = ioutil.ReadFile("../testdata/test.yaml")
+	} else {
+		configYAML, err = ioutil.ReadFile(file)
+	}
 	if err != nil {
 		return Params{}, fmt.Errorf("Error getting yaml file: %w", err)
 	}