Merge pull request #115 from Icinga/prometheus-config

Prometheus config fallbacks

cmd/icinga-kubernetes/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/icinga/icinga-kubernetes/internal"
 	cachev1 "github.com/icinga/icinga-kubernetes/internal/cache/v1"
 	"github.com/icinga/icinga-kubernetes/pkg/cluster"
+	"github.com/icinga/icinga-kubernetes/pkg/com"
 	"github.com/icinga/icinga-kubernetes/pkg/daemon"
 	kdatabase "github.com/icinga/icinga-kubernetes/pkg/database"
 	"github.com/icinga/icinga-kubernetes/pkg/metrics"
@@ -37,6 +38,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	kclientcmd "k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
+	"net/http"
 	"os"
 	"strings"
 	"sync"
@@ -301,10 +303,35 @@ func main() {
 		return SyncServicePods(ctx, kdb, factory.Core().V1().Services(), factory.Core().V1().Pods())
 	})
 
+	err = internal.SyncPrometheusConfig(ctx, db, &cfg.Prometheus, clusterInstance.Uuid)
+	if err != nil {
+		klog.Error(errors.Wrap(err, "cannot sync prometheus config"))
+	}
+
+	if cfg.Prometheus.Url == "" {
+		err = internal.AutoDetectPrometheus(ctx, clientset, &cfg.Prometheus)
+		if err != nil {
+			klog.Error(errors.Wrap(err, "cannot auto-detect prometheus"))
+		}
+	}
+
 	if cfg.Prometheus.Url != "" {
-		promClient, err := promapi.NewClient(promapi.Config{Address: cfg.Prometheus.Url})
+		var basicAuthTransport http.RoundTripper
+
+		if cfg.Prometheus.Username != "" && cfg.Prometheus.Password != "" {
+			basicAuthTransport = &com.BasicAuthTransport{
+				RoundTripper: http.DefaultTransport,
+				Username:     cfg.Prometheus.Username,
+				Password:     cfg.Prometheus.Password,
+			}
+		}
+
+		promClient, err := promapi.NewClient(promapi.Config{
+			Address:      cfg.Prometheus.Url,
+			RoundTripper: basicAuthTransport,
+		})
 		if err != nil {
-			klog.Fatal(errors.Wrap(err, "error creating promClient"))
+			klog.Fatal(errors.Wrap(err, "error creating Prometheus client"))
 		}
 
 		promApiClient := promv1.NewAPI(promClient)

internal/notifications.go

@@ -77,7 +77,7 @@ func SyncNotificationsConfig(ctx context.Context, db *database.DB, config *notif
 				return errors.Wrap(err, "cannot delete Icinga Notifications config")
 			}
 
-			rows, err := db.QueryxContext(ctx, db.BuildSelectStmt(&schemav1.Config{}, &schemav1.Config{}))
+			rows, err := tx.QueryxContext(ctx, db.BuildSelectStmt(&schemav1.Config{}, &schemav1.Config{}))
 			if err != nil {
 				return errors.Wrap(err, "cannot fetch Icinga Notifications config from DB")
 			}
@@ -103,7 +103,7 @@ func SyncNotificationsConfig(ctx context.Context, db *database.DB, config *notif
 			return nil
 		})
 		if err != nil {
-			return errors.Wrap(err, "cannot upsert Icinga Notifications config")
+			return errors.Wrap(err, "cannot retrieve Icinga Notifications config")
 		}
 	}
 

internal/prometheus.go

