feat: add cronjob metrics (#215)

* wip: add cron metrics

* refactor: change metrics

* refactor: restore backward compatibility

* refactor: renaming

* doc: add comments

* doc: add comments

* fix: gorm tests

* test: add ProvideCronjobMetrics

* test: stub

Author: Reasno

cronopts/example_metrics_test.go

@@ -0,0 +1,42 @@
+package cronopts_test
+
+import (
+	"context"
+	"fmt"
+	"github.com/DoNewsCode/core"
+	"github.com/DoNewsCode/core/cronopts"
+	"github.com/DoNewsCode/core/observability"
+	"github.com/robfig/cron/v3"
+	"math/rand"
+	"time"
+)
+
+type CronModule struct {
+	metrics *cronopts.CronJobMetrics
+}
+
+func NewCronModule(metrics *cronopts.CronJobMetrics) CronModule {
+	return CronModule{metrics: metrics.Module("test_module")}
+}
+
+func (c CronModule) ProvideCron(crontab *cron.Cron) {
+	// Create a new cron job, and measure its execution durations.
+	crontab.AddJob("* * * * *", c.metrics.Job("test_job").Measure(cron.FuncJob(func() {
+		fmt.Println("running")
+		// For 50% chance, the job may fail. Report it to metrics collector.
+		if rand.Float64() > 0.5 {
+			c.metrics.Fail()
+		}
+	})))
+}
+
+func Example_cronJobMetrics() {
+	c := core.Default()
+	c.Provide(observability.Providers())
+	c.AddModuleFunc(NewCronModule)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 1500*time.Millisecond)
+	defer cancel()
+
+	c.Serve(ctx)
+}

cronopts/metrics.go

@@ -0,0 +1,67 @@
+package cronopts
+
+import (
+	"time"
+
+	"github.com/go-kit/kit/metrics"
+	"github.com/robfig/cron/v3"
+)
+
+// CronJobMetrics collects metrics for cron jobs.
+type CronJobMetrics struct {
+	cronJobDurationSeconds metrics.Histogram
+	cronJobFailCount       metrics.Counter
+
+	// labels that has been set
+	module string
+	job    string
+}
+
+// NewCronJobMetrics constructs a new *CronJobMetrics, setting default labels to "unknown".
+func NewCronJobMetrics(histogram metrics.Histogram, counter metrics.Counter) *CronJobMetrics {
+	return &CronJobMetrics{
+		cronJobDurationSeconds: histogram,
+		cronJobFailCount:       counter,
+		module:                 "unknown",
+		job:                    "unknown",
+	}
+}
+
+// Module specifies the module label for CronJobMetrics.
+func (c *CronJobMetrics) Module(module string) *CronJobMetrics {
+	return &CronJobMetrics{
+		cronJobDurationSeconds: c.cronJobDurationSeconds,
+		cronJobFailCount:       c.cronJobFailCount,
+		module:                 module,
+		job:                    c.job,
+	}
+}
+
+// Job specifies the job label for CronJobMetrics.
+func (c *CronJobMetrics) Job(job string) *CronJobMetrics {
+	return &CronJobMetrics{
+		cronJobDurationSeconds: c.cronJobDurationSeconds,
+		cronJobFailCount:       c.cronJobFailCount,
+		module:                 c.module,
+		job:                    job,
+	}
+}
+
+// Fail marks the job as failed.
+func (c *CronJobMetrics) Fail() {
+	c.cronJobFailCount.With("module", c.module, "job", c.job).Add(1)
+}
+
+// Measure wraps the given job and records the duration and success.
+func (c *CronJobMetrics) Measure(job cron.Job) cron.Job {
+	return cron.FuncJob(func() {
+		start := time.Now()
+		defer c.cronJobDurationSeconds.With("module", c.module, "job", c.job).Observe(time.Since(start).Seconds())
+		job.Run()
+	})
+}
+
+// Measure returns a job wrapper that wraps the given job and records the duration and success.
+func Measure(c *CronJobMetrics) cron.JobWrapper {
+	return c.Measure
+}

cronopts/metrics_test.go

@@ -0,0 +1,25 @@
+package cronopts
+
+import (
+	"testing"
+	"time"
+
+	"github.com/DoNewsCode/core/internal/stub"
+	"github.com/robfig/cron/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMeasure(t *testing.T) {
+	histogram := &stub.Histogram{}
+	counter := &stub.Counter{}
+	metrics := NewCronJobMetrics(histogram, counter)
+	metrics = metrics.Module("x").Job("y")
+	Measure(metrics)(cron.FuncJob(func() {
+		time.Sleep(time.Millisecond)
+	})).Run()
+	assert.ElementsMatch(t, histogram.LabelValues, []string{"module", "x", "job", "y"})
+	assert.True(t, histogram.ObservedValue > 0)
+	metrics.Fail()
+	assert.ElementsMatch(t, counter.LabelValues, []string{"module", "x", "job", "y"})
+	assert.True(t, counter.CounterValue == 1)
+}

