Implement prometheus receiver target allocator (#12586)

* feat(prometheusreciever): add target_collector config base line * implementation draft * feat(prometheusreceiver): functioning target allocator job retrieval * docs(prometheusreceiver): add changelog entry * chore(prometheusreceiver): go mod tidy * chore(prometheusreceiver): code style * chore(prometheusreceiver): fixup root go mod * Merge branch 'main' into implement_prometheus_receiver_target_allocator * fix linting * test: create tests * test: add targetallocator tests * chore: update Prometheus to 0.37.0 * fix: main drift * chore(prometheusreceiver): add targetallocator docs and changelog releaser * fix(prometheusreceiver): apply whole Promconfig instead of only discovery * fix(prometheusreceiver): only abort during validation if both scrape_config and target_allocator is missing * chore(prometheusreceiver): fix linting errors * chore: go mod tidy all * tests: fix config test * chore: do not export unnecessarily types * fix: config errors in some scenarios * fix: config resets * chore: go mod tidy * chore: go mod tidy * fix: discovery generation * tests: add test to confirm that jobs are removed and added to the discovery using the target allocator * chore: remove dev comment * chore: add mapstructure to endpoint * chore: remove io/util import * chore: go mod tidy * chore: modimpi Co-authored-by: Alex Boten <[email protected]>
open-telemetry · Aug 19, 2022 · 41eceae · 41eceae
1 parent 4d5a542
commit 41eceae
Show file tree

Hide file tree

Showing 24 changed files with 863 additions and 23 deletions.
diff --git a/cmd/configschema/go.mod b/cmd/configschema/go.mod
@@ -258,6 +258,7 @@ require (
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/mitchellh/hashstructure v1.1.0 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/moby/sys/mountinfo v0.5.0 // indirect

diff --git a/cmd/configschema/go.sum b/cmd/configschema/go.sum
diff --git a/exporter/prometheusexporter/go.mod b/exporter/prometheusexporter/go.mod
@@ -105,6 +105,7 @@ require (
 	github.com/miekg/dns v1.1.50 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect

diff --git a/exporter/prometheusexporter/go.sum b/exporter/prometheusexporter/go.sum
diff --git a/go.mod b/go.mod
@@ -403,6 +403,7 @@ require (
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/mitchellh/hashstructure v1.1.0 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/moby/sys/mountinfo v0.5.0 // indirect

diff --git a/go.sum b/go.sum
diff --git a/processor/spanmetricsprocessor/go.sum b/processor/spanmetricsprocessor/go.sum
diff --git a/receiver/prometheusexecreceiver/go.mod b/receiver/prometheusexecreceiver/go.mod
@@ -95,6 +95,7 @@ require (
 	github.com/miekg/dns v1.1.50 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
+	github.com/mitchellh/hashstructure/v2 v2.0.2 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect

diff --git a/receiver/prometheusexecreceiver/go.sum b/receiver/prometheusexecreceiver/go.sum
diff --git a/receiver/prometheusreceiver/README.md b/receiver/prometheusreceiver/README.md
@@ -82,6 +82,19 @@ receivers:
               action: keep
 ```
 
+## OpenTelemetry Operator 
+Additional to this static job definitions this receiver allows to query a list of jobs from the 
+OpenTelemetryOperators TargetAllocator or a compatible endpoint. 
+
+```yaml
+receivers:
+  prometheus:
+    target_allocator:
+      endpoint: http://my-targetallocator-service
+      interval: 30s
+      collector_id: collector-1
+```
+
 [sc]: https://github.com/prometheus/prometheus/blob/v2.28.1/docs/configuration/configuration.md#scrape_config
 
 [beta]: https://github.com/open-telemetry/opentelemetry-collector#beta

diff --git a/receiver/prometheusreceiver/config.go b/receiver/prometheusreceiver/config.go
@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net/url"
 	"os"
 	"path/filepath"
 	"sort"
@@ -28,6 +29,7 @@ import (
 	commonconfig "github.com/prometheus/common/config"
 	promconfig "github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/discovery/file"
+	promHTTP "github.com/prometheus/prometheus/discovery/http"
 	"github.com/prometheus/prometheus/discovery/kubernetes"
 	"github.com/prometheus/prometheus/discovery/targetgroup"
 	"go.opentelemetry.io/collector/config"
@@ -38,6 +40,10 @@ import (
 const (
 	// The key for Prometheus scraping configs.
 	prometheusConfigKey = "config"
+
+	// keys to access the http_sd_config from config root
+	targetAllocatorConfigKey       = "target_allocator"
+	targetAllocatorHTTPSDConfigKey = "http_sd_config"
 )
 
 // Config defines configuration for Prometheus receiver.
@@ -55,12 +61,25 @@ type Config struct {
 	UseStartTimeMetric   bool   `mapstructure:"use_start_time_metric"`
 	StartTimeMetricRegex string `mapstructure:"start_time_metric_regex"`
 
+	TargetAllocator *targetAllocator `mapstructure:"target_allocator"`
+
 	// ConfigPlaceholder is just an entry to make the configuration pass a check
 	// that requires that all keys present in the config actually exist on the
 	// structure, ie.: it will error if an unknown key is present.
 	ConfigPlaceholder interface{} `mapstructure:"config"`
 }
 
+type targetAllocator struct {
+	Endpoint    string        `mapstructure:"endpoint"`
+	Interval    time.Duration `mapstructure:"interval"`
+	CollectorID string        `mapstructure:"collector_id"`
+	// ConfigPlaceholder is just an entry to make the configuration pass a check
+	// that requires that all keys present in the config actually exist on the
+	// structure, ie.: it will error if an unknown key is present.
+	ConfigPlaceholder interface{}        `mapstructure:"http_sd_config"`
+	HTTPSDConfig      *promHTTP.SDConfig `mapstructure:"-"`
+}
+
 var _ config.Receiver = (*Config)(nil)
 var _ config.Unmarshallable = (*Config)(nil)
 
@@ -129,9 +148,23 @@ func checkSDFile(filename string) error {
 // Validate checks the receiver configuration is valid.
 func (cfg *Config) Validate() error {
 	promConfig := cfg.PrometheusConfig
-	if promConfig == nil {
-		return nil // noop receiver
+	if promConfig != nil {
+		err := cfg.validatePromConfig(promConfig)
+		if err != nil {
+			return err
+		}
 	}
+
+	if cfg.TargetAllocator != nil {
+		err := cfg.validateTargetAllocatorConfig()
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (cfg *Config) validatePromConfig(promConfig *promconfig.Config) error {
 	if len(promConfig.ScrapeConfigs) == 0 {
 		return errors.New("no Prometheus scrape_configs")
 	}
@@ -209,6 +242,24 @@ func (cfg *Config) Validate() error {
 	return nil
 }
 
+func (cfg *Config) validateTargetAllocatorConfig() error {
+	// validate targetAllocator
+	targetAllocatorConfig := cfg.TargetAllocator
+	if targetAllocatorConfig == nil {
+		return nil
+	}
+	// ensure valid endpoint
+	if _, err := url.ParseRequestURI(targetAllocatorConfig.Endpoint); err != nil {
+		return fmt.Errorf("TargetAllocator endpoint is not valid: %s", targetAllocatorConfig.Endpoint)
+	}
+	// ensure valid collectorID without variables
+	if targetAllocatorConfig.CollectorID == "" || strings.Contains(targetAllocatorConfig.CollectorID, "${") {
+		return fmt.Errorf("CollectorID is not a valid ID")
+	}
+
+	return nil
+}
+
 // Unmarshal a config.Parser into the config struct.
 func (cfg *Config) Unmarshal(componentParser *confmap.Conf) error {
 	if componentParser == nil {
@@ -237,5 +288,28 @@ func (cfg *Config) Unmarshal(componentParser *confmap.Conf) error {
 		return fmt.Errorf("prometheus receiver failed to unmarshal yaml to prometheus config: %w", err)
 	}
 
+	// Unmarshal targetAllocator configs
+	targetAllocatorCfg, err := componentParser.Sub(targetAllocatorConfigKey)
+	if err != nil {
+		return err
+	}
+	targetAllocatorHTTPSDCfg, err := targetAllocatorCfg.Sub(targetAllocatorHTTPSDConfigKey)
+	if err != nil {
+		return err
+	}
+
+	targetAllocatorHTTPSDMap := targetAllocatorHTTPSDCfg.ToStringMap()
+	if len(targetAllocatorHTTPSDMap) != 0 {
+		targetAllocatorHTTPSDMap["url"] = "http://placeholder" // we have to set it as else the marshal will fail
+		httpSDConf, err := yaml.Marshal(targetAllocatorHTTPSDMap)
+		if err != nil {
+			return fmt.Errorf("prometheus receiver failed to marshal config to yaml: %w", err)
+		}
+		err = yaml.UnmarshalStrict(httpSDConf, &cfg.TargetAllocator.HTTPSDConfig)
+		if err != nil {
+			return fmt.Errorf("prometheus receiver failed to unmarshal yaml to prometheus config: %w", err)
+		}
+	}
+
 	return nil
 }
diff --git a/receiver/prometheusreceiver/config_test.go b/receiver/prometheusreceiver/config_test.go
@@ -20,6 +20,8 @@ import (
 	"testing"
 	"time"
 
+	promConfig "github.com/prometheus/common/config"
+	promModel "github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
@@ -48,6 +50,47 @@ func TestLoadConfig(t *testing.T) {
 	assert.Equal(t, time.Duration(r1.PrometheusConfig.ScrapeConfigs[0].ScrapeInterval), 5*time.Second)
 	assert.Equal(t, r1.UseStartTimeMetric, true)
 	assert.Equal(t, r1.StartTimeMetricRegex, "^(.+_)*process_start_time_seconds$")
+
+	assert.Equal(t, "http://my-targetallocator-service", r1.TargetAllocator.Endpoint)
+	assert.Equal(t, 30*time.Second, r1.TargetAllocator.Interval)
+	assert.Equal(t, "collector-1", r1.TargetAllocator.CollectorID)
+	assert.Equal(t, promModel.Duration(60*time.Second), r1.TargetAllocator.HTTPSDConfig.RefreshInterval)
+	assert.Equal(t, "prometheus", r1.TargetAllocator.HTTPSDConfig.HTTPClientConfig.BasicAuth.Username)
+	assert.Equal(t, promConfig.Secret("changeme"), r1.TargetAllocator.HTTPSDConfig.HTTPClientConfig.BasicAuth.Password)
+}
+
+func TestLoadTargetAllocatorConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Receivers[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "config_target_allocator.yaml"), factories)
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	assert.Equal(t, len(cfg.Receivers), 3)
+
+	r0 := cfg.Receivers[config.NewComponentID(typeStr)].(*Config)
+	assert.Nil(t, r0.PrometheusConfig)
+	assert.Equal(t, "http://localhost:8080", r0.TargetAllocator.Endpoint)
+	assert.Equal(t, 30*time.Second, r0.TargetAllocator.Interval)
+	assert.Equal(t, "collector-1", r0.TargetAllocator.CollectorID)
+
+	r1 := cfg.Receivers[config.NewComponentIDWithName(typeStr, "withScrape")].(*Config)
+	assert.Nil(t, r0.PrometheusConfig)
+	assert.Equal(t, "http://localhost:8080", r0.TargetAllocator.Endpoint)
+	assert.Equal(t, 30*time.Second, r0.TargetAllocator.Interval)
+	assert.Equal(t, "collector-1", r0.TargetAllocator.CollectorID)
+
+	assert.Equal(t, 1, len(r1.PrometheusConfig.ScrapeConfigs))
+	assert.Equal(t, "demo", r1.PrometheusConfig.ScrapeConfigs[0].JobName)
+	assert.Equal(t, promModel.Duration(5*time.Second), r1.PrometheusConfig.ScrapeConfigs[0].ScrapeInterval)
+
+	r2 := cfg.Receivers[config.NewComponentIDWithName(typeStr, "withOnlyScrape")].(*Config)
+	assert.Equal(t, 1, len(r2.PrometheusConfig.ScrapeConfigs))
+	assert.Equal(t, "demo", r2.PrometheusConfig.ScrapeConfigs[0].JobName)
+	assert.Equal(t, promModel.Duration(5*time.Second), r2.PrometheusConfig.ScrapeConfigs[0].ScrapeInterval)
 }
 
 func TestLoadConfigFailsOnUnknownSection(t *testing.T) {

diff --git a/receiver/prometheusreceiver/go.mod b/receiver/prometheusreceiver/go.mod
@@ -6,6 +6,7 @@ require (
 	github.com/go-kit/log v0.2.1
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/snappy v0.0.4
+	github.com/mitchellh/hashstructure/v2 v2.0.2
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter v0.58.0
 	github.com/prometheus/common v0.37.0
 	github.com/prometheus/prometheus v0.37.0
@@ -121,6 +122,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite v0.58.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect
+	github.com/pelletier/go-toml v1.9.4 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect

diff --git a/receiver/prometheusreceiver/go.sum b/receiver/prometheusreceiver/go.sum