@@ -0,0 +1,152 @@
+package internal
+
+import (
+	"context"
+	"fmt"
+	"github.com/icinga/icinga-go-library/database"
+	"github.com/icinga/icinga-go-library/types"
+	"github.com/icinga/icinga-kubernetes/pkg/metrics"
+	schemav1 "github.com/icinga/icinga-kubernetes/pkg/schema/v1"
+	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
+	v1 "k8s.io/api/core/v1"
+	kmetav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"strings"
+)
+
+func SyncPrometheusConfig(ctx context.Context, db *database.DB, config *metrics.PrometheusConfig, clusterUuid types.UUID) error {
+	_true := types.Bool{Bool: true, Valid: true}
+
+	if config.Url != "" {
+		toDb := []schemav1.Config{
+			{ClusterUuid: clusterUuid, Key: schemav1.ConfigKeyPrometheusUrl, Value: config.Url, Locked: _true},
+		}
+
+		if config.Username != "" {
+			toDb = append(
+				toDb,
+				schemav1.Config{ClusterUuid: clusterUuid, Key: schemav1.ConfigKeyPrometheusUsername, Value: config.Username, Locked: _true},
+				schemav1.Config{ClusterUuid: clusterUuid, Key: schemav1.ConfigKeyPrometheusPassword, Value: config.Password, Locked: _true},
+			)
+		}
+
+		err := db.ExecTx(ctx, func(ctx context.Context, tx *sqlx.Tx) error {
+			if _, err := tx.ExecContext(
+				ctx,
+				fmt.Sprintf(
+					`DELETE FROM "%s" WHERE "cluster_uuid" = ? AND "key" LIKE ? AND "locked" = ?`,
+					database.TableName(&schemav1.Config{}),
+				),
+				clusterUuid,
+				`prometheus.%`,
+				_true,
+			); err != nil {
+				return errors.Wrap(err, "cannot delete Prometheus config")
+			}
+
+			stmt, _ := db.BuildInsertStmt(schemav1.Config{})
+			if _, err := tx.NamedExecContext(ctx, stmt, toDb); err != nil {
+				return errors.Wrap(err, "cannot insert Prometheus config")
+			}
+
+			return nil
+		})
+		if err != nil {
+			return errors.Wrap(err, "cannot upsert Prometheus config")
+		}
+	} else {
+		err := db.ExecTx(ctx, func(ctx context.Context, tx *sqlx.Tx) error {
+			if _, err := tx.ExecContext(
+				ctx,
+				fmt.Sprintf(
+					`DELETE FROM "%s" WHERE "cluster_uuid" = ? AND "key" LIKE ? AND "locked" = ?`,
+					database.TableName(&schemav1.Config{}),
+				),
+				clusterUuid,
+				`prometheus.%`,
+				_true,
+			); err != nil {
+				return errors.Wrap(err, "cannot delete Prometheus config")
+			}
+
+			rows, err := tx.QueryxContext(ctx, db.BuildSelectStmt(&schemav1.Config{}, &schemav1.Config{}))
+			if err != nil {
+				return errors.Wrap(err, "cannot fetch Prometheus config from DB")
+			}
+
+			for rows.Next() {
+				var r schemav1.Config
+				if err := rows.StructScan(&r); err != nil {
+					return errors.Wrap(err, "cannot fetch Prometheus config from DB")
+				}
+
+				switch r.Key {
+				case schemav1.ConfigKeyPrometheusUrl:
+					config.Url = r.Value
+				case schemav1.ConfigKeyPrometheusUsername:
+					config.Username = r.Value
+				case schemav1.ConfigKeyPrometheusPassword:
+					config.Password = r.Value
+				}
+			}
+
+			return nil
+		})
+		if err != nil {
+			return errors.Wrap(err, "cannot retrieve Prometheus config")
+		}
+	}
+
+	return nil
+}
+
+// AutoDetectPrometheus tries to auto-detect the Prometheus service in the monitoring namespace and
+// if found sets the URL in the supplied Prometheus configuration. The first service with the label
+// "app.kubernetes.io/name=prometheus" is used. Until now the ServiceTypes ClusterIP and NodePort are supported.
+func AutoDetectPrometheus(ctx context.Context, clientset *kubernetes.Clientset, config *metrics.PrometheusConfig) error {
+	services, err := clientset.CoreV1().Services("monitoring").List(ctx, kmetav1.ListOptions{
+		LabelSelector: "app.kubernetes.io/name=prometheus",
+	})
+	if err != nil {
+		return errors.Wrap(err, "cannot list Prometheus services")
+	}
+
+	if len(services.Items) == 0 {
+		return errors.New("no Prometheus service found")
+	}
+
+	var ip string
+	var port int32
+
+	// Check if we are running in a Kubernetes cluster. If so, use the
+	// service's ClusterIP. Otherwise, use the API Server's IP and NodePort.
+	if _, err = rest.InClusterConfig(); err == nil {
+		for _, service := range services.Items {
+			if service.Spec.Type == v1.ServiceTypeClusterIP {
+				ip = service.Spec.ClusterIP
+				port = service.Spec.Ports[0].Port
+
+				break
+			}
+		}
+	} else if errors.Is(err, rest.ErrNotInCluster) {
+		for _, service := range services.Items {
+			if service.Spec.Type == v1.ServiceTypeNodePort {
+				ip = strings.Split(clientset.RESTClient().Get().URL().Host, ":")[0]
+				port = service.Spec.Ports[0].NodePort
+
+				break
+			}
+		}
+	}
+
+	if ip == "" {
+		return errors.New("no Prometheus found")
+	}
+
+	config.Url = fmt.Sprintf("http://%s:%d", ip, port)
+
+	return nil
+}

