Use configurable buckets for praefect replication metrics

3 files changed, 57 insertions, 43 deletions

diff --git a/cmd/praefect/main.go b/cmd/praefect/main.go
index d9a454850..eaf0a00a9 100644
--- a/cmd/praefect/main.go
+++ b/cmd/praefect/main.go
@@ -17,6 +17,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/config"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/conn"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect/metrics"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/protoregistry"
 	"gitlab.com/gitlab-org/gitaly/internal/version"
 	"gitlab.com/gitlab-org/labkit/monitoring"
@@ -86,6 +87,8 @@ func configure() (config.Config, error) {
 		logger.WithField("address", conf.PrometheusListenAddr).Info("Starting prometheus listener")
 		conf.Prometheus.Configure()
 
+		metrics.Register(conf.Prometheus)
+
 		go func() {
 			if err := monitoring.Serve(
 				monitoring.WithListenerAddress(conf.PrometheusListenAddr),
diff --git a/internal/praefect/metrics/prom.go b/internal/praefect/metrics/prom.go
new file mode 100644
index 000000000..cb7ada15a
--- /dev/null
+++ b/internal/praefect/metrics/prom.go
@@ -0,0 +1,48 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	promconfig "gitlab.com/gitlab-org/gitaly/internal/config/prometheus"
+)
+
+var (
+	replicationLatency prometheus.Histogram
+
+	replicationJobsInFlight = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: "gitaly",
+			Subsystem: "praefect",
+			Name:      "replication_jobs",
+		},
+	)
+
+	// RecordReplicationLatency records replication latency
+	RecordReplicationLatency = func(d float64) {
+		go replicationLatency.Observe(d)
+	}
+
+	// IncReplicationJobsInFlight increases the gauge that keeps track of in flight replication jobs
+	IncReplicationJobsInFlight = func() {
+		go replicationJobsInFlight.Inc()
+	}
+
+	// DecReplicationJobsInFlight decreases the gauge that keeps track of in flight replication jobs
+	DecReplicationJobsInFlight = func() {
+		go replicationJobsInFlight.Dec()
+	}
+)
+
+// Register registers praefect prometheus metrics
+func Register(conf promconfig.Config) {
+	replicationLatency = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: "gitaly",
+			Subsystem: "praefect",
+			Name:      "replication_latency",
+			Buckets:   conf.GRPCLatencyBuckets,
+		},
+	)
+
+	prometheus.MustRegister(replicationLatency)
+	prometheus.MustRegister(replicationJobsInFlight)
+}
diff --git a/internal/praefect/replicator.go b/internal/praefect/replicator.go
index 6334c5c54..1817f1193 100644
--- a/internal/praefect/replicator.go
+++ b/internal/praefect/replicator.go
@@ -5,54 +5,17 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc"
-
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/conn"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect/metrics"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/models"
 	"gitlab.com/gitlab-org/gitaly/proto/go/gitalypb"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
 )
 
-var (
-	replicationLatency = prometheus.NewHistogram(
-		prometheus.HistogramOpts{
-			Namespace: "gitaly",
-			Subsystem: "praefect",
-			Name:      "replication_latency",
-			Buckets:   prometheus.LinearBuckets(0, 100, 100),
-		},
-	)
-
-	replicationJobsInFlight = prometheus.NewGauge(
-		prometheus.GaugeOpts{
-			Namespace: "gitaly",
-			Subsystem: "praefect",
-			Name:      "replication_jobs",
-		},
-	)
-
-	recordReplicationLatency = func(d float64) {
-		go replicationLatency.Observe(d)
-	}
-
-	incReplicationJobsInFlight = func() {
-		go replicationJobsInFlight.Inc()
-	}
-
-	decReplicationJobsInFlight = func() {
-		go replicationJobsInFlight.Dec()
-	}
-)
-
-func init() {
-	prometheus.MustRegister(replicationLatency)
-	prometheus.MustRegister(replicationJobsInFlight)
-}
-
 // Replicator performs the actual replication logic between two nodes
 type Replicator interface {
 	// Replicate propagates changes from the source to the target
@@ -315,8 +278,8 @@ func (r ReplMgr) processReplJob(ctx context.Context, job datastore.ReplJob) {
 	}
 
 	replStart := time.Now()
-	incReplicationJobsInFlight()
-	defer decReplicationJobsInFlight()
+	metrics.IncReplicationJobsInFlight()
+	defer metrics.DecReplicationJobsInFlight()
 
 	switch job.Change {
 	case datastore.UpdateRepo:
@@ -332,7 +295,7 @@ func (r ReplMgr) processReplJob(ctx context.Context, job datastore.ReplJob) {
 	}
 
 	replDuration := time.Since(replStart)
-	recordReplicationLatency(float64(replDuration / time.Millisecond))
+	metrics.RecordReplicationLatency(float64(replDuration / time.Millisecond))
 
 	if err := r.datastore.UpdateReplJob(job.ID, datastore.JobStateComplete); err != nil {
 		l.WithError(err).Error("error when updating replication job status to complete")