From 79a17061007727e0f576e3b8370b394e9bed145b Mon Sep 17 00:00:00 2001
From: Sandeep Sukhani <sandeep.d.sukhani@gmail.com>
Date: Tue, 25 Aug 2020 14:14:37 +0530
Subject: [PATCH] add some metrics for monitoring compactor

---
 pkg/loki/modules.go                           |  2 +-
 .../stores/shipper/compactor/compactor.go     | 23 +++++++++++--
 .../stores/shipper/compactor/metrics.go       | 33 +++++++++++++++++++
 3 files changed, 54 insertions(+), 4 deletions(-)
 create mode 100644 pkg/storage/stores/shipper/compactor/metrics.go

diff --git a/pkg/loki/modules.go b/pkg/loki/modules.go
index 2bf467748c87..76cc9b05243b 100644
--- a/pkg/loki/modules.go
+++ b/pkg/loki/modules.go
@@ -436,7 +436,7 @@ func (t *Loki) initMemberlistKV() (services.Service, error) {
 
 func (t *Loki) initCompactor() (services.Service, error) {
 	var err error
-	t.compactor, err = compactor.NewCompactor(t.cfg.CompactorConfig, t.cfg.StorageConfig.Config)
+	t.compactor, err = compactor.NewCompactor(t.cfg.CompactorConfig, t.cfg.StorageConfig.Config, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/storage/stores/shipper/compactor/compactor.go b/pkg/storage/stores/shipper/compactor/compactor.go
index cc2783750fa7..e0bd7b78ec61 100644
--- a/pkg/storage/stores/shipper/compactor/compactor.go
+++ b/pkg/storage/stores/shipper/compactor/compactor.go
@@ -13,6 +13,7 @@ import (
 	pkg_util "github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/services"
 	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/grafana/loki/pkg/storage/stores/shipper"
 	"github.com/grafana/loki/pkg/storage/stores/util"
@@ -34,9 +35,11 @@ type Compactor struct {
 
 	cfg          Config
 	objectClient chunk.ObjectClient
+
+	metrics *metrics
 }
 
-func NewCompactor(cfg Config, storageConfig storage.Config) (*Compactor, error) {
+func NewCompactor(cfg Config, storageConfig storage.Config, r prometheus.Registerer) (*Compactor, error) {
 	objectClient, err := storage.NewObjectClient(cfg.SharedStoreType, storageConfig)
 	if err != nil {
 		return nil, err
@@ -50,6 +53,7 @@ func NewCompactor(cfg Config, storageConfig storage.Config) (*Compactor, error)
 	compactor := Compactor{
 		cfg:          cfg,
 		objectClient: util.NewPrefixedObjectClient(objectClient, shipper.StorageKeyPrefix),
+		metrics:      newMetrics(r),
 	}
 
 	compactor.Service = services.NewTimerService(4*time.Hour, nil, compactor.Run, nil)
@@ -57,8 +61,19 @@ func NewCompactor(cfg Config, storageConfig storage.Config) (*Compactor, error)
 }
 
 func (c *Compactor) Run(ctx context.Context) error {
+	status := statusSuccess
+	start := time.Now()
+
+	defer func() {
+		c.metrics.compactTablesOperationTotal.WithLabelValues(status).Inc()
+		if status == statusSuccess {
+			c.metrics.compactTablesOperationDurationSeconds.Set(time.Since(start).Seconds())
+		}
+	}()
+
 	_, dirs, err := c.objectClient.List(ctx, "")
 	if err != nil {
+		status = statusFailure
 		return err
 	}
 
@@ -70,13 +85,15 @@ func (c *Compactor) Run(ctx context.Context) error {
 	for _, tableName := range tables {
 		table, err := newTable(ctx, filepath.Join(c.cfg.WorkingDirectory, tableName), c.objectClient)
 		if err != nil {
-			level.Error(pkg_util.Logger).Log("msg", "failed to initialize table for compaction", "err", err)
+			status = statusFailure
+			level.Error(pkg_util.Logger).Log("msg", "failed to initialize table for compaction", "table", tableName, "err", err)
 			continue
 		}
 
 		err = table.compact()
 		if err != nil {
-			level.Error(pkg_util.Logger).Log("msg", "failed to compact files", "err", err)
+			status = statusFailure
+			level.Error(pkg_util.Logger).Log("msg", "failed to compact files", "table", tableName, "err", err)
 		}
 
 		// check if context was cancelled before going for next table.
diff --git a/pkg/storage/stores/shipper/compactor/metrics.go b/pkg/storage/stores/shipper/compactor/metrics.go
new file mode 100644
index 000000000000..25f9563e5612
--- /dev/null
+++ b/pkg/storage/stores/shipper/compactor/metrics.go
@@ -0,0 +1,33 @@
+package compactor
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	statusFailure = "failure"
+	statusSuccess = "success"
+)
+
+type metrics struct {
+	compactTablesOperationTotal *prometheus.CounterVec
+	compactTablesOperationDurationSeconds prometheus.Gauge
+}
+
+func newMetrics(r prometheus.Registerer) *metrics {
+	m := metrics{
+		compactTablesOperationTotal:           promauto.With(r).NewCounterVec(prometheus.CounterOpts{
+			Namespace: "loki_boltdb_shipper",
+			Name:      "compact_tables_operation_total",
+			Help:      "Total number of tables compaction done by status",
+		}, []string{"status"}),
+		compactTablesOperationDurationSeconds: promauto.With(r).NewGauge(prometheus.GaugeOpts{
+			Namespace: "loki_boltdb_shipper",
+			Name:      "compact_tables_operation_duration_seconds",
+			Help:      "Time (in seconds) spent in compacting all the tables",
+		}),
+	}
+
+	return &m
+}