pkg/com/basic_auth_transport.go

@@ -0,0 +1,19 @@
+package com
+
+import (
+	"net/http"
+)
+
+// BasicAuthTransport is a http.RoundTripper that authenticates all requests using HTTP Basic Authentication.
+type BasicAuthTransport struct {
+	http.RoundTripper
+	Username string
+	Password string
+}
+
+// RoundTrip executes a single HTTP transaction with the basic auth credentials.
+func (t *BasicAuthTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.SetBasicAuth(t.Username, t.Password)
+
+	return t.RoundTripper.RoundTrip(req)
+}

pkg/metrics/config.go

@@ -1,11 +1,21 @@
 package metrics
 
+import (
+	"github.com/pkg/errors"
+)
+
 // PrometheusConfig defines Prometheus configuration.
 type PrometheusConfig struct {
-	Url string `yaml:"url"`
+	Url      string `yaml:"url"`
+	Username string `yaml:"username"`
+	Password string `yaml:"password"`
 }
 
 // Validate checks constraints in the supplied Prometheus configuration and returns an error if they are violated.
 func (c *PrometheusConfig) Validate() error {
+	if c.Url != "" && (c.Username == "") != (c.Password == "") {
+		return errors.New("both username and password must be provided")
+	}
+
 	return nil
 }

pkg/notifications/client.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"github.com/icinga/icinga-kubernetes/pkg/com"
 	"github.com/pkg/errors"
 	"io"
 	"k8s.io/klog/v2"
@@ -31,10 +32,10 @@ func NewClient(name string, config Config) (*Client, error) {
 
 	return &Client{
 		client: http.Client{
-			Transport: &basicAuthTransport{
+			Transport: &com.BasicAuthTransport{
 				RoundTripper: http.DefaultTransport,
-				username:     config.Username,
-				password:     config.Password,
+				Username:     config.Username,
+				Password:     config.Password,
 			},
 		},
 		userAgent:       name,
@@ -93,15 +94,3 @@ func (c *Client) Stream(ctx context.Context, entities <-chan any) error {
 		}
 	}
 }
-
-type basicAuthTransport struct {
-	http.RoundTripper
-	username string
-	password string
-}
-
-func (t *basicAuthTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	req.SetBasicAuth(t.username, t.password)
-
-	return t.RoundTripper.RoundTrip(req)
-}

pkg/schema/v1/config.go

@@ -18,4 +18,7 @@ const (
 	ConfigKeyNotificationsPassword         ConfigKey = "notifications.password"
 	ConfigKeyNotificationsUrl              ConfigKey = "notifications.url"
 	ConfigKeyNotificationsKubernetesWebUrl ConfigKey = "notifications.kubernetes_web_url"
+	ConfigKeyPrometheusUrl                 ConfigKey = "prometheus.url"
+	ConfigKeyPrometheusUsername            ConfigKey = "prometheus.username"
+	ConfigKeyPrometheusPassword            ConfigKey = "prometheus.password"
 )

schema/mysql/schema.sql

@@ -1021,8 +1021,11 @@ CREATE TABLE config (
     'notifications.url',
     'notifications.username',
     'notifications.password',
-    'notifications.kubernetes_web_url'
-  ) COLLATE utf8mb4_unicode_ci NOT NULL,
+    'notifications.kubernetes_web_url',
+    'prometheus.url',
+    'prometheus.username',
+    'prometheus.password'
+    ) COLLATE utf8mb4_unicode_ci NOT NULL,
   value varchar(255) NOT NULL,
   locked enum('n', 'y') COLLATE utf8mb4_unicode_ci NOT NULL,