Add new selector for pod and service monitor

apis/v1alpha1/opentelemetrycollector_types.go

@@ -177,6 +177,10 @@ type OpenTelemetryTargetAllocator struct {
 	// All CR instances which the ServiceAccount has access to will be retrieved. This includes other namespaces.
 	// +optional
 	PrometheusCR OpenTelemetryTargetAllocatorPrometheusCR `json:"prometheusCR,omitempty"`
+	// PodMonitors to be selected for target discovery.
+	PodMonitorSelector *map[string]string `json:"podMonitorSelector,omitempty"`
+	// ServiceMonitors to be selected for target discovery.
+	ServiceMonitorSelector *map[string]string `json:"serviceMonitorSelector,omitempty"`
 }
 
 type OpenTelemetryTargetAllocatorPrometheusCR struct {

bundle/manifests/opentelemetry.io_opentelemetrycollectors.yaml

@@ -1711,6 +1711,11 @@ spec:
                     description: Image indicates the container image to use for the
                       OpenTelemetry TargetAllocator.
                     type: string
+                  podMonitorSelector:
+                    additionalProperties:
+                      type: string
+                    description: PodMonitors to be selected for target discovery.
+                    type: object
                   prometheusCR:
                     description: PrometheusCR defines the configuration for the retrieval
                       of PrometheusOperator CRDs ( servicemonitor.monitoring.coreos.com/v1
@@ -1735,6 +1740,11 @@ spec:
                     description: ServiceAccount indicates the name of an existing
                       service account to use with this instance.
                     type: string
+                  serviceMonitorSelector:
+                    additionalProperties:
+                      type: string
+                    description: ServiceMonitors to be selected for target discovery.
+                    type: object
                 type: object
               tolerations:
                 description: Toleration to schedule OpenTelemetry Collector pods.

cmd/otel-allocator/config/config.go

@@ -39,10 +39,12 @@ const DefaultResyncTime = 5 * time.Minute
 const DefaultConfigFilePath string = "/conf/targetallocator.yaml"
 
 type Config struct {
-	LabelSelector      map[string]string  `yaml:"label_selector,omitempty"`
-	Config             *promconfig.Config `yaml:"config"`
-	AllocationStrategy *string            `yaml:"allocation_strategy,omitempty"`
-	FilterStrategy     *string            `yaml:"filter_strategy,omitempty"`
+	LabelSelector          map[string]string  `yaml:"label_selector,omitempty"`
+	Config                 *promconfig.Config `yaml:"config"`
+	AllocationStrategy     *string            `yaml:"allocation_strategy,omitempty"`
+	FilterStrategy         *string            `yaml:"filter_strategy,omitempty"`
+	PodMonitorSelector     *map[string]string `yaml:"pod_monitor_selector,omitempty"`
+	ServiceMonitorSelector *map[string]string `yaml:"service_monitor_selector,omitempty"`
 }
 
 func (c Config) GetAllocationStrategy() string {

cmd/otel-allocator/config/config_test.go

@@ -15,47 +15,143 @@
 package config
 
 import (
+	"fmt"
 	"testing"
+	"time"
+
+	commonconfig "github.com/prometheus/common/config"
+	promconfig "github.com/prometheus/prometheus/config"
 
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
 	"github.com/prometheus/prometheus/discovery/file"
 	"github.com/stretchr/testify/assert"
 )
 
-const testFile = "./testdata/config_test.yaml"
-
-func TestConfigLoad(t *testing.T) {
-	expectedFileSDConfig := &file.SDConfig{
-		Files:           []string{"./file_sd_test.json"},
-		RefreshInterval: model.Duration(300000000000),
+func TestLoad(t *testing.T) {
+	type args struct {
+		file string
 	}
-	expectedStaticSDConfig := discovery.StaticConfig{
+	tests := []struct {
+		name    string
+		args    args
+		want    Config
+		wantErr assert.ErrorAssertionFunc
+	}{
+		{
+			name: "file sd load",
+			args: args{
+				file: "./testdata/config_test.yaml",
+			},
+			want: Config{
+				LabelSelector: map[string]string{
+					"app.kubernetes.io/instance":   "default.test",
+					"app.kubernetes.io/managed-by": "opentelemetry-operator",
+				},
+				Config: &promconfig.Config{
+					GlobalConfig: promconfig.GlobalConfig{
+						ScrapeInterval:     model.Duration(60 * time.Second),
+						ScrapeTimeout:      model.Duration(10 * time.Second),
+						EvaluationInterval: model.Duration(60 * time.Second),
+					},
+					ScrapeConfigs: []*promconfig.ScrapeConfig{
+						{
+							JobName:         "prometheus",
+							HonorTimestamps: true,
+							ScrapeInterval:  model.Duration(60 * time.Second),
+							ScrapeTimeout:   model.Duration(10 * time.Second),
+							MetricsPath:     "/metrics",
+							Scheme:          "http",
+							HTTPClientConfig: commonconfig.HTTPClientConfig{
+								FollowRedirects: true,
+							},
+							ServiceDiscoveryConfigs: []discovery.Config{
+								&file.SDConfig{
+									Files:           []string{"./file_sd_test.json"},
+									RefreshInterval: model.Duration(5 * time.Minute),
+								},
+								discovery.StaticConfig{
+									{
+										Targets: []model.LabelSet{
+											{model.AddressLabel: "prom.domain:9001"},
+											{model.AddressLabel: "prom.domain:9002"},
+											{model.AddressLabel: "prom.domain:9003"},
+										},
+										Labels: model.LabelSet{
+											"my": "label",
+										},
+										Source: "0",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			wantErr: assert.NoError,
+		},
 		{
-			Targets: []model.LabelSet{
-				{model.AddressLabel: "prom.domain:9001"},
-				{model.AddressLabel: "prom.domain:9002"},
-				{model.AddressLabel: "prom.domain:9003"},
+			name: "service monitor pod monitor selector",
+			args: args{
+				file: "./testdata/pod_service_selector_test.yaml",
 			},
-			Labels: model.LabelSet{
-				"my": "label",
+			want: Config{
+				LabelSelector: map[string]string{
+					"app.kubernetes.io/instance":   "default.test",
+					"app.kubernetes.io/managed-by": "opentelemetry-operator",
+				},
+				Config: &promconfig.Config{
+					GlobalConfig: promconfig.GlobalConfig{
+						ScrapeInterval:     model.Duration(60 * time.Second),
+						ScrapeTimeout:      model.Duration(10 * time.Second),
+						EvaluationInterval: model.Duration(60 * time.Second),
+					},
+					ScrapeConfigs: []*promconfig.ScrapeConfig{
+						{
+							JobName:         "prometheus",
+							HonorTimestamps: true,
+							ScrapeInterval:  model.Duration(60 * time.Second),
+							ScrapeTimeout:   model.Duration(10 * time.Second),
+							MetricsPath:     "/metrics",
+							Scheme:          "http",
+							HTTPClientConfig: commonconfig.HTTPClientConfig{
+								FollowRedirects: true,
+							},
+							ServiceDiscoveryConfigs: []discovery.Config{
+								discovery.StaticConfig{
+									{
+										Targets: []model.LabelSet{
+											{model.AddressLabel: "prom.domain:9001"},
+											{model.AddressLabel: "prom.domain:9002"},
+											{model.AddressLabel: "prom.domain:9003"},
+										},
+										Labels: model.LabelSet{
+											"my": "label",
+										},
+										Source: "0",
+									},
+								},
+							},
+						},
+					},
+				},
+				PodMonitorSelector: &map[string]string{
+					"release": "test",
+				},
+				ServiceMonitorSelector: &map[string]string{
+					"release": "test",
+				},
 			},
-			Source: "0",
+			wantErr: assert.NoError,
 		},
 	}
-
-	cfg := Config{}
-	err := unmarshal(&cfg, testFile)
-	assert.NoError(t, err)
-
-	scrapeConfig := *cfg.Config.ScrapeConfigs[0]
-	actualFileSDConfig := scrapeConfig.ServiceDiscoveryConfigs[0]
-	actulaStaticSDConfig := scrapeConfig.ServiceDiscoveryConfigs[1]
-	t.Log(actulaStaticSDConfig)
-
-	assert.Equal(t, cfg.LabelSelector["app.kubernetes.io/instance"], "default.test")
-	assert.Equal(t, cfg.LabelSelector["app.kubernetes.io/managed-by"], "opentelemetry-operator")
-	assert.Equal(t, scrapeConfig.JobName, "prometheus")
-	assert.Equal(t, expectedFileSDConfig, actualFileSDConfig)
-	assert.Equal(t, expectedStaticSDConfig, actulaStaticSDConfig)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := Load(tt.args.file)
+			if !tt.wantErr(t, err, fmt.Sprintf("Load(%v)", tt.args.file)) {
+				return
+			}
+			assert.Equalf(t, tt.want, got, "Load(%v)", tt.args.file)
+		})
+	}
 }

cmd/otel-allocator/config/testdata/pod_service_selector_test.yaml

@@ -0,0 +1,14 @@
+label_selector:
+  app.kubernetes.io/instance: default.test
+  app.kubernetes.io/managed-by: opentelemetry-operator
+pod_monitor_selector:
+  release: test
+service_monitor_selector:
+  release: test
+config:
+  scrape_configs:
+    - job_name: prometheus
+      static_configs:
+        - targets: ["prom.domain:9001", "prom.domain:9002", "prom.domain:9003"]
+          labels:
+            my: label

cmd/otel-allocator/main.go

@@ -83,7 +83,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	watcher, err := allocatorWatcher.NewWatcher(setupLog, cliConf, allocator)
+	watcher, err := allocatorWatcher.NewWatcher(setupLog, cfg, cliConf, allocator)
 	if err != nil {
 		setupLog.Error(err, "Can't start the watchers")
 		os.Exit(1)

cmd/otel-allocator/watcher/main.go

@@ -59,21 +59,21 @@ func (e EventSource) String() string {
 	return eventSourceToString[e]
 }
 
-func NewWatcher(logger logr.Logger, config config.CLIConfig, allocator allocation.Allocator) (*Manager, error) {
+func NewWatcher(logger logr.Logger, cfg config.Config, cliConfig config.CLIConfig, allocator allocation.Allocator) (*Manager, error) {
 	watcher := Manager{
 		allocator: allocator,
 		Events:    make(chan Event),
 		Errors:    make(chan error),
 	}
 
-	fileWatcher, err := newConfigMapWatcher(logger, config)
+	fileWatcher, err := newConfigMapWatcher(logger, cliConfig)
 	if err != nil {
 		return nil, err
 	}
 	watcher.watchers = append(watcher.watchers, &fileWatcher)
 
-	if *config.PromCRWatcherConf.Enabled {
-		promWatcher, err := newCRDMonitorWatcher(config)
+	if *cliConfig.PromCRWatcherConf.Enabled {
+		promWatcher, err := newCRDMonitorWatcher(cfg, cliConfig)
 		if err != nil {
 			return nil, err
 		}

cmd/otel-allocator/watcher/promOperator.go

@@ -33,8 +33,8 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-func newCRDMonitorWatcher(config allocatorconfig.CLIConfig) (*PrometheusCRWatcher, error) {
-	mClient, err := monitoringclient.NewForConfig(config.ClusterConfig)
+func newCRDMonitorWatcher(cfg allocatorconfig.Config, cliConfig allocatorconfig.CLIConfig) (*PrometheusCRWatcher, error) {
+	mClient, err := monitoringclient.NewForConfig(cliConfig.ClusterConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -61,11 +61,17 @@ func newCRDMonitorWatcher(config allocatorconfig.CLIConfig) (*PrometheusCRWatche
 		return nil, err
 	}
 
+	servMonSelector := getSelector(cfg.ServiceMonitorSelector)
+
+	podMonSelector := getSelector(cfg.PodMonitorSelector)
+
 	return &PrometheusCRWatcher{
-		kubeMonitoringClient: mClient,
-		informers:            monitoringInformers,
-		stopChannel:          make(chan struct{}),
-		configGenerator:      generator,
+		kubeMonitoringClient:   mClient,
+		informers:              monitoringInformers,
+		stopChannel:            make(chan struct{}),
+		configGenerator:        generator,
+		serviceMonitorSelector: servMonSelector,
+		podMonitorSelector:     podMonSelector,
 	}, nil
 }
 
@@ -74,6 +80,17 @@ type PrometheusCRWatcher struct {
 	informers            map[string]*informers.ForResource
 	stopChannel          chan struct{}
 	configGenerator      *prometheus.ConfigGenerator
+
+	serviceMonitorSelector labels.Selector
+	podMonitorSelector     labels.Selector
+}
+
+func getSelector(s *map[string]string) labels.Selector {
+	sel := labels.NewSelector()
+	if s == nil {
+		return sel
+	}
+	return labels.SelectorFromSet(*s)
 }
 
 // Start wrapped informers and wait for an initial sync.
@@ -118,7 +135,8 @@ func (w *PrometheusCRWatcher) Close() error {
 
 func (w *PrometheusCRWatcher) CreatePromConfig(kubeConfigPath string) (*promconfig.Config, error) {
 	serviceMonitorInstances := make(map[string]*monitoringv1.ServiceMonitor)
-	smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(labels.NewSelector(), func(sm interface{}) {
+
+	smRetrieveErr := w.informers[monitoringv1.ServiceMonitorName].ListAll(w.serviceMonitorSelector, func(sm interface{}) {
 		monitor := sm.(*monitoringv1.ServiceMonitor)
 		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
 		serviceMonitorInstances[key] = monitor
@@ -128,7 +146,7 @@ func (w *PrometheusCRWatcher) CreatePromConfig(kubeConfigPath string) (*promconf
 	}
 
 	podMonitorInstances := make(map[string]*monitoringv1.PodMonitor)
-	pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(labels.NewSelector(), func(pm interface{}) {
+	pmRetrieveErr := w.informers[monitoringv1.PodMonitorName].ListAll(w.podMonitorSelector, func(pm interface{}) {
 		monitor := pm.(*monitoringv1.PodMonitor)
 		key, _ := cache.DeletionHandlingMetaNamespaceKeyFunc(monitor)
 		podMonitorInstances[key] = monitor

config/crd/bases/opentelemetry.io_opentelemetrycollectors.yaml

@@ -1709,6 +1709,11 @@ spec:
                     description: Image indicates the container image to use for the
                       OpenTelemetry TargetAllocator.
                     type: string
+                  podMonitorSelector:
+                    additionalProperties:
+                      type: string
+                    description: PodMonitors to be selected for target discovery.
+                    type: object
                   prometheusCR:
                     description: PrometheusCR defines the configuration for the retrieval
                       of PrometheusOperator CRDs ( servicemonitor.monitoring.coreos.com/v1
@@ -1733,6 +1738,11 @@ spec:
                     description: ServiceAccount indicates the name of an existing
                       service account to use with this instance.
                     type: string
+                  serviceMonitorSelector:
+                    additionalProperties:
+                      type: string
+                    description: ServiceMonitors to be selected for target discovery.
+                    type: object
                 type: object
               tolerations:
                 description: Toleration to schedule OpenTelemetry Collector pods.