/metrics: prometheus metrics for git syncs and bundle builds (#118)

This change adds a couple of bundle build and git sync related metrics to the
service, and exposes them via a /metrics endpoint for consumption by Prometheus.

Related to #90.

Signed-off-by: bdjgs <jgs@bankdata.dk>

Author: bdjgs

e2e/cli/run_git_bundle_build.txtar

@@ -0,0 +1,51 @@
+# local example git repo
+cd repo
+env HOME=.
+exec git init .
+exec git branch -m main
+exec git config --global user.name "botbotbot" 
+exec git config --global user.email "bot@bot.bot" 
+exec git add .
+exec git commit -m initial-commit
+cd ..
+
+exec $OPACTL run --addr 127.0.0.1:8283 --config config.d/bundle.yml --data-dir tmp1 &opactl&
+! stderr .
+! stdout .
+
+exec curl --retry 5 --retry-all-errors http://127.0.0.1:8283/metrics
+
+# assert git sync and bundle build metrics
+stdout 'ocp_git_sync_count_total{repo="./repo/",source="git-data",state="SUCCESS"} 1'
+stdout 'ocp_git_sync_duration_seconds_count{repo="./repo/",source="git-data"} 1'
+stdout 'ocp_git_sync_duration_seconds_sum.*'
+stdout 'ocp_git_sync_duration_seconds_bucket.*'
+stdout 'ocp_bundle_build_duration_seconds_count{bundle="hello-world"} 1'
+stdout 'ocp_bundle_build_duration_seconds_sum.*'
+stdout 'ocp_bundle_build_duration_seconds_bucket.*'
+stdout 'ocp_bundle_build_count_total{bundle="hello-world",state="SUCCESS"} 1'
+
+kill opactl
+
+-- repo/users/data.json --
+[
+  {
+    "id": "alice"
+  },
+  {
+    "id": "bob"
+  }
+]
+-- config.d/bundle.yml --
+bundles:
+  hello-world:
+    object_storage:
+      filesystem:
+        path: bundles/hello-world/bundle.tar.gz
+    requirements:
+    - source: git-data
+sources:
+  git-data:
+    git:
+      repo: ./repo/
+      reference: main

internal/gitsync/gitsync.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/bradleyfalzon/ghinstallation/v2"
 	"github.com/go-git/go-git/v5"
@@ -24,6 +25,7 @@ import (
 	"github.com/go-git/go-git/v5/plumbing/transport/http"
 	gitssh "github.com/go-git/go-git/v5/plumbing/transport/ssh"
 	"github.com/styrainc/opa-control-plane/internal/config"
+	"github.com/styrainc/opa-control-plane/internal/metrics"
 	"golang.org/x/crypto/ssh"
 	"gopkg.in/yaml.v3"
 )
@@ -59,9 +61,14 @@ func New(path string, config config.Git, sourceName string) *Synchronizer {
 // Execute performs the synchronization of the configured Git repository. If the repository does not exist
 // on disk, clone it. If it does exist, pull the latest changes and rebase the local branch onto the remote branch.
 func (s *Synchronizer) Execute(ctx context.Context) error {
+	startTime := time.Now()
+
 	if err := s.execute(ctx); err != nil {
+		metrics.GitSyncFailed(s.sourceName, s.config.Repo)
 		return fmt.Errorf("source %q: git synchronizer: %v: %w", s.sourceName, s.config.Repo, err)
 	}
+
+	metrics.GitSyncSucceeded(s.sourceName, s.config.Repo, startTime)
 	return nil
 }
 

internal/metrics/metrics.go

@@ -0,0 +1,11 @@
+package metrics
+
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func Handler() http.Handler {
+	return promhttp.Handler()
+}

internal/metrics/metrics_gitsync.go

@@ -0,0 +1,36 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	GitSyncCount = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "ocp_git_sync_count_total",
+			Help: "Number of times a git sync has been performed and its state",
+		},
+		[]string{"source", "repo", "state"},
+	)
+
+	GitSyncDuration = promauto.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "ocp_git_sync_duration_seconds",
+			Help:    "Git sync duration in seconds",
+			Buckets: []float64{0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1, 1.5, 2, 5, 10, 30, 60},
+		},
+		[]string{"source", "repo"},
+	)
+)
+
+func GitSyncFailed(source string, repo string) {
+	GitSyncCount.WithLabelValues(source, repo, "FAILED").Inc()
+}
+
+func GitSyncSucceeded(source string, repo string, startTime time.Time) {
+	GitSyncCount.WithLabelValues(source, repo, "SUCCESS").Inc()
+	GitSyncDuration.WithLabelValues(source, repo).Observe(float64(time.Since(startTime).Seconds()))
+}

internal/metrics/metrics_worker.go

@@ -0,0 +1,36 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var (
+	BundleBuildCount = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "ocp_bundle_build_count_total",
+			Help: "Number of times a bundle build has been performed and its state",
+		},
+		[]string{"bundle", "state"},
+	)
+
+	BundleBuildDuration = promauto.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Name:    "ocp_bundle_build_duration_seconds",
+			Help:    "Bundle build duration in seconds",
+			Buckets: []float64{0.1, 0.2, 0.5, 1, 1.5, 2, 5, 10, 30, 60},
+		},
+		[]string{"bundle"},
+	)
+)
+
+func BundleBuildFailed(bundle string, state string) {
+	BundleBuildCount.WithLabelValues(bundle, state).Inc()
+}
+
+func BundleBuildSucceeded(bundle string, state string, startTime time.Time) {
+	BundleBuildCount.WithLabelValues(bundle, state).Inc()
+	BundleBuildDuration.WithLabelValues(bundle).Observe(float64(time.Since(startTime).Seconds()))
+}

