API server and Prometheus Metrics (#10)

* feat(api): introduce a new api server

* feat(cmd): wire and start the API

* feat(notifier) add a metrics middleware

* feat(election): add leader election metrics

* feat(helm): expose api on prometheus-elector, add readiness and liveness

* feat(cmd): optionally export go runtime metrics

* feat(example): demonstrate scrapping agent elector

* chore(README): prometheus metrics are already here

* fix(worflows): turn off cache for lint job

.github/workflows/ci.yaml

@@ -54,7 +54,6 @@ jobs:
         with:
           go-version-file: './go.mod'
           check-latest: true
-          cache: true
       - name: Install Ko
         uses: imjasonh/[email protected]
       - name: Release a New Version

README.md

@@ -58,7 +58,6 @@ This is still a proof of concept, until Prometheus releases https://github.com/p
 
 Here's what will come next!
 
-- Prometheus Metrics!
 - Proxy to route requests to the leader!
 - Release pipeline for the Helm chart?
 - (optional) Notify prometheus using signal ?

api/server.go

@@ -0,0 +1,60 @@
+package api
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"k8s.io/klog/v2"
+)
+
+type Server struct {
+	httpSrv http.Server
+
+	shutdownGraceDelay time.Duration
+}
+
+func NewServer(listenAddr string, shutdownGraceDelay time.Duration, metricsRegistry prometheus.Gatherer) *Server {
+	var mux http.ServeMux
+
+	mux.HandleFunc("/healthz", func(rw http.ResponseWriter, r *http.Request) { rw.WriteHeader(http.StatusOK) })
+	mux.Handle("/metrics", promhttp.HandlerFor(
+		metricsRegistry,
+		promhttp.HandlerOpts{EnableOpenMetrics: true},
+	))
+
+	return &Server{
+		shutdownGraceDelay: shutdownGraceDelay,
+		httpSrv: http.Server{
+			Addr:    listenAddr,
+			Handler: &mux,
+		},
+	}
+}
+
+func (s *Server) Serve(ctx context.Context) error {
+	shutdownDone := make(chan error)
+
+	go func() {
+		<-ctx.Done()
+
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), s.shutdownGraceDelay)
+		defer cancel()
+
+		err := s.httpSrv.Shutdown(shutdownCtx)
+		if err != nil {
+			klog.Info("Server shutdown reported an error, forcing close")
+			err = s.httpSrv.Close()
+		}
+
+		shutdownDone <- err
+	}()
+
+	if err := s.httpSrv.ListenAndServe(); err != http.ErrServerClosed {
+		return err
+	}
+
+	return <-shutdownDone
+}

api/server_test.go

@@ -0,0 +1,58 @@
+package api_test
+
+import (
+	"context"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+
+	"github.com/jlevesy/prometheus-elector/api"
+)
+
+func TestServer_Serve(t *testing.T) {
+	var (
+		ctx, cancel = context.WithCancel(context.Background())
+
+		srv = api.NewServer(
+			":63549",
+			15*time.Second,
+			prometheus.NewRegistry(),
+		)
+
+		srvDone = make(chan struct{})
+	)
+
+	go func() {
+		err := srv.Serve(ctx)
+		require.NoError(t, err)
+
+		close(srvDone)
+	}()
+
+	var attempt = 0
+
+	for {
+		time.Sleep(200 * time.Millisecond)
+		attempt += 1
+
+		if attempt == 5 {
+			t.Fatal("Exausted max retries getting the /healthz endpoint")
+		}
+
+		resp, err := http.Get("http://localhost:63549/healthz")
+		if err != nil {
+			continue
+		}
+		if resp.StatusCode != http.StatusOK {
+			continue
+		}
+
+		break
+	}
+
+	cancel()
+	<-srvDone
+}

cmd/config.go

@@ -37,6 +37,12 @@ type cliConfig struct {
 	notifyRetryMaxAttempts int
 	notifyRetryDelay       time.Duration
 
+	// API setup
+	apiListenAddr         string
+	apiShutdownGraceDelay time.Duration
+
+	runtimeMetrics bool
+
 	// Path to a kubeconfig (if running outside from the cluster).
 	kubeConfigPath string
 }
@@ -93,6 +99,14 @@ func (c *cliConfig) validateRuntimeConfig() error {
 		return errors.New("invalid notify-retry-delay, should be >= 1")
 	}
 
+	if c.apiListenAddr == "" {
+		return errors.New("missing api-listen-address")
+	}
+
+	if c.apiShutdownGraceDelay < 0 {
+		return errors.New("invalid api-shudown-grace-delay, should be >= 0")
+	}
+
 	return nil
 }
 
@@ -110,6 +124,9 @@ func (c *cliConfig) setupFlags() {
 	flag.IntVar(&c.notifyRetryMaxAttempts, "notify-retry-max-attempts", 5, "How many times to retry notifying prometheus on failure.")
 	flag.DurationVar(&c.notifyRetryDelay, "notify-retry-delay", 10*time.Second, "How much time to wait between two notify retries.")
 	flag.BoolVar(&c.init, "init", false, "Only init the prometheus config file")
+	flag.StringVar(&c.apiListenAddr, "api-listen-address", ":9095", "HTTP listen address to use for the API.")
+	flag.DurationVar(&c.apiShutdownGraceDelay, "api-shutdown-grace-delay", 15*time.Second, "Grace delay to apply when shutting down the API server")
+	flag.BoolVar(&c.runtimeMetrics, "runtime-metrics", false, "Export go runtime metrics")
 }
 
 // this is how the http standard library validates the method in NewRequestWithContext.

