[target-allocator] Populate store assets (authorization information) for Prometheus CR watcher

.chloggen/1710-prometheus-cr-scrape-config-credentials.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. operator, target allocator, github action)
+component: target allocator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Populate credentials for Prometheus CR (service and pod monitor) scrape configs.
+
+# One or more tracking issues related to the change
+issues: [1669]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/otel-allocator/main.go

@@ -103,7 +103,7 @@ func main() {
 	defer close(interrupts)
 
 	if *cliConf.PromCRWatcherConf.Enabled {
-		promWatcher, err = allocatorWatcher.NewPrometheusCRWatcher(cfg, cliConf)
+		promWatcher, err = allocatorWatcher.NewPrometheusCRWatcher(setupLog.WithName("prometheus-cr-watcher"), cfg, cliConf)
 		if err != nil {
 			setupLog.Error(err, "Can't start the prometheus watcher")
 			os.Exit(1)
@@ -189,7 +189,7 @@ func main() {
 				select {
 				case event := <-eventChan:
 					eventsMetric.WithLabelValues(event.Source.String()).Inc()
-					loadConfig, err := event.Watcher.LoadConfig()
+					loadConfig, err := event.Watcher.LoadConfig(ctx)
 					if err != nil {
 						setupLog.Error(err, "Unable to load configuration")
 						continue

cmd/otel-allocator/watcher/file.go

@@ -15,6 +15,7 @@
 package watcher
 
 import (
+	"context"
 	"path/filepath"
 
 	"github.com/fsnotify/fsnotify"
@@ -48,7 +49,7 @@ func NewFileWatcher(logger logr.Logger, config config.CLIConfig) (*FileWatcher,
 	}, nil
 }
 
-func (f *FileWatcher) LoadConfig() (*promconfig.Config, error) {
+func (f *FileWatcher) LoadConfig(_ context.Context) (*promconfig.Config, error) {
 	cfg, err := config.Load(f.configFilePath)
 	if err != nil {
 		f.logger.Error(err, "Unable to load configuration")

cmd/otel-allocator/watcher/promOperator.go

@@ -15,11 +15,13 @@
 package watcher
 
 import (
+	"context"
 	"fmt"
 
 	allocatorconfig "github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/config"
 
 	"github.com/go-kit/log"
+	"github.com/go-logr/logr"
 	monitoringv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
 	"github.com/prometheus-operator/prometheus-operator/pkg/assets"
 	monitoringclient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned"
@@ -30,32 +32,28 @@ import (
 	"gopkg.in/yaml.v2"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 )
 
-func NewPrometheusCRWatcher(cfg allocatorconfig.Config, cliConfig allocatorconfig.CLIConfig) (*PrometheusCRWatcher, error) {
+func NewPrometheusCRWatcher(logger logr.Logger, cfg allocatorconfig.Config, cliConfig allocatorconfig.CLIConfig) (*PrometheusCRWatcher, error) {
 	mClient, err := monitoringclient.NewForConfig(cliConfig.ClusterConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, allocatorconfig.DefaultResyncTime, nil) //TODO decide what strategy to use regarding namespaces
-
-	serviceMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName))
+	clientset, err := kubernetes.NewForConfig(cliConfig.ClusterConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	podMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName))
+	factory := informers.NewMonitoringInformerFactories(map[string]struct{}{v1.NamespaceAll: {}}, map[string]struct{}{}, mClient, allocatorconfig.DefaultResyncTime, nil) //TODO decide what strategy to use regarding namespaces
+
+	monitoringInformers, err := getInformers(factory)
 	if err != nil {
 		return nil, err
 	}
 
-	monitoringInformers := map[string]*informers.ForResource{
-		monitoringv1.ServiceMonitorName: serviceMonitorInformers,
-		monitoringv1.PodMonitorName:     podMonitorInformers,
-	}
-
 	generator, err := prometheus.NewConfigGenerator(log.NewNopLogger(), &monitoringv1.Prometheus{}, true) // TODO replace Nop?
 	if err != nil {
 		return nil, err
@@ -66,7 +64,9 @@ func NewPrometheusCRWatcher(cfg allocatorconfig.Config, cliConfig allocatorconfi
 	podMonSelector := getSelector(cfg.PodMonitorSelector)
 
 	return &PrometheusCRWatcher{
+		logger:                 logger,
 		kubeMonitoringClient:   mClient,
+		k8sClient:              clientset,
 		informers:              monitoringInformers,
 		stopChannel:            make(chan struct{}),
 		configGenerator:        generator,
@@ -77,7 +77,9 @@ func NewPrometheusCRWatcher(cfg allocatorconfig.Config, cliConfig allocatorconfi
 }
 
 type PrometheusCRWatcher struct {
-	kubeMonitoringClient *monitoringclient.Clientset
+	logger               logr.Logger
+	kubeMonitoringClient monitoringclient.Interface
+	k8sClient            kubernetes.Interface
 	informers            map[string]*informers.ForResource
 	stopChannel          chan struct{}
 	configGenerator      *prometheus.ConfigGenerator
@@ -88,13 +90,30 @@ type PrometheusCRWatcher struct {
 }
 
 func getSelector(s map[string]string) labels.Selector {
-	sel := labels.NewSelector()
 	if s == nil {
-		return sel
+		return labels.NewSelector()
 	}
 	return labels.SelectorFromSet(s)
 }
 
+// getInformers returns a map of informers for the given resources.
+func getInformers(factory informers.FactoriesForNamespaces) (map[string]*informers.ForResource, error) {
+	serviceMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.ServiceMonitorName))
+	if err != nil {
+		return nil, err
+	}
+
+	podMonitorInformers, err := informers.NewInformersForResource(factory, monitoringv1.SchemeGroupVersion.WithResource(monitoringv1.PodMonitorName))
+	if err != nil {
+		return nil, err
+	}
+
+	return map[string]*informers.ForResource{
+		monitoringv1.ServiceMonitorName: serviceMonitorInformers,
+		monitoringv1.PodMonitorName:     podMonitorInformers,
+	}, nil
+}
+
 // Watch wrapped informers and wait for an initial sync.
 func (w *PrometheusCRWatcher) Watch(upstreamEvents chan Event, upstreamErrors chan error) error {
 	event := Event{
@@ -133,12 +152,13 @@ func (w *PrometheusCRWatcher) Close() error {
 	return nil
 }
 
-func (w *PrometheusCRWatcher) LoadConfig() (*promconfig.Config, error) {
+func (w *PrometheusCRWatcher) LoadConfig(ctx context.Context) (*promconfig.Config, error) {
+	store := assets.NewStore(w.k8sClient.CoreV1(), w.k8sClient.CoreV1())
 	serviceMonitorInstances := make(map[string]*monitoringv1.ServiceMonitor)
-
 	smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(w.serviceMonitorSelector, func(sm interface{}) {
 		monitor := sm.(*monitoringv1.ServiceMonitor)
 		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
+		w.addStoreAssetsForServiceMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.Endpoints, store)
 		serviceMonitorInstances[key] = monitor
 	})
 	if smRetrieveErr != nil {
@@ -149,19 +169,13 @@ func (w *PrometheusCRWatcher) LoadConfig() (*promconfig.Config, error) {
 	pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(w.podMonitorSelector, func(pm interface{}) {
 		monitor := pm.(*monitoringv1.PodMonitor)
 		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
+		w.addStoreAssetsForPodMonitor(ctx, monitor.Name, monitor.Namespace, monitor.Spec.PodMetricsEndpoints, store)
 		podMonitorInstances[key] = monitor
 	})
 	if pmRetrieveErr != nil {
 		return nil, pmRetrieveErr
 	}
 
-	store := assets.Store{
-		TLSAssets:       nil,
-		TokenAssets:     nil,
-		BasicAuthAssets: nil,
-		OAuth2Assets:    nil,
-		SigV4Assets:     nil,
-	}
 	// TODO: We should make these durations configurable
 	prom := &monitoringv1.Prometheus{
 		Spec: monitoringv1.PrometheusSpec{
@@ -171,7 +185,18 @@ func (w *PrometheusCRWatcher) LoadConfig() (*promconfig.Config, error) {
 			},
 		},
 	}
-	generatedConfig, err := w.configGenerator.Generate(prom, serviceMonitorInstances, podMonitorInstances, map[string]*monitoringv1.Probe{}, &store, nil, nil, nil, []string{})
+
+	generatedConfig, err := w.configGenerator.Generate(
+		prom,
+		serviceMonitorInstances,
+		podMonitorInstances,
+		map[string]*monitoringv1.Probe{},
+		store,
+		nil,
+		nil,
+		nil,
+		[]string{},
+	)
 	if err != nil {
 		return nil, err
 	}
@@ -194,3 +219,89 @@ func (w *PrometheusCRWatcher) LoadConfig() (*promconfig.Config, error) {
 	}
 	return promCfg, nil
 }
+
+// addStoreAssetsForServiceMonitor adds authentication / authorization related information to the assets store,
+// based on the service monitor and endpoints specs.
+// This code borrows from
+// https://github.com/prometheus-operator/prometheus-operator/blob/06b5c4189f3f72737766d86103d049115c3aff48/pkg/prometheus/resource_selector.go#L73.
+func (w *PrometheusCRWatcher) addStoreAssetsForServiceMonitor(
+	ctx context.Context,
+	smName, smNamespace string,
+	endps []monitoringv1.Endpoint,
+	store *assets.Store,
+) {
+	var err error
+	for i, endp := range endps {
+		objKey := fmt.Sprintf("serviceMonitor/%s/%s/%d", smNamespace, smName, i)
+
+		if err = store.AddBearerToken(ctx, smNamespace, endp.BearerTokenSecret, objKey); err != nil {
+			break
+		}
+
+		if err = store.AddBasicAuth(ctx, smNamespace, endp.BasicAuth, objKey); err != nil {
+			break
+		}
+
+		if endp.TLSConfig != nil {
+			if err = store.AddTLSConfig(ctx, smNamespace, endp.TLSConfig); err != nil {
+				break
+			}
+		}
+
+		if err = store.AddOAuth2(ctx, smNamespace, endp.OAuth2, objKey); err != nil {
+			break
+		}
+
+		smAuthKey := fmt.Sprintf("serviceMonitor/auth/%s/%s/%d", smNamespace, smName, i)
+		if err = store.AddSafeAuthorizationCredentials(ctx, smNamespace, endp.Authorization, smAuthKey); err != nil {
+			break
+		}
+	}
+
+	if err != nil {
+		w.logger.Error(err, "Failed to obtain credentials for a ServiceMonitor", "serviceMonitor", smName)
+	}
+}
+
+// addStoreAssetsForServiceMonitor adds authentication / authorization related information to the assets store,
+// based on the service monitor and pod metrics endpoints specs.
+// This code borrows from
+// https://github.com/prometheus-operator/prometheus-operator/blob/06b5c4189f3f72737766d86103d049115c3aff48/pkg/prometheus/resource_selector.go#L314.
+func (w *PrometheusCRWatcher) addStoreAssetsForPodMonitor(
+	ctx context.Context,
+	pmName, pmNamespace string,
+	podMetricsEndps []monitoringv1.PodMetricsEndpoint,
+	store *assets.Store,
+) {
+	var err error
+	for i, endp := range podMetricsEndps {
+		objKey := fmt.Sprintf("podMonitor/%s/%s/%d", pmNamespace, pmName, i)
+
+		if err = store.AddBearerToken(ctx, pmNamespace, endp.BearerTokenSecret, objKey); err != nil {
+			break
+		}
+
+		if err = store.AddBasicAuth(ctx, pmNamespace, endp.BasicAuth, objKey); err != nil {
+			break
+		}
+
+		if endp.TLSConfig != nil {
+			if err = store.AddSafeTLSConfig(ctx, pmNamespace, &endp.TLSConfig.SafeTLSConfig); err != nil {
+				break
+			}
+		}
+
+		if err = store.AddOAuth2(ctx, pmNamespace, endp.OAuth2, objKey); err != nil {
+			break
+		}
+
+		smAuthKey := fmt.Sprintf("podMonitor/auth/%s/%s/%d", pmNamespace, pmName, i)
+		if err = store.AddSafeAuthorizationCredentials(ctx, pmNamespace, endp.Authorization, smAuthKey); err != nil {
+			break
+		}
+	}
+
+	if err != nil {
+		w.logger.Error(err, "Failed to obtain credentials for a PodMonitor", "podMonitor", pmName)
+	}
+}