Add scrape configs endpoint

Makefile

@@ -175,7 +175,7 @@ set-test-image-vars:
 # Build the container image, used only for local dev purposes
 .PHONY: container
 container:
-	docker build -t ${IMG} --build-arg VERSION_PKG=${VERSION_PKG} --build-arg VERSION=${VERSION} --build-arg VERSION_DATE=${VERSION_DATE} --build-arg OTELCOL_VERSION=${OTELCOL_VERSION} --build-arg TARGETALLOCATOR_VERSION=${TARGETALLOCATOR_VERSION} --build-arg AUTO_INSTRUMENTATION_JAVA_VERSION=${AUTO_INSTRUMENTATION_JAVA_VERSION}  --build-arg AUTO_INSTRUMENTATION_NODEJS_VERSION=${AUTO_INSTRUMENTATION_NODEJS_VERSION} --build-arg AUTO_INSTRUMENTATION_PYTHON_VERSION=${AUTO_INSTRUMENTATION_PYTHON_VERSION} --build-arg AUTO_INSTRUMENTATION_DOTNET_VERSION=${AUTO_INSTRUMENTATION_DOTNET_VERSION} .
+	docker buildx build --platform linux/amd64 -t ${IMG} --build-arg VERSION_PKG=${VERSION_PKG} --build-arg VERSION=${VERSION} --build-arg VERSION_DATE=${VERSION_DATE} --build-arg OTELCOL_VERSION=${OTELCOL_VERSION} --build-arg TARGETALLOCATOR_VERSION=${TARGETALLOCATOR_VERSION} --build-arg AUTO_INSTRUMENTATION_JAVA_VERSION=${AUTO_INSTRUMENTATION_JAVA_VERSION}  --build-arg AUTO_INSTRUMENTATION_NODEJS_VERSION=${AUTO_INSTRUMENTATION_NODEJS_VERSION} --build-arg AUTO_INSTRUMENTATION_PYTHON_VERSION=${AUTO_INSTRUMENTATION_PYTHON_VERSION} --build-arg AUTO_INSTRUMENTATION_DOTNET_VERSION=${AUTO_INSTRUMENTATION_DOTNET_VERSION} .
 
 # Push the container image, used only for local dev purposes
 .PHONY: container-push
@@ -184,7 +184,7 @@ container-push:
 
 .PHONY: container-target-allocator
 container-target-allocator:
-	docker build -t ${TARGETALLOCATOR_IMG} cmd/otel-allocator
+	docker buildx build --platform linux/amd64 -t ${TARGETALLOCATOR_IMG} cmd/otel-allocator
 
 .PHONY: start-kind
 start-kind:

cmd/otel-allocator/discovery/discovery.go

@@ -47,6 +47,16 @@ func NewManager(log logr.Logger, ctx context.Context, logger log.Logger, options
 	}
 }
 