cmd/config_test.go

@@ -46,7 +46,6 @@ func TestCliConfig_ValidateInitConfig(t *testing.T) {
 				testCase.wantErr,
 				testCase.cfg.validateInitConfig(),
 			)
-
 		})
 	}
 }
@@ -71,6 +70,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantMemberID: "bloupi",
 			wantErr:      nil,
@@ -85,6 +86,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantMemberID: hostname,
 			wantErr:      nil,
@@ -98,6 +101,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("missing lease-name flag"),
 		},
@@ -110,6 +115,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("missing lease-namespace flag"),
 		},
@@ -122,6 +129,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("missing notify-http-url flag"),
 		},
@@ -134,6 +143,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       "",
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("invalid notify-http-method"),
 		},
@@ -146,6 +157,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       "///3eee",
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("invalid notify-http-method"),
 		},
@@ -158,6 +171,8 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: -1,
 				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("invalid notify-retry-max-attempts, should be >= 1"),
 		},
@@ -170,9 +185,39 @@ func TestCliConfig_ValidateRuntimeConfig(t *testing.T) {
 				notifyHTTPMethod:       http.MethodPost,
 				notifyRetryMaxAttempts: 1,
 				notifyRetryDelay:       -10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  15 * time.Second,
 			},
 			wantErr: errors.New("invalid notify-retry-delay, should be >= 1"),
 		},
+		{
+			desc: "missing api-listen-address",
+			cfg: cliConfig{
+				leaseName:              "lease",
+				leaseNamespace:         "namespace",
+				notifyHTTPURL:          "http://reload.com",
+				notifyHTTPMethod:       http.MethodPost,
+				notifyRetryMaxAttempts: 1,
+				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          "",
+				apiShutdownGraceDelay:  15 * time.Second,
+			},
+			wantErr: errors.New("missing api-listen-address"),
+		},
+		{
+			desc: "invalid api-shutdown-grace-delay",
+			cfg: cliConfig{
+				leaseName:              "lease",
+				leaseNamespace:         "namespace",
+				notifyHTTPURL:          "http://reload.com",
+				notifyHTTPMethod:       http.MethodPost,
+				notifyRetryMaxAttempts: 1,
+				notifyRetryDelay:       10 * time.Second,
+				apiListenAddr:          ":5678",
+				apiShutdownGraceDelay:  -15 * time.Second,
+			},
+			wantErr: errors.New("invalid api-shudown-grace-delay, should be >= 0"),
+		},
 	} {
 		t.Run(testCase.desc, func(t *testing.T) {
 			assert.Equal(

cmd/main.go

@@ -12,10 +12,13 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
 
+	"github.com/jlevesy/prometheus-elector/api"
 	"github.com/jlevesy/prometheus-elector/config"
 	"github.com/jlevesy/prometheus-elector/election"
 	"github.com/jlevesy/prometheus-elector/notifier"
 	"github.com/jlevesy/prometheus-elector/watcher"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
 )
 
 func main() {
@@ -30,7 +33,7 @@ func main() {
 	flag.Parse()
 
 	if err := cfg.validateInitConfig(); err != nil {
-		klog.Fatal("Invalid  init config: ", err)
+		klog.Fatal("Invalid init config: ", err)
 	}
 
 	reconciller := config.NewReconciller(cfg.configPath, cfg.outputPath)
@@ -48,10 +51,21 @@ func main() {
 		klog.Fatal("Invalid election config: ", err)
 	}
 
+	metricsRegistry := prometheus.NewRegistry()
+	if cfg.runtimeMetrics {
+		metricsRegistry.MustRegister(collectors.NewBuildInfoCollector())
+		metricsRegistry.MustRegister(collectors.NewGoCollector(
+			collectors.WithGoCollectorRuntimeMetrics(collectors.MetricsAll),
+		))
+	}
+
 	notifier := notifier.WithRetry(
-		notifier.NewHTTP(
-			cfg.notifyHTTPURL,
-			cfg.notifyHTTPMethod,
+		notifier.WithMetrics(
+			metricsRegistry,
+			notifier.NewHTTP(
+				cfg.notifyHTTPURL,
+				cfg.notifyHTTPMethod,
+			),
 		),
 		cfg.notifyRetryMaxAttempts,
 		cfg.notifyRetryDelay,
@@ -86,22 +100,28 @@ func main() {
 		k8sClient,
 		reconciller,
 		notifier,
+		metricsRegistry,
 	)
 
 	if err != nil {
 		klog.Fatal("Can't setup election", err)
 	}
 
+	apiServer := api.NewServer(
+		cfg.apiListenAddr,
+		cfg.apiShutdownGraceDelay,
+		metricsRegistry,
+	)
+
 	grp, grpCtx := errgroup.WithContext(ctx)
 
 	grp.Go(func() error {
 		elector.Run(grpCtx)
 		return nil
 	})
 
-	grp.Go(func() error {
-		return watcher.Watch(grpCtx)
-	})
+	grp.Go(func() error { return watcher.Watch(grpCtx) })
+	grp.Go(func() error { return apiServer.Serve(grpCtx) })
 
 	if err := grp.Wait(); err != nil {
 		klog.Fatal("leader-agent failed, reason: ", err)