Metrics package

Added metrics package which
provides tools useful for gathering
and exposing system metrics for external
monitoring tools.

Author: lukasz-zimnoch

pkg/metrics/gauge.go

@@ -0,0 +1,49 @@
+package metrics
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Gauge is a metric type that represents a single numerical value that can
+// arbitrarily go up and down.
+type Gauge struct {
+	name   string
+	labels map[string]string
+
+	value     float64
+	timestamp int64
+	mutex     sync.RWMutex
+}
+
+// Set allows setting the gauge to an arbitrary value.
+func (g *Gauge) Set(value float64) {
+	g.mutex.Lock()
+	defer g.mutex.Unlock()
+
+	g.value = value
+	g.timestamp = time.Now().UnixNano() / int64(time.Millisecond)
+}
+
+// Exposes the gauge in the text-based exposition format.
+func (g *Gauge) expose() string {
+	g.mutex.RLock()
+	defer g.mutex.RUnlock()
+
+	typeLine := fmt.Sprintf("# TYPE %v %v", g.name, "gauge")
+
+	labelsStrings := make([]string, 0)
+	for name, value := range g.labels {
+		labelsStrings = append(
+			labelsStrings,
+			fmt.Sprintf("%v=\"%v\"", name, value),
+		)
+	}
+	labels := strings.Join(labelsStrings, ",")
+
+	body := fmt.Sprintf("%v{%v} %v %v", g.name, labels, g.value, g.timestamp)
+
+	return fmt.Sprintf("%v\n%v", typeLine, body)
+}

pkg/metrics/gauge_test.go

@@ -0,0 +1,61 @@
+package metrics
+
+import (
+	"testing"
+)
+
+func TestGaugeSet(t *testing.T) {
+	gauge := &Gauge{
+		name:   "test_gauge",
+		labels: map[string]string{"label": "value"},
+	}
+
+	if gauge.value != 0 {
+		t.Fatal("incorrect gauge initial value")
+	}
+
+	if gauge.timestamp != 0 {
+		t.Fatal("incorrect gauge initial timestamp")
+	}
+
+	newGaugeValue := float64(500)
+
+	gauge.Set(newGaugeValue)
+
+	if gauge.value != newGaugeValue {
+		t.Fatalf(
+			"incorrect gauge value:\n"+
+				"expected: [%v]\n"+
+				"actual:   [%v]",
+			newGaugeValue,
+			gauge.value,
+		)
+	}
+
+	if gauge.timestamp == 0 {
+		t.Fatal("timestamp should be set")
+	}
+}
+
+func TestGaugeExpose(t *testing.T) {
+	gauge := &Gauge{
+		name:      "test_gauge",
+		labels:    map[string]string{"label": "value"},
+		value:     500,
+		timestamp: 1000,
+	}
+
+	actualText := gauge.expose()
+
+	expectedText := "# TYPE test_gauge gauge\ntest_gauge{label=\"value\"} 500 1000"
+
+	if actualText != expectedText {
+		t.Fatalf(
+			"incorrect gauge expose text:\n"+
+				"expected: [%v]\n"+
+				"actual:   [%v]",
+			expectedText,
+			actualText,
+		)
+	}
+}

pkg/metrics/observer.go