internal/stub/metrics.go

@@ -0,0 +1,59 @@
+package stub
+
+import "github.com/go-kit/kit/metrics"
+
+// Histogram is a stub implementation of the go-kit metrics.Histogram interface.
+type Histogram struct {
+	LabelValues   []string
+	ObservedValue float64
+}
+
+// With returns a new Histogram with the given label values.
+func (h *Histogram) With(labelValues ...string) metrics.Histogram {
+	h.LabelValues = labelValues
+	return h
+}
+
+// Observe records the given value.
+func (h *Histogram) Observe(value float64) {
+	h.ObservedValue = value
+}
+
+// Gauge is a stub implementation of the go-kit metrics.Gauge interface.
+type Gauge struct {
+	LabelValues []string
+	GaugeValue  float64
+}
+
+// With returns a new Gauge with the given label values.
+func (g *Gauge) With(labelValues ...string) metrics.Gauge {
+	g.LabelValues = labelValues
+	return g
+}
+
+// Set sets the gauge value.
+func (g *Gauge) Set(value float64) {
+	g.GaugeValue = value
+}
+
+// Add adds the given value to the gauge.
+func (g *Gauge) Add(delta float64) {
+	g.GaugeValue = g.GaugeValue + delta
+}
+
+// Counter is a stub implementation of the go-kit metrics.Counter interface.
+type Counter struct {
+	LabelValues  []string
+	CounterValue float64
+}
+
+// With returns a new Counter with the given label values.
+func (c *Counter) With(labelValues ...string) metrics.Counter {
+	c.LabelValues = labelValues
+	return c
+}
+
+// Add adds the given value to the counter.
+func (c *Counter) Add(delta float64) {
+	c.CounterValue = c.CounterValue + delta
+}

internal/stub/metrics_test.go

@@ -0,0 +1,26 @@
+package stub
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestMetrics(t *testing.T) {
+	h := &Histogram{}
+	g := &Gauge{}
+	c := &Counter{}
+
+	h.With("foo", "bar").Observe(1.0)
+	assert.ElementsMatch(t, h.LabelValues, []string{"foo", "bar"})
+	assert.Equal(t, h.ObservedValue, 1.0)
+
+	g.With("foo", "bar").Set(1.0)
+	g.Add(1)
+	assert.Equal(t, g.GaugeValue, 2.0)
+	assert.ElementsMatch(t, g.LabelValues, []string{"foo", "bar"})
+
+	c.With("foo", "bar").Add(1)
+	c.Add(1)
+	assert.ElementsMatch(t, c.LabelValues, []string{"foo", "bar"})
+	assert.Equal(t, c.CounterValue, 2.0)
+}

observability/metrics.go

@@ -1,6 +1,7 @@
 package observability
 
 import (
+	"github.com/DoNewsCode/core/cronopts"
 	"github.com/DoNewsCode/core/di"
 	"github.com/DoNewsCode/core/otgorm"
 	"github.com/DoNewsCode/core/otkafka"
@@ -53,6 +54,34 @@ func ProvideGRPCRequestDurationSeconds(in MetricsIn) *srvgrpc.RequestDurationSec
 	return srvgrpc.NewRequestDurationSeconds(prometheus.NewHistogram(grpc))
 }
 
+// ProvideCronJobMetrics returns a *cronopts.CronJobMetrics that is designed to
+// measure cron job metrics. The returned metrics can be used like this:
+//  metrics := cronopts.NewCronJobMetrics(...)
+//  job := cron.NewChain(
+//  	cron.Recover(logger),
+//  	cronopts.Measure(metrics),
+//	).Then(job)
+func ProvideCronJobMetrics(in MetricsIn) *cronopts.CronJobMetrics {
+	histogram := stdprometheus.NewHistogramVec(stdprometheus.HistogramOpts{
+		Name: "cronjob_duration_seconds",
+		Help: "Total time spent running cron jobs.",
+	}, []string{"module", "job"})
+
+	counter := stdprometheus.NewCounterVec(stdprometheus.CounterOpts{
+		Name: "cronjob_failures_total",
+		Help: "Total number of cron jobs that failed.",
+	}, []string{"module", "job"})
+
+	if in.Registerer == nil {
+		in.Registerer = stdprometheus.DefaultRegisterer
+	}
+
+	in.Registerer.MustRegister(histogram)
+	in.Registerer.MustRegister(counter)
+
+	return cronopts.NewCronJobMetrics(prometheus.NewHistogram(histogram), prometheus.NewCounter(counter))
+}
+
 // ProvideGORMMetrics returns a *otgorm.Gauges that measures the connection info
 // in databases. It is meant to be consumed by the otgorm.Providers.
 func ProvideGORMMetrics(in MetricsIn) *otgorm.Gauges {

observability/observability.go

@@ -28,6 +28,7 @@ func Providers() di.Deps {
 		ProvideRedisMetrics,
 		ProvideKafkaReaderMetrics,
 		ProvideKafkaWriterMetrics,
+		ProvideCronJobMetrics,
 		provideConfig,
 	}
 }