+func (m *Manager) GetScrapeConfigs() map[string]*config.ScrapeConfig {
+	jobToScrapeConfig := map[string]*config.ScrapeConfig{}
+	for _, c := range m.configsMap {
+		for _, scrapeConfig := range c.ScrapeConfigs {
+			jobToScrapeConfig[scrapeConfig.JobName] = scrapeConfig
+		}
+	}
+	return jobToScrapeConfig
+}
+
 func (m *Manager) ApplyConfig(source allocatorWatcher.EventSource, cfg *config.Config) error {
 	m.configsMap[source] = cfg
 

cmd/otel-allocator/go.mod

@@ -3,13 +3,17 @@ module github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator
 go 1.19
 
 require (
+	github.com/buraksezer/consistent v0.9.0
+	github.com/cespare/xxhash/v2 v2.1.2
 	github.com/fsnotify/fsnotify v1.5.1
+	github.com/ghodss/yaml v1.0.0
 	github.com/go-kit/log v0.2.0
 	github.com/go-logr/logr v1.2.0
 	github.com/gorilla/mux v1.8.0
 	github.com/prometheus-operator/prometheus-operator v0.53.1
 	github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring v0.53.1
 	github.com/prometheus-operator/prometheus-operator/pkg/client v0.53.1
+	github.com/prometheus/client_golang v1.11.0
 	github.com/prometheus/common v0.32.1
 	github.com/prometheus/prometheus v1.8.2-0.20211214150951-52c693a63be1
 	github.com/spf13/pflag v1.0.5
@@ -42,8 +46,6 @@ require (
 	github.com/aws/aws-sdk-go v1.44.41 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
-	github.com/buraksezer/consistent v0.9.0 // indirect
-	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe // indirect
 	github.com/containerd/containerd v1.5.7 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -59,7 +61,6 @@ require (
 	github.com/envoyproxy/protoc-gen-validate v0.6.2 // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
 	github.com/fatih/color v1.12.0 // indirect
-	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/go-logr/zapr v1.2.0 // indirect
 	github.com/go-openapi/analysis v0.20.0 // indirect
@@ -123,7 +124,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus-community/prom-label-proxy v0.4.1-0.20211215142838-1eac0933d512 // indirect
 	github.com/prometheus/alertmanager v0.23.1-0.20210914172521-e35efbddb66a // indirect
-	github.com/prometheus/client_golang v1.11.0 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common/sigv4 v0.1.0 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect

cmd/otel-allocator/main.go

@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"github.com/ghodss/yaml"
+	yaml2 "gopkg.in/yaml.v2"
 	"net/http"
 	"net/url"
 	"os"
@@ -152,6 +154,7 @@ func newServer(log logr.Logger, allocator allocation.Allocator, discoveryManager
 	}
 	router := mux.NewRouter().UseEncodedPath()
 	router.Use(s.PrometheusMiddleware)
+	router.HandleFunc("/scrape_configs", s.ScrapeConfigsHandler).Methods("GET")
 	router.HandleFunc("/jobs", s.JobHandler).Methods("GET")
 	router.HandleFunc("/jobs/{job_id}/targets", s.TargetsHandler).Methods("GET")
 	router.Path("/metrics").Handler(promhttp.Handler())
@@ -190,6 +193,27 @@ func (s *server) Shutdown(ctx context.Context) error {
 	return s.server.Shutdown(ctx)
 }
 
+// ScrapeConfigsHandler returns the available scrape configuration discovered by the target allocator.
+// The target allocator first marshals these configurations such that the underlying prometheus marshalling is used.
+// After that, the YAML is converted in to a JSON format for consumers to use.
+func (s *server) ScrapeConfigsHandler(w http.ResponseWriter, r *http.Request) {
+	configs := s.discoveryManager.GetScrapeConfigs()
+	configBytes, err := yaml2.Marshal(configs)
+	if err != nil {
+		errorHandler(err, w, r)
+	}
+	jsonConfig, err := yaml.YAMLToJSON(configBytes)
+	if err != nil {
+		errorHandler(err, w, r)
+	}
+	// We don't use the jsonHandler method because we don't want our bytes to be re-encoded
+	w.Header().Set("Content-Type", "application/json")
+	_, err = w.Write(jsonConfig)
+	if err != nil {
+		errorHandler(err, w, r)
+	}
+}
+
 func (s *server) JobHandler(w http.ResponseWriter, r *http.Request) {
 	displayData := make(map[string]allocation.LinkJSON)
 	for _, v := range s.allocator.TargetItems() {

cmd/otel-allocator/watcher/file.go

@@ -12,6 +12,7 @@ import (
 type FileWatcher struct {
 	configFilePath string
 	watcher        *fsnotify.Watcher
+	closer         chan bool
 }
 
 func newConfigMapWatcher(logger logr.Logger, config config.CLIConfig) (FileWatcher, error) {
@@ -24,6 +25,7 @@ func newConfigMapWatcher(logger logr.Logger, config config.CLIConfig) (FileWatch
 	return FileWatcher{
 		configFilePath: *config.ConfigFilePath,
 		watcher:        fileWatcher,
+		closer:         make(chan bool),
 	}, nil
 }
 
@@ -37,6 +39,8 @@ func (f *FileWatcher) Start(upstreamEvents chan Event, upstreamErrors chan error
 	go func() {
 		for {
 			select {
+			case _ = <-f.closer:
+				return
 			case fileEvent := <-f.watcher.Events:
 				if fileEvent.Op == fsnotify.Create {
 					upstreamEvents <- Event{
@@ -52,5 +56,6 @@ func (f *FileWatcher) Start(upstreamEvents chan Event, upstreamErrors chan error
 }
 
 func (f *FileWatcher) Close() error {
+	f.closer <- true
 	return f.watcher.Close()
 }

pkg/collector/container.go

@@ -78,6 +78,15 @@ func Container(cfg config.Config, logger logr.Logger, otelcol v1alpha1.OpenTelem
 		},
 	})
 
+	if otelcol.Spec.TargetAllocator.Enabled {
+		// We need to add a SHARD here so the collector is able to keep targets after the hashmod operation which is
+		// added by default by the operator generator.
+		envVars = append(envVars, corev1.EnvVar{
+			Name:  "SHARD",
+			Value: "0",
+		})
+	}
+
 	var livenessProbe *corev1.Probe
 	if config, err := adapters.ConfigFromString(otelcol.Spec.Config); err == nil {
 		if probe, err := adapters.ConfigToContainerProbe(config); err == nil {

pkg/collector/reconcile/service.go

@@ -235,6 +235,7 @@ func expectedServices(ctx context.Context, params Params, expected []corev1.Serv
 			updated.ObjectMeta.Labels[k] = v
 		}
 		updated.Spec.Ports = desired.Spec.Ports
+		updated.Spec.Selector = desired.Spec.Selector
 
 		patch := client.MergeFrom(existing)
 

pkg/collector/reconcile/service_test.go

@@ -165,7 +165,23 @@ func TestExpectedServices(t *testing.T) {
 		assert.True(t, exists)
 		assert.Equal(t, instanceUID, actual.OwnerReferences[0].UID)
 		assert.Contains(t, actual.Spec.Ports, extraPorts)
+	})
+	t.Run("should update service on version change", func(t *testing.T) {
+		serviceInstance := service("test-collector", params().Instance.Spec.Ports)
+		createObjectIfNotExists(t, "test-collector", &serviceInstance)
 
+		newService := service("test-collector", params().Instance.Spec.Ports)
+		newService.Spec.Selector["app.kubernetes.io/version"] = "Newest"
+		err := expectedServices(context.Background(), params(), []v1.Service{newService})
+		assert.NoError(t, err)
+
+		actual := v1.Service{}
+		exists, err := populateObjectIfExists(t, &actual, types.NamespacedName{Namespace: "default", Name: "test-collector"})
+
+		assert.NoError(t, err)
+		assert.True(t, exists)
+		assert.Equal(t, instanceUID, actual.OwnerReferences[0].UID)
+		assert.Equal(t, "Newest", actual.Spec.Selector["app.kubernetes.io/version"])
 	})
 }