Prometheus support (#450)

Co-authored-by: nils måsén <nils@piksel.se>
Co-authored-by: MihailITPlace <ya.halo-halo@yandex.ru>
Co-authored-by: Sebastiaan Tammer <sebastiaantammer@gmail.com>

pkg/api/api.go

@@ -1,63 +1,76 @@
 package api
 
 import (
-	"errors"
-	"io"
-	"net/http"
-	"os"
-
+	"fmt"
 	log "github.com/sirupsen/logrus"
+	"net/http"
 )
 
-var (
-	lock chan bool
-)
+const tokenMissingMsg = "api token is empty or has not been set. exiting"
 
-func init() {
-	lock = make(chan bool, 1)
-	lock <- true
+// API is the http server responsible for serving the HTTP API endpoints
+type API struct {
+	Token       string
+	hasHandlers bool
 }
 
-// SetupHTTPUpdates configures the endpoint needed for triggering updates via http
-func SetupHTTPUpdates(apiToken string, updateFunction func()) error {
-	if apiToken == "" {
-		return errors.New("api token is empty or has not been set. not starting api")
+// New is a factory function creating a new API instance
+func New(token string) *API {
+	return &API{
+		Token:       token,
+		hasHandlers: false,
+	}
+}
+
+// RequireToken is wrapper around http.HandleFunc that checks token validity
+func (api *API) RequireToken(fn http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") != fmt.Sprintf("Bearer %s", api.Token) {
+			log.Errorf("Invalid token \"%s\"", r.Header.Get("Authorization"))
+			log.Debugf("Expected token to be \"%s\"", api.Token)
+			return
+		}
+		log.Println("Valid token found.")
+		fn(w, r)
+	}
+}
+
+// RegisterFunc is a wrapper around http.HandleFunc that also sets the flag used to determine whether to launch the API
+func (api *API) RegisterFunc(path string, fn http.HandlerFunc) {
+	api.hasHandlers = true
+	http.HandleFunc(path, api.RequireToken(fn))
+}
+
+// RegisterHandler is a wrapper around http.Handler that also sets the flag used to determine whether to launch the API
+func (api *API) RegisterHandler(path string, handler http.Handler) {
+	api.hasHandlers = true
+	http.Handle(path, api.RequireToken(handler.ServeHTTP))
+}
+
+// Start the API and serve over HTTP. Requires an API Token to be set.
+func (api *API) Start(block bool) error {
+
+	if !api.hasHandlers {
+		log.Debug("Watchtower HTTP API skipped.")
+		return nil
 	}
 
-	log.Println("Watchtower HTTP API started.")
-
-	http.HandleFunc("/v1/update", func(w http.ResponseWriter, r *http.Request) {
-		log.Info("Updates triggered by HTTP API request.")
-
-		_, err := io.Copy(os.Stdout, r.Body)
-		if err != nil {
-			log.Println(err)
-			return
-		}
-
-		if r.Header.Get("Token") != apiToken {
-			log.Println("Invalid token. Not updating.")
-			return
-		}
-
-		log.Println("Valid token found. Attempting to update.")
-
-		select {
-		case chanValue := <-lock:
-			defer func() { lock <- chanValue }()
-			updateFunction()
-		default:
-			log.Debug("Skipped. Another update already running.")
-		}
-
-	})
+	if api.Token == "" {
+		log.Fatal(tokenMissingMsg)
+	}
 
+	log.Info("Watchtower HTTP API started.")
+	if block {
+		runHTTPServer()
+	} else {
+		go func() {
+			runHTTPServer()
+		}()
+	}
 	return nil
 }
 
-// WaitForHTTPUpdates starts the http server and listens for requests.
-func WaitForHTTPUpdates() error {
+func runHTTPServer() {
+	log.Info("Serving HTTP")
 	log.Fatal(http.ListenAndServe(":8080", nil))
-	os.Exit(0)
-	return nil
 }

pkg/api/metrics/metrics.go

@@ -0,0 +1,27 @@
+package metrics
+
+import (
+	"github.com/containrrr/watchtower/pkg/metrics"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// Handler is an HTTP handle for serving metric data
+type Handler struct {
+	Path    string
+	Handle  http.HandlerFunc
+	Metrics *metrics.Metrics
+}
+
+// New is a factory function creating a new Metrics instance
+func New() *Handler {
+	m := metrics.Default()
+	handler := promhttp.Handler()
+
+	return &Handler{
+		Path:    "/v1/metrics",
+		Handle:  handler.ServeHTTP,
+		Metrics: m,
+	}
+}

pkg/api/metrics/metrics_test.go

@@ -0,0 +1,77 @@
+package metrics_test
+
+import (
+	"fmt"
+	"github.com/containrrr/watchtower/pkg/metrics"
+	"io/ioutil"
+	"net/http"
+	"testing"
+
+	"github.com/containrrr/watchtower/pkg/api"
+	metricsAPI "github.com/containrrr/watchtower/pkg/api/metrics"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+const Token = "123123123"
+
+func TestContainer(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Metrics Suite")
+}
+
+func runTestServer(m *metricsAPI.Handler) {
+	http.Handle(m.Path, m.Handle)
+	go func() {
+		http.ListenAndServe(":8080", nil)
+	}()
+}
+
+func getWithToken(c http.Client, url string) (*http.Response, error) {
+	req, _ := http.NewRequest("GET", url, nil)
+	req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", Token))
+	return c.Do(req)
+}
+
+var _ = Describe("the metrics", func() {
+	httpAPI := api.New(Token)
+	m := metricsAPI.New()
+	httpAPI.RegisterHandler(m.Path, m.Handle)
+	httpAPI.Start(false)
+
+	// We should likely split this into multiple tests, but as prometheus requires a restart of the binary
+	// to reset the metrics and gauges, we'll just do it all at once.
+
+	It("should serve metrics", func() {
+		metric := &metrics.Metric{
+			Scanned: 4,
+			Updated: 3,
+			Failed:  1,
+		}
+		metrics.RegisterScan(metric)
+		c := http.Client{}
+		res, err := getWithToken(c, "http://localhost:8080/v1/metrics")
+
+		Expect(err).NotTo(HaveOccurred())
+		contents, err := ioutil.ReadAll(res.Body)
+		fmt.Printf("%s\n", string(contents))
+		Expect(string(contents)).To(ContainSubstring("watchtower_containers_updated 3"))
+		Expect(string(contents)).To(ContainSubstring("watchtower_containers_failed 1"))
+		Expect(string(contents)).To(ContainSubstring("watchtower_containers_scanned 4"))
+		Expect(string(contents)).To(ContainSubstring("watchtower_scans_total 1"))
+		Expect(string(contents)).To(ContainSubstring("watchtower_scans_skipped 0"))
+
+		for i := 0; i < 3; i++ {
+			metrics.RegisterScan(nil)
+		}
+
+		res, err = getWithToken(c, "http://localhost:8080/v1/metrics")
+		Expect(err).NotTo(HaveOccurred())
+		contents, err = ioutil.ReadAll(res.Body)
+		fmt.Printf("%s\n", string(contents))
+
+		Expect(string(contents)).To(ContainSubstring("watchtower_scans_total 4"))
+		Expect(string(contents)).To(ContainSubstring("watchtower_scans_skipped 3"))
+	})
+})

pkg/api/update/update.go

@@ -0,0 +1,50 @@
+package update
+
+import (
+	"io"
+	"net/http"
+	"os"
+
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	lock chan bool
+)
+
+// New is a factory function creating a new  Handler instance
+func New(updateFn func()) *Handler {
+	lock = make(chan bool, 1)
+	lock <- true
+
+	return &Handler{
+		fn:   updateFn,
+		Path: "/v1/update",
+	}
+}
+
+// Handler is an API handler used for triggering container update scans
+type Handler struct {
+	fn   func()
+	Path string
+}
+
+// Handle is the actual http.Handle function doing all the heavy lifting
+func (handle *Handler) Handle(w http.ResponseWriter, r *http.Request) {
+	log.Info("Updates triggered by HTTP API request.")
+
+	_, err := io.Copy(os.Stdout, r.Body)
+	if err != nil {
+		log.Println(err)
+		return
+	}
+
+	select {
+	case chanValue := <-lock:
+		defer func() { lock <- chanValue }()
+		handle.fn()
+	default:
+		log.Debug("Skipped. Another update already running.")
+	}
+
+}

pkg/metrics/metrics.go

@@ -0,0 +1,91 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var metrics *Metrics
+
+// Metric is the data points of a single scan
+type Metric struct {
+	Scanned int
+	Updated int
+	Failed  int
+}
+
+// Metrics is the handler processing all individual scan metrics
+type Metrics struct {
+	channel chan *Metric
+	scanned prometheus.Gauge
+	updated prometheus.Gauge
+	failed  prometheus.Gauge
+	total   prometheus.Counter
+	skipped prometheus.Counter
+}
+
+// Register registers metrics for an executed scan
+func (metrics *Metrics) Register(metric *Metric) {
+	metrics.channel <- metric
+}
+
+// Default creates a new metrics handler if none exists, otherwise returns the existing one
+func Default() *Metrics {
+	if metrics != nil {
+		return metrics
+	}
+
+	metrics = &Metrics{
+		scanned: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "watchtower_containers_scanned",
+			Help: "Number of containers scanned for changes by watchtower during the last scan",
+		}),
+		updated: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "watchtower_containers_updated",
+			Help: "Number of containers updated by watchtower during the last scan",
+		}),
+		failed: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "watchtower_containers_failed",
+			Help: "Number of containers where update failed during the last scan",
+		}),
+		total: promauto.NewCounter(prometheus.CounterOpts{
+			Name: "watchtower_scans_total",
+			Help: "Number of scans since the watchtower started",
+		}),
+		skipped: promauto.NewCounter(prometheus.CounterOpts{
+			Name: "watchtower_scans_skipped",
+			Help: "Number of skipped scans since watchtower started",
+		}),
+		channel: make(chan *Metric, 10),
+	}
+
+	go metrics.HandleUpdate(metrics.channel)
+
+	return metrics
+}
+
+// RegisterScan fetches a metric handler and enqueues a metric
+func RegisterScan(metric *Metric) {
+	metrics := Default()
+	metrics.Register(metric)
+}
+
+// HandleUpdate dequeue the metric channel and processes it
+func (metrics *Metrics) HandleUpdate(channel <-chan *Metric) {
+	for change := range channel {
+		if change == nil {
+			// Update was skipped and rescheduled
+			metrics.total.Inc()
+			metrics.skipped.Inc()
+			metrics.scanned.Set(0)
+			metrics.updated.Set(0)
+			metrics.failed.Set(0)
+			continue
+		}
+		// Update metrics with the new values
+		metrics.total.Inc()
+		metrics.scanned.Set(float64(change.Scanned))
+		metrics.updated.Set(float64(change.Updated))
+		metrics.failed.Set(float64(change.Failed))
+	}
+}

pkg/notifications/gotify.go

@@ -87,5 +87,5 @@ func (n *gotifyTypeNotifier) GetURL() string {
 
 func (n *gotifyTypeNotifier) StartNotification()  {}
 func (n *gotifyTypeNotifier) SendNotification()   {}
-func (n *gotifyTypeNotifier) Close()              {}
+func (n *gotifyTypeNotifier) Close() {}
 func (n *gotifyTypeNotifier) Levels() []log.Level { return nil }

pkg/notifications/msteams.go

@@ -63,6 +63,6 @@ func (n *msTeamsTypeNotifier) GetURL() string {
 
 func (n *msTeamsTypeNotifier) StartNotification()          {}
 func (n *msTeamsTypeNotifier) SendNotification()           {}
-func (n *msTeamsTypeNotifier) Close()                      {}
+func (n *msTeamsTypeNotifier) Close() {}
 func (n *msTeamsTypeNotifier) Levels() []log.Level         { return nil }
 func (n *msTeamsTypeNotifier) Fire(entry *log.Entry) error { return nil }