observability/observability_test.go

@@ -1,6 +1,7 @@
 package observability
 
 import (
+	"github.com/DoNewsCode/core/cronopts"
 	"os"
 	"strings"
 	"testing"
@@ -78,14 +79,16 @@ func TestProvideRedisMetrics(t *testing.T) {
 	c.Provide(otredis.Providers())
 	c.Invoke(func(cli redis.UniversalClient, g *otredis.Gauges) {
 		stats := cli.PoolStats()
-		withValues := []string{"dbname", "default"}
-
-		g.Hits.With(withValues...).Set(float64(stats.Hits))
-		g.Misses.With(withValues...).Set(float64(stats.Misses))
-		g.Timeouts.With(withValues...).Set(float64(stats.Timeouts))
-		g.TotalConns.With(withValues...).Set(float64(stats.TotalConns))
-		g.IdleConns.With(withValues...).Set(float64(stats.IdleConns))
-		g.StaleConns.With(withValues...).Set(float64(stats.StaleConns))
+		g.Observe(stats)
+	})
+}
+
+func TestProvideCronjobMetrics(t *testing.T) {
+	c := core.New()
+	c.ProvideEssentials()
+	c.Provide(Providers())
+	c.Invoke(func(metrics *cronopts.CronJobMetrics) {
+		metrics.Fail()
 	})
 }
 

otgorm/gorm_metrics.go

@@ -20,58 +20,48 @@ type Gauges struct {
 	inUse metrics.Gauge
 	open  metrics.Gauge
 
-	dbName bool
-	driver bool
+	dbName string
+	driver string
 }
 
 // NewGauges returns a new Gauges.
 func NewGauges(idle, inUse, open metrics.Gauge) *Gauges {
 	return &Gauges{
-		idle:  idle,
-		inUse: inUse,
-		open:  open,
+		idle:   idle,
+		inUse:  inUse,
+		open:   open,
+		dbName: "unknown",
+		driver: "default",
 	}
 }
 
 // DBName sets the dbname label of metrics.
 func (g *Gauges) DBName(dbName string) *Gauges {
-	withValues := []string{"dbname", dbName}
 	return &Gauges{
-		idle:   g.idle.With(withValues...),
-		inUse:  g.inUse.With(withValues...),
-		open:   g.open.With(withValues...),
-		dbName: true,
+		idle:   g.idle,
+		inUse:  g.inUse,
+		open:   g.open,
+		dbName: dbName,
 		driver: g.driver,
 	}
 }
 
 // Driver sets the driver label of metrics.
 func (g *Gauges) Driver(driver string) *Gauges {
-	withValues := []string{"driver", driver}
 	return &Gauges{
-		idle:   g.idle.With(withValues...),
-		inUse:  g.inUse.With(withValues...),
-		open:   g.open.With(withValues...),
+		idle:   g.idle,
+		inUse:  g.inUse,
+		open:   g.open,
 		dbName: g.dbName,
-		driver: true,
+		driver: driver,
 	}
 }
 
 // Observe records the DBStats collected. It should be called periodically.
 func (g *Gauges) Observe(stats sql.DBStats) {
-	if !g.dbName {
-		g.idle = g.idle.With("dbname", "")
-		g.inUse = g.inUse.With("dbname", "")
-		g.open = g.open.With("dbname", "")
-	}
-	if !g.driver {
-		g.idle = g.idle.With("driver", "")
-		g.inUse = g.inUse.With("driver", "")
-		g.open = g.open.With("driver", "")
-	}
-	g.idle.Set(float64(stats.Idle))
-	g.inUse.Set(float64(stats.InUse))
-	g.open.Set(float64(stats.OpenConnections))
+	g.idle.With("dbname", g.dbName, "driver", g.driver).Set(float64(stats.Idle))
+	g.inUse.With("dbname", g.dbName, "driver", g.driver).Set(float64(stats.InUse))
+	g.open.With("dbname", g.dbName, "driver", g.driver).Set(float64(stats.OpenConnections))
 }
 
 // newCollector creates a new database wrapper containing the name of the database,