[targetallocator] PrometheusOperator CRD MVC

apis/v1alpha1/opentelemetrycollector_types.go

@@ -120,11 +120,21 @@ type OpenTelemetryTargetAllocator struct {
 	// +optional
 	Enabled bool `json:"enabled,omitempty"`
 
+	// PrometheusCR defines the configuration for Prometheus custom resource retrieval.
+	// +optional
+	PrometheusCR OpenTelemetryTargetAllocatorPrometheusCR `json:"prometheusCR,omitempty"`
+
 	// Image indicates the container image to use for the OpenTelemetry TargetAllocator.
 	// +optional
 	Image string `json:"image,omitempty"`
 }
 
+type OpenTelemetryTargetAllocatorPrometheusCR struct {
+	// Enabled indicates whether to use a Prometheus custom resources as target or not.
+	// +optional
+	Enabled bool `json:"enabled,omitempty"`
+}
+
 // OpenTelemetryCollectorStatus defines the observed state of OpenTelemetryCollector.
 type OpenTelemetryCollectorStatus struct {
 	// Replicas is currently not being set and might be removed in the next version.

bundle/manifests/opentelemetry.io_opentelemetrycollectors.yaml

@@ -647,6 +647,15 @@ spec:
                     description: Image indicates the container image to use for the
                       OpenTelemetry TargetAllocator.
                     type: string
+                  prometheusCR:
+                    description: PrometheusCR defines the configuration for Prometheus
+                      custom resource retrieval.
+                    properties:
+                      enabled:
+                        description: Enabled indicates whether to use a Prometheus
+                          custom resources as target or not.
+                        type: boolean
+                    type: object
                 type: object
               tolerations:
                 description: Toleration to schedule OpenTelemetry Collector pods.

cmd/otel-allocator/Dockerfile

@@ -23,4 +23,4 @@ WORKDIR /root/
 # Copy the pre-built binary file from the previous stage
 COPY --from=builder /app/main .
 
-CMD ["./main"]
+ENTRYPOINT ["./main"]

cmd/otel-allocator/allocation/allocator.go

@@ -2,6 +2,7 @@ package allocation
 
 import (
 	"fmt"
+	"net/url"
 	"sync"
 
 	"github.com/go-logr/logr"
@@ -80,7 +81,7 @@ func (allocator *Allocator) SetWaitingTargets(targets []TargetItem) {
 // SetCollectors sets the set of collectors with key=collectorName, value=Collector object.
 // SetCollectors is called when Collectors are added or removed
 func (allocator *Allocator) SetCollectors(collectors []string) {
-	log := allocator.log.WithValues("opentelemetry-targetallocator")
+	log := allocator.log.WithValues("component", "opentelemetry-targetallocator")
 
 	allocator.m.Lock()
 	defer allocator.m.Unlock()
@@ -132,7 +133,7 @@ func (allocator *Allocator) processWaitingTargets() {
 			allocator.TargetItems[k] = &v
 			targetItem := TargetItem{
 				JobName:   v.JobName,
-				Link:      LinkJSON{fmt.Sprintf("/jobs/%s/targets", v.JobName)},
+				Link:      LinkJSON{fmt.Sprintf("/jobs/%s/targets", url.QueryEscape(v.JobName))},
 				TargetURL: v.TargetURL,
 				Label:     v.Label,
 				Collector: col,

cmd/otel-allocator/allocation/http.go

@@ -2,6 +2,7 @@ package allocation
 
 import (
 	"fmt"
+	"net/url"
 
 	"github.com/prometheus/common/model"
 )
@@ -43,7 +44,7 @@ func GetAllTargetsByJob(job string, cMap map[string][]TargetItem, allocator *All
 				targetGroupList = append(targetGroupList, targetGroupJSON{Targets: targets, Labels: labelSetMap[targets[0]]})
 
 			}
-			displayData[j.Collector.Name] = collectorJSON{Link: fmt.Sprintf("/jobs/%s/targets?collector_id=%s", j.JobName, j.Collector.Name), Jobs: targetGroupList}
+			displayData[j.Collector.Name] = collectorJSON{Link: fmt.Sprintf("/jobs/%s/targets?collector_id=%s", url.QueryEscape(j.JobName), j.Collector.Name), Jobs: targetGroupList}
 
 		}
 	}

cmd/otel-allocator/collector/collector.go

@@ -30,13 +30,8 @@ type Client struct {
 	close         chan struct{}
 }
 
-func NewClient(logger logr.Logger) (*Client, error) {
-	config, err := rest.InClusterConfig()
-	if err != nil {
-		return &Client{}, err
-	}
-
-	clientset, err := kubernetes.NewForConfig(config)
+func NewClient(logger logr.Logger, kubeConfig *rest.Config) (*Client, error) {
+	clientset, err := kubernetes.NewForConfig(kubeConfig)
 	if err != nil {
 		return &Client{}, err
 	}
@@ -50,7 +45,7 @@ func NewClient(logger logr.Logger) (*Client, error) {
 
 func (k *Client) Watch(ctx context.Context, labelMap map[string]string, fn func(collectors []string)) {
 	collectorMap := map[string]bool{}
-	log := k.log.WithValues("opentelemetry-targetallocator")
+	log := k.log.WithValues("component", "opentelemetry-targetallocator")
 
 	opts := metav1.ListOptions{
 		LabelSelector: labels.SelectorFromSet(labelMap).String(),
@@ -83,15 +78,15 @@ func (k *Client) Watch(ctx context.Context, labelMap map[string]string, fn func(
 				return
 			}
 			if msg := runWatch(ctx, k, watcher.ResultChan(), collectorMap, fn); msg != "" {
-				log.Info("Collector pod watch event stopped", msg)
+				log.Info("Collector pod watch event stopped " + msg)
 				return
 			}
 		}
 	}()
 }
 
 func runWatch(ctx context.Context, k *Client, c <-chan watch.Event, collectorMap map[string]bool, fn func(collectors []string)) string {
-	log := k.log.WithValues("opentelemetry-targetallocator")
+	log := k.log.WithValues("component", "opentelemetry-targetallocator")
 	for {
 		select {
 		case <-k.close:

cmd/otel-allocator/config/config.go

@@ -2,31 +2,54 @@ package config
 
 import (
 	"errors"
+	"flag"
 	"fmt"
+	"io/fs"
 	"io/ioutil"
+	"path/filepath"
+	"time"
 
+	"github.com/go-logr/logr"
 	promconfig "github.com/prometheus/prometheus/config"
 	_ "github.com/prometheus/prometheus/discovery/install"
+	"github.com/spf13/pflag"
 	"gopkg.in/yaml.v2"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/util/homedir"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
 
 // ErrInvalidYAML represents an error in the format of the original YAML configuration file.
 var (
 	ErrInvalidYAML = errors.New("couldn't parse the loadbalancer configuration")
 )
 
-const defaultConfigFile string = "/conf/targetallocator.yaml"
+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"`
 }
 
-func Load(file string) (Config, error) {
-	if file == "" {
-		file = defaultConfigFile
-	}
+type PrometheusCRWatcherConfig struct {
+	Enabled *bool
+}
 
+type CLIConfig struct {
+	ListenAddr     *string
+	ConfigFilePath *string
+	ClusterConfig  *rest.Config
+	// KubeConfigFilePath empty if in cluster configuration is in use
+	KubeConfigFilePath string
+	RootLogger         logr.Logger
+	PromCRWatcherConf  PrometheusCRWatcherConfig
+}
+
+func Load(file string) (Config, error) {
 	var cfg Config
 	if err := unmarshal(&cfg, file); err != nil {
 		return Config{}, err
@@ -45,3 +68,36 @@ func unmarshal(cfg *Config, configFile string) error {
 	}
 	return nil
 }
+
+func ParseCLI() (CLIConfig, error) {
+	opts := zap.Options{}
+	opts.BindFlags(flag.CommandLine)
+	cLIConf := CLIConfig{
+		ListenAddr:     pflag.String("listen-addr", ":8080", "The address where this service serves."),
+		ConfigFilePath: pflag.String("config-file", DefaultConfigFilePath, "The path to the config file."),
+		PromCRWatcherConf: PrometheusCRWatcherConfig{
+			Enabled: pflag.Bool("enable-prometheus-cr-watcher", false, "Enable Prometheus CRs as target sources"),
+		},
+	}
+	kubeconfigPath := pflag.String("kubeconfig-path", filepath.Join(homedir.HomeDir(), ".kube", "config"), "absolute path to the KubeconfigPath file")
+	pflag.Parse()
+
+	cLIConf.RootLogger = zap.New(zap.UseFlagOptions(&opts))
+	klog.SetLogger(cLIConf.RootLogger)
+	ctrl.SetLogger(cLIConf.RootLogger)
+
+	clusterConfig, err := clientcmd.BuildConfigFromFlags("", *kubeconfigPath)
+	cLIConf.KubeConfigFilePath = *kubeconfigPath
+	if err != nil {
+		if _, ok := err.(*fs.PathError); !ok {
+			return CLIConfig{}, err
+		}
+		clusterConfig, err = rest.InClusterConfig()
+		if err != nil {
+			return CLIConfig{}, err
+		}
+		cLIConf.KubeConfigFilePath = "" // reset as we use in cluster configuration
+	}
+	cLIConf.ClusterConfig = clusterConfig
+	return cLIConf, nil
+}

cmd/otel-allocator/discovery/discovery.go

@@ -6,16 +6,18 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-logr/logr"
 	"github.com/otel-allocator/allocation"
-	"github.com/otel-allocator/config"
+	allocatorWatcher "github.com/otel-allocator/watcher"
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/discovery"
 )
 
 type Manager struct {
-	log     logr.Logger
-	manager *discovery.Manager
-	logger  log.Logger
-	close   chan struct{}
+	log        logr.Logger
+	manager    *discovery.Manager
+	logger     log.Logger
+	close      chan struct{}
+	configsMap map[allocatorWatcher.EventSource]*config.Config
 }
 
 func NewManager(log logr.Logger, ctx context.Context, logger log.Logger, options ...func(*discovery.Manager)) *Manager {
@@ -27,24 +29,29 @@ func NewManager(log logr.Logger, ctx context.Context, logger log.Logger, options
 		}
 	}()
 	return &Manager{
-		log:     log,
-		manager: manager,
-		logger:  logger,
-		close:   make(chan struct{}),
+		log:        log,
+		manager:    manager,
+		logger:     logger,
+		close:      make(chan struct{}),
+		configsMap: make(map[allocatorWatcher.EventSource]*config.Config),
 	}
 }
 
-func (m *Manager) ApplyConfig(cfg config.Config) error {
+func (m *Manager) ApplyConfig(source allocatorWatcher.EventSource, cfg *config.Config) error {
+	m.configsMap[source] = cfg
+
 	discoveryCfg := make(map[string]discovery.Configs)
 
-	for _, scrapeConfig := range cfg.Config.ScrapeConfigs {
-		discoveryCfg[scrapeConfig.JobName] = scrapeConfig.ServiceDiscoveryConfigs
+	for _, value := range m.configsMap {
+		for _, scrapeConfig := range value.ScrapeConfigs {
+			discoveryCfg[scrapeConfig.JobName] = scrapeConfig.ServiceDiscoveryConfigs
+		}
 	}
 	return m.manager.ApplyConfig(discoveryCfg)
 }
 
 func (m *Manager) Watch(fn func(targets []allocation.TargetItem)) {
-	log := m.log.WithValues("opentelemetry-targetallocator")
+	log := m.log.WithValues("component", "opentelemetry-targetallocator")
 
 	go func() {
 		for {