internal/server/server.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/styrainc/opa-control-plane/internal/config"
 	"github.com/styrainc/opa-control-plane/internal/database"
+	"github.com/styrainc/opa-control-plane/internal/metrics"
 	"github.com/styrainc/opa-control-plane/internal/server/types"
 )
 
@@ -33,6 +34,7 @@ func (s *Server) Init() *Server {
 		s.router = http.NewServeMux()
 	}
 
+	s.router.Handle("/metrics", metrics.Handler())
 	s.router.HandleFunc("GET    /health", s.health)
 	s.router.HandleFunc("GET    /v1/sources/{source}/data/{path...}", authenticationMiddleware(s.db, s.v1SourcesDataGet))
 	s.router.HandleFunc("POST   /v1/sources/{source}/data/{path...}", authenticationMiddleware(s.db, s.v1SourcesDataPut))

internal/service/worker.go

@@ -9,6 +9,7 @@ import (
 	"github.com/styrainc/opa-control-plane/internal/builder"
 	"github.com/styrainc/opa-control-plane/internal/config"
 	"github.com/styrainc/opa-control-plane/internal/logging"
+	"github.com/styrainc/opa-control-plane/internal/metrics"
 	"github.com/styrainc/opa-control-plane/internal/progress"
 	"github.com/styrainc/opa-control-plane/internal/s3"
 )
@@ -93,6 +94,7 @@ func (worker *BundleWorker) UpdateConfig(b *config.Bundle, sources []*config.Sou
 // Execute runs a bundle synchronization iteration: git sync, bundle construct
 // and then push bundles to object storage.
 func (w *BundleWorker) Execute(ctx context.Context) time.Time {
+	startTime := time.Now() // Used for timing metric
 
 	defer w.bar.Add(1)
 
@@ -106,22 +108,22 @@ func (w *BundleWorker) Execute(ctx context.Context) time.Time {
 	for _, src := range w.sources {
 		if err := src.Wipe(); err != nil {
 			w.log.Warnf("failed to remove a directory for bundle %q: %v", w.bundleConfig.Name, err)
-			return w.report(ctx, BuildStateInternalError, err)
+			return w.report(ctx, BuildStateInternalError, startTime, err)
 		}
 	}
 
 	for _, synchronizer := range w.synchronizers {
 		err := synchronizer.Execute(ctx)
 		if err != nil {
 			w.log.Warnf("failed to synchronize bundle %q: %v", w.bundleConfig.Name, err)
-			return w.report(ctx, BuildStateSyncFailed, err)
+			return w.report(ctx, BuildStateSyncFailed, startTime, err)
 		}
 	}
 
 	for _, src := range w.sources {
 		if err := src.Transform(ctx); err != nil {
 			w.log.Warnf("failed to evaluate source %q for bundle %q: %v", src.Name, w.bundleConfig.Name, err)
-			return w.report(ctx, BuildStateTransformFailed, err)
+			return w.report(ctx, BuildStateTransformFailed, startTime, err)
 		}
 	}
 
@@ -135,24 +137,24 @@ func (w *BundleWorker) Execute(ctx context.Context) time.Time {
 	err := b.Build(ctx)
 	if err != nil {
 		w.log.Warnf("failed to build a bundle %q: %v", w.bundleConfig.Name, err)
-		return w.report(ctx, BuildStateBuildFailed, err)
+		return w.report(ctx, BuildStateBuildFailed, startTime, err)
 	}
 
 	if w.storage != nil {
 		if err := w.storage.Upload(ctx, bytes.NewReader(buffer.Bytes())); err != nil {
 			w.log.Warnf("failed to upload bundle %q: %v", w.bundleConfig.Name, err)
-			return w.report(ctx, BuildStatePushFailed, err)
+			return w.report(ctx, BuildStatePushFailed, startTime, err)
 		}
 
 		w.log.Debugf("Bundle %q built and uploaded.", w.bundleConfig.Name)
-		return w.report(ctx, BuildStateSuccess, nil)
+		return w.report(ctx, BuildStateSuccess, startTime, nil)
 	}
 
 	w.log.Debugf("Bundle %q built.", w.bundleConfig.Name)
-	return w.report(ctx, BuildStateSuccess, nil)
+	return w.report(ctx, BuildStateSuccess, startTime, nil)
 }
 
-func (w *BundleWorker) report(ctx context.Context, state BuildState, err error) time.Time {
+func (w *BundleWorker) report(ctx context.Context, state BuildState, startTime time.Time, err error) time.Time {
 	w.status.State = state
 	if err != nil {
 		if _, ok := err.(ast.Errors); ok {
@@ -162,6 +164,12 @@ func (w *BundleWorker) report(ctx context.Context, state BuildState, err error)
 		}
 	}
 
+	if state == BuildStateSuccess {
+		metrics.BundleBuildSucceeded(w.bundleConfig.Name, state.String(), startTime)
+	} else {
+		metrics.BundleBuildFailed(w.bundleConfig.Name, state.String())
+	}
+
 	if w.singleShot {
 		return w.die(ctx)
 	}