Use the new scrape config endpoint for the prometheus receiver

receiver/prometheusreceiver/config.go

@@ -165,8 +165,8 @@ func (cfg *Config) Validate() error {
 }
 
 func (cfg *Config) validatePromConfig(promConfig *promconfig.Config) error {
-	if len(promConfig.ScrapeConfigs) == 0 {
-		return errors.New("no Prometheus scrape_configs")
+	if len(promConfig.ScrapeConfigs) == 0 && cfg.TargetAllocator == nil {
+		return errors.New("no Prometheus scrape_configs or target_allocator set")
 	}
 
 	// Reject features that Prometheus supports but that the receiver doesn't support:

receiver/prometheusreceiver/metrics_receiver.go

@@ -15,23 +15,28 @@
 package prometheusreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver"
 
 import (
+	"bytes"
 	"context"
-	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"regexp"
+	"sync"
 	"time"
 
 	"github.com/go-kit/log"
 	"github.com/mitchellh/hashstructure/v2"
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/discovery"
 	promHTTP "github.com/prometheus/prometheus/discovery/http"
 	"github.com/prometheus/prometheus/scrape"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.uber.org/zap"
+	"gopkg.in/yaml.v2"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver/internal"
 )
@@ -41,33 +46,46 @@ const (
 	gcIntervalDelta   = 1 * time.Minute
 )
 
+// closeOnce ensures that the scrape manager is started after the discovery manager
+type closeOnce struct {
+	C     chan struct{}
+	once  sync.Once
+	Close func()
+}
+
 // pReceiver is the type that provides Prometheus scraper/receiver functionality.
 type pReceiver struct {
-	cfg        *Config
-	consumer   consumer.Metrics
-	cancelFunc context.CancelFunc
+	cfg         *Config
+	consumer    consumer.Metrics
+	cancelFunc  context.CancelFunc
+	reloadReady *closeOnce
 
 	settings                      component.ReceiverCreateSettings
 	scrapeManager                 *scrape.Manager
 	discoveryManager              *discovery.Manager
 	targetAllocatorIntervalTicker *time.Ticker
 }
 
-type linkJSON struct {
-	Link string `json:"_link"`
-}
-
 // New creates a new prometheus.Receiver reference.
 func newPrometheusReceiver(set component.ReceiverCreateSettings, cfg *Config, next consumer.Metrics) *pReceiver {
+	reloadReady := &closeOnce{
+		C: make(chan struct{}),
+	}
+	reloadReady.Close = func() {
+		reloadReady.once.Do(func() {
+			close(reloadReady.C)
+		})
+	}
 	pr := &pReceiver{
-		cfg:      cfg,
-		consumer: next,
-		settings: set,
+		cfg:         cfg,
+		consumer:    next,
+		settings:    set,
+		reloadReady: reloadReady,
 	}
 	return pr
 }
 
-// Start is the method that starts Prometheus scraping and it
+// Start is the method that starts Prometheus scraping. It
 // is controlled by having previously defined a Configuration using perhaps New.
 func (r *pReceiver) Start(_ context.Context, host component.Host) error {
 	discoveryCtx, cancel := context.WithCancel(context.Background())
@@ -92,36 +110,51 @@ func (r *pReceiver) Start(_ context.Context, host component.Host) error {
 
 	allocConf := r.cfg.TargetAllocator
 	if allocConf != nil {
-		go func() {
-			// immediately sync jobs and not wait for the first tick
-			savedHash, _ := r.syncTargetAllocator(uint64(0), allocConf, baseCfg)
-			r.targetAllocatorIntervalTicker = time.NewTicker(allocConf.Interval)
-			for {
-				<-r.targetAllocatorIntervalTicker.C
-				hash, err := r.syncTargetAllocator(savedHash, allocConf, baseCfg)
-				if err != nil {
-					r.settings.Logger.Error(err.Error())
-					continue
-				}
-				savedHash = hash
-			}
-		}()
+		err = r.initTargetAllocator(allocConf, baseCfg)
+		if err != nil {
+			return err
+		}
 	}
 
+	r.reloadReady.Close()
+
 	return nil
 }
 
+func (r *pReceiver) initTargetAllocator(allocConf *targetAllocator, baseCfg *config.Config) error {
+	r.settings.Logger.Info("Starting target allocator discovery")
+	// immediately sync jobs and not wait for the first tick
+	savedHash, err := r.syncTargetAllocator(uint64(0), allocConf, baseCfg)
+	if err != nil {
+		return err
+	}
+	r.targetAllocatorIntervalTicker = time.NewTicker(allocConf.Interval)
+	go func() {
+		<-r.reloadReady.C
+		for {
+			<-r.targetAllocatorIntervalTicker.C
+			hash, newErr := r.syncTargetAllocator(savedHash, allocConf, baseCfg)
+			if newErr != nil {
+				r.settings.Logger.Error(newErr.Error())
+				continue
+			}
+			savedHash = hash
+		}
+	}()
+	return err
+}
+
 // syncTargetAllocator request jobs from targetAllocator and update underlying receiver, if the response does not match the provided compareHash.
 // baseDiscoveryCfg can be used to provide additional ScrapeConfigs which will be added to the retrieved jobs.
 func (r *pReceiver) syncTargetAllocator(compareHash uint64, allocConf *targetAllocator, baseCfg *config.Config) (uint64, error) {
 	r.settings.Logger.Debug("Syncing target allocator jobs")
-	jobObject, err := getJobResponse(allocConf.Endpoint)
+	scrapeConfigsResponse, err := r.getScrapeConfigsResponse(allocConf.Endpoint)
 	if err != nil {
 		r.settings.Logger.Error("Failed to retrieve job list", zap.Error(err))
 		return 0, err
 	}
 
-	hash, err := hashstructure.Hash(jobObject, hashstructure.FormatV2, nil)
+	hash, err := hashstructure.Hash(scrapeConfigsResponse, hashstructure.FormatV2, nil)
 	if err != nil {
 		r.settings.Logger.Error("Failed to hash job list", zap.Error(err))
 		return 0, err
@@ -131,29 +164,29 @@ func (r *pReceiver) syncTargetAllocator(compareHash uint64, allocConf *targetAll
 		return hash, nil
 	}
 
-	cfg := *baseCfg
+	// Clear out the current configurations
+	baseCfg.ScrapeConfigs = []*config.ScrapeConfig{}
 
-	for _, linkJSON := range *jobObject {
+	for jobName, scrapeConfig := range scrapeConfigsResponse {
 		var httpSD promHTTP.SDConfig
 		if allocConf.HTTPSDConfig == nil {
-			httpSD = promHTTP.SDConfig{}
+			httpSD = promHTTP.SDConfig{
+				RefreshInterval: model.Duration(30 * time.Second),
+			}
 		} else {
 			httpSD = *allocConf.HTTPSDConfig
 		}
-
-		httpSD.URL = fmt.Sprintf("%s%s?collector_id=%s", allocConf.Endpoint, linkJSON.Link, allocConf.CollectorID)
-
-		scrapeCfg := &config.ScrapeConfig{
-			JobName: linkJSON.Link,
-			ServiceDiscoveryConfigs: discovery.Configs{
-				&httpSD,
-			},
+		escapedJob := url.QueryEscape(jobName)
+		httpSD.URL = fmt.Sprintf("%s/jobs/%s/targets?collector_id=%s", allocConf.Endpoint, escapedJob, allocConf.CollectorID)
+		httpSD.HTTPClientConfig.FollowRedirects = false
+		scrapeConfig.ServiceDiscoveryConfigs = discovery.Configs{
+			&httpSD,
 		}
 
-		cfg.ScrapeConfigs = append(cfg.ScrapeConfigs, scrapeCfg)
+		baseCfg.ScrapeConfigs = append(baseCfg.ScrapeConfigs, scrapeConfig)
 	}
 
-	err = r.applyCfg(&cfg)
+	err = r.applyCfg(baseCfg)
 	if err != nil {
 		r.settings.Logger.Error("Failed to apply new scrape configuration", zap.Error(err))
 		return 0, err
@@ -162,26 +195,43 @@ func (r *pReceiver) syncTargetAllocator(compareHash uint64, allocConf *targetAll
 	return hash, nil
 }
 
-func getJobResponse(baseURL string) (*map[string]linkJSON, error) {
-	jobURLString := fmt.Sprintf("%s/jobs", baseURL)
-	_, err := url.Parse(jobURLString) // check if valid
+// instantiateShard inserts the SHARD environment variable in the returned configuration
+func (r *pReceiver) instantiateShard(body []byte) []byte {
+	shard, ok := os.LookupEnv("SHARD")
+	if !ok {
+		shard = "0"
+	}
+	return bytes.ReplaceAll(body, []byte("$(SHARD)"), []byte(shard))
+}
+
+func (r *pReceiver) getScrapeConfigsResponse(baseURL string) (map[string]*config.ScrapeConfig, error) {
+	scrapeConfigsURL := fmt.Sprintf("%s/scrape_configs", baseURL)
+	_, err := url.Parse(scrapeConfigsURL) // check if valid
 	if err != nil {
 		return nil, err
 	}
 
-	resp, err := http.Get(jobURLString) //nolint
+	resp, err := http.Get(scrapeConfigsURL) //nolint
 	if err != nil {
 		return nil, err
 	}
 
-	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
 
-	jobObject := &map[string]linkJSON{}
-	err = json.NewDecoder(resp.Body).Decode(jobObject)
+	jobToScrapeConfig := map[string]*config.ScrapeConfig{}
+	envReplacedBody := r.instantiateShard(body)
+	err = yaml.Unmarshal(envReplacedBody, &jobToScrapeConfig)
 	if err != nil {
 		return nil, err
 	}
-	return jobObject, nil
+	err = resp.Body.Close()
+	if err != nil {
+		return nil, err
+	}
+	return jobToScrapeConfig, nil
 }
 
 func (r *pReceiver) applyCfg(cfg *config.Config) error {
@@ -192,6 +242,7 @@ func (r *pReceiver) applyCfg(cfg *config.Config) error {
 	discoveryCfg := make(map[string]discovery.Configs)
 	for _, scrapeConfig := range cfg.ScrapeConfigs {
 		discoveryCfg[scrapeConfig.JobName] = scrapeConfig.ServiceDiscoveryConfigs
+		r.settings.Logger.Info("Scrape job added", zap.String("jobName", scrapeConfig.JobName))
 	}
 	if err := r.discoveryManager.ApplyConfig(discoveryCfg); err != nil {
 		return err
@@ -203,6 +254,7 @@ func (r *pReceiver) initPrometheusComponents(ctx context.Context, host component
 	r.discoveryManager = discovery.NewManager(ctx, logger)
 
 	go func() {
+		r.settings.Logger.Info("Starting discovery manager")
 		if err := r.discoveryManager.Run(); err != nil {
 			r.settings.Logger.Error("Discovery manager failed", zap.Error(err))
 			host.ReportFatalError(err)
@@ -228,7 +280,10 @@ func (r *pReceiver) initPrometheusComponents(ctx context.Context, host component
 		r.cfg.PrometheusConfig.GlobalConfig.ExternalLabels,
 	)
 	r.scrapeManager = scrape.NewManager(&scrape.Options{PassMetadataInContext: true}, logger, store)
+
 	go func() {
+		<-r.reloadReady.C
+		r.settings.Logger.Info("Starting scrape manager")
 		if err := r.scrapeManager.Run(r.discoveryManager.SyncCh()); err != nil {
 			r.settings.Logger.Error("Scrape manager failed", zap.Error(err))
 			host.ReportFatalError(err)
@@ -257,6 +312,7 @@ func gcInterval(cfg *config.Config) time.Duration {
 func (r *pReceiver) Shutdown(context.Context) error {
 	r.cancelFunc()
 	r.scrapeManager.Stop()
+	r.reloadReady.Close()
 	if r.targetAllocatorIntervalTicker != nil {
 		r.targetAllocatorIntervalTicker.Stop()
 	}