@@ -0,0 +1,36 @@
+package metrics
+
+import (
+	"context"
+	"time"
+)
+
+// Source defines a source of metric data.
+type Source func() float64
+
+// Sink defines a destination of collected metric data.
+type Sink interface {
+	Set(value float64)
+}
+
+// Observe triggers a cyclic metric observation goroutine.
+func Observe(
+	ctx context.Context,
+	source Source,
+	sink Sink,
+	tick time.Duration,
+) {
+	go func() {
+		ticker := time.NewTicker(tick)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				sink.Set(source())
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+}

pkg/metrics/observer_test.go

@@ -0,0 +1,38 @@
+package metrics
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestObserve(t *testing.T) {
+	gauge := &Gauge{}
+	ctx, _ := context.WithTimeout(context.Background(), 5*time.Millisecond)
+
+	Observe(
+		ctx,
+		func() float64 {
+			return 5000
+		},
+		gauge,
+		1*time.Millisecond,
+	)
+
+	<-ctx.Done()
+
+	expectedGaugeValue := float64(5000)
+	if gauge.value != expectedGaugeValue {
+		t.Fatalf(
+			"incorrect gauge value:\n"+
+				"expected: [%v]\n"+
+				"actual:   [%v]",
+			expectedGaugeValue,
+			gauge.value,
+		)
+	}
+
+	if gauge.timestamp == 0 {
+		t.Fatal("timestamp should be set")
+	}
+}

pkg/metrics/registry.go

@@ -0,0 +1,102 @@
+// Package `metrics` provides some tools useful for gathering and
+// exposing system metrics for external monitoring tools.
+//
+// Currently, this package is intended to use with Prometheus but
+// can be easily extended if needed. Also, not all Prometheus metric
+// types are implemented. The main motivation of creating a custom
+// package was a need to avoid usage of external unaudited dependencies.
+//
+// Following specifications was used as reference:
+// - https://prometheus.io/docs/instrumenting/writing_clientlibs/
+// - https://prometheus.io/docs/instrumenting/exposition_formats/
+package metrics
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/ipfs/go-log"
+)
+
+var logger = log.Logger("keep-metrics")
+
+type metric interface {
+	expose() string
+}
+
+// Registry performs all management of metrics. Specifically, it allows
+// to registering new metrics and exposing them through the metrics server.
+type Registry struct {
+	application string
+	identifier  string
+
+	metrics      map[string]metric
+	metricsMutex sync.RWMutex
+}
+
+// NewRegistry creates a new metrics registry.
+func NewRegistry(application, identifier string) *Registry {
+	return &Registry{
+		application: application,
+		identifier:  identifier,
+		metrics:     make(map[string]metric),
+	}
+}
+
+// EnableServer enables the metrics server on the given port. Data will
+// be exposed on `/metrics` path.
+func (r *Registry) EnableServer(port int) {
+	server := &http.Server{Addr: ":" + strconv.Itoa(port)}
+
+	http.HandleFunc("/metrics", func(response http.ResponseWriter, _ *http.Request) {
+		if _, err := io.WriteString(response, r.exposeMetrics()); err != nil {
+			logger.Errorf("could not write response: [%v]", err)
+		}
+	})
+
+	go func() {
+		if err := server.ListenAndServe(); err != http.ErrServerClosed {
+			logger.Errorf("metrics server error: [%v]", err)
+		}
+	}()
+}
+
+// Exposes all registered metrics in their text format.
+func (r *Registry) exposeMetrics() string {
+	r.metricsMutex.RLock()
+	defer r.metricsMutex.RUnlock()
+
+	metrics := make([]string, 0)
+	for _, metric := range r.metrics {
+		metrics = append(metrics, metric.expose())
+	}
+
+	return strings.Join(metrics, "\n\n")
+}
+
+// NewGauge creates and registers a new gauge metric which will be exposed
+// through the metrics server. In case a metric already exists, an error
+// will be returned.
+func (r *Registry) NewGauge(name string) (*Gauge, error) {
+	r.metricsMutex.Lock()
+	defer r.metricsMutex.Unlock()
+
+	if _, exists := r.metrics[name]; exists {
+		return nil, fmt.Errorf("gauge [%v] already exists", name)
+	}
+
+	gauge := &Gauge{
+		name: name,
+		labels: map[string]string{
+			"application": r.application,
+			"identifier":  r.identifier,
+		},
+	}
+
+	r.metrics[name] = gauge
+	return gauge, nil
+}

pkg/metrics/registry_test.go

@@ -0,0 +1,22 @@
+package metrics
+
+import (
+	"testing"
+)
+
+func TestRegistryNewGauge(t *testing.T) {
+	registry := NewRegistry("test-app", "test-id")
+
+	gauge, err := registry.NewGauge("test-gauge")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err = registry.NewGauge("test-gauge"); err == nil {
+		t.Fatalf("should fail when creating gauge with the same name")
+	}
+
+	if _, exists := registry.metrics[gauge.name]; !exists {
+		t.Fatalf("metric with name [%v] should exist in the registry", gauge.name)
+	}
+}