Merge pull request #5 from mailsac/prom-metrics-744

.github/workflows/main.yml

@@ -7,7 +7,7 @@ jobs:
     runs-on: ubuntu-20.04
     strategy:
       matrix:
-        go: [ '1.17' ]
+        go: [ '1.18' ]
     name: Go v${{ matrix.go }}
     steps:
       - uses: actions/checkout@v2

.github/workflows/release.yml

@@ -16,7 +16,7 @@ jobs:
       - name: Setup go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.17
+          go-version: 1.18
       - name: build artifacts
         run: make build-all
       - name: Create Release

README.md

@@ -170,6 +170,28 @@ Entries are grouped in a `namespace`.
 
 See `server/server_test.go` for examples.
 
+## Prometheus metrics
+
+Basic garbage collection metrics are exposed when using the server flag `--prom=0.0.0.0:9090` flag (you can use a custom host and port).
+
+```text
+# HELP dracula_key_sum_in_gc_namespaces Count of key values in last garbed collected namespace valid keys
+# TYPE dracula_key_sum_in_gc_namespaces gauge
+dracula_key_sum_in_gc_namespaces 0
+# HELP dracula_keys_in_gc_namespaces Count of unexpired keys in last garbage collected namespaces
+# TYPE dracula_keys_in_gc_namespaces gauge
+dracula_keys_in_gc_namespaces 0
+# HELP dracula_max_namespaces_denom Denominator/portion of namespaces to be garbage collected each cleanup run
+# TYPE dracula_max_namespaces_denom gauge
+dracula_max_namespaces_denom 3
+# HELP dracula_namespaces_count Number of top level key namespaces
+# TYPE dracula_namespaces_count gauge
+dracula_namespaces_count 3
+# HELP dracula_namespaces_gc_count Number of namespaces which had keys garbage collected during last cleanup run
+# TYPE dracula_namespaces_gc_count gauge
+dracula_namespaces_gc_count 1
+```
+
 ## High Availability / Failover
 
 Rudimentary and experimental HA is possible via replication by using the `-p` peers list and `-i` self `IP:host` pair flags such as:
@@ -219,7 +241,7 @@ See dependencies listed in go.mod for copyright notices and licenses.
 
 ----
 
-Copyright (c) 2021 Forking Software LLC
+Copyright (c) 2021-2023 Forking Software LLC
 
 | Project           | License SPDX Identifier |
 |-------------------|-------------------------|

cmd/server/main.go

@@ -18,6 +18,7 @@ var (
 	peers           = flag.String("c", "", "Enable cluster replication. Peers must be comma-separated ip:port like `192.168.0.1:3509,192.168.0.2:3555`.")
 	verbose         = flag.Bool("v", false, "Verbose logging")
 	printVersion    = flag.Bool("version", false, "Print version")
+	promHostPort    = flag.String("prom", "", "Enable prometheus metrics. May cause pauses. Example: '0.0.0.0:9090'")
 )
 
 // Version should be replaced at build time
@@ -60,12 +61,19 @@ func main() {
 	}
 	err := s.Listen(*port)
 	if err != nil {
-		fmt.Println(err)
+		fmt.Println("Dracula startup listen error", err)
 		os.Exit(1)
 	}
 	if *verbose {
 		fmt.Println("will expire keys after", *expireAfterSecs, "seconds")
 	}
+	if *promHostPort != "" {
+		err = s.StoreMetrics.ListenAndServe(*promHostPort)
+		if err != nil {
+			fmt.Println("Prometheus startup listen error", *promHostPort, err)
+			os.Exit(1)
+		}
+	}
 
 	var wg sync.WaitGroup
 	wg.Add(1)

go.mod

@@ -5,5 +5,6 @@ go 1.16
 require (
 	github.com/OneOfOne/xxhash v1.2.8
 	github.com/emirpasic/gods v1.12.0
-	github.com/stretchr/testify v1.7.0
+	github.com/prometheus/client_golang v1.15.0
+	github.com/stretchr/testify v1.8.0
 )

server/server.go

@@ -26,6 +26,7 @@ var (
 
 type Server struct {
 	store             *store.Store
+	StoreMetrics      *store.Metrics
 	conn              *net.UDPConn
 	disposed          bool
 	preSharedKey      []byte
@@ -76,15 +77,17 @@ func NewServer(expireAfterSecs int64, preSharedKey string) *Server {
 		panic(ErrExpiryTooSmall)
 	}
 	psk := []byte(preSharedKey)
-	s := &Server{
-		store:             store.NewStore(expireAfterSecs),
+	st := store.NewStore(expireAfterSecs)
+	serv := &Server{
+		store:             st,
+		StoreMetrics:      st.LastMetrics,
 		preSharedKey:      psk,
 		expireAfterSecs:   expireAfterSecs,
 		messageProcessing: make(chan *rawMessage, runtime.NumCPU()),
 		log:               log.New(os.Stdout, "", 0),
 	}
-	s.DebugDisable()
-	return s
+	serv.DebugDisable()
+	return serv
 }
 
 func (s *Server) DebugEnable(prefix string) {

store/store.go

@@ -3,14 +3,40 @@ package store
 import (
 	"github.com/emirpasic/gods/maps/hashmap"
 	"github.com/mailsac/dracula/store/tree"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"net/http"
 	"sync"
 	"time"
 )
 
 var runDuration = time.Second * 15
 
-// denominator of how many keys to garbage collect max on a run. if 3 then 1/3 or `<total keys>/3`
-const maxNamespacesDenom = 3
+// denominator of how many namespaces to garbage collect max on a run. if 3 then 1/3 or `<total keys>/3`
+const maxNamespacesDenom = 2
+
+type Metrics struct {
+	registry                          *prometheus.Registry
+	maxNamespacesDenom                prometheus.Gauge
+	namespacesTotalCount              prometheus.Gauge
+	namespacesGarbageCollected        prometheus.Gauge
+	keysRemainingInGCNamespaces       prometheus.Gauge
+	countTotalRemainingInGCNamespaces prometheus.Gauge
+	gcPauseTime                       prometheus.Gauge
+}
+
+func (m *Metrics) ListenAndServe(promHostPort string) error {
+	http.Handle(
+		"/metrics", promhttp.HandlerFor(
+			m.registry,
+			promhttp.HandlerOpts{
+				EnableOpenMetrics: true,
+			}),
+	)
+	// To test: curl -H 'Accept: application/openmetrics-text' localhost:8080/metrics
+	err := http.ListenAndServe(promHostPort, nil)
+	return err
+}
 
 // Store provides a way to store entries and count them based on namespaces.
 // Old entries are garbage collected in a way that attempts to not block for too long.
@@ -19,14 +45,52 @@ type Store struct {
 	namespaces            *hashmap.Map
 	expireAfterSecs       int64
 	cleanupServiceEnabled bool
+	LastMetrics           *Metrics
 }
 
 func NewStore(expireAfterSecs int64) *Store {
+	registry := prometheus.NewRegistry()
+	maxNamespacesDenomGauge := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_max_namespaces_denom",
+		Help: "Denominator/portion of namespaces to be garbage collected each cleanup run",
+	})
+	namespacesTotalCount := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_namespaces_count",
+		Help: "Number of top level key namespaces",
+	})
+	namespacesGarbageCollected := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_namespaces_gc_count",
+		Help: "Number of namespaces which had keys garbage collected during last cleanup run",
+	})
+	keysRemainingInGCNamespaces := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_keys_in_gc_namespaces",
+		Help: "Count of unexpired keys in last garbage collected namespaces",
+	})
+	countTotalRemainingInGCNamespaces := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_key_sum_in_gc_namespaces",
+		Help: "Count of key values in last garbed collected namespace valid keys",
+	})
+	gcPauseTime := prometheus.NewGauge(prometheus.GaugeOpts{
+		Name: "dracula_gc_pause_millis",
+		Help: "How long last garbage collection took in milliseconds",
+	})
+	registry.MustRegister(maxNamespacesDenomGauge, namespacesTotalCount, namespacesGarbageCollected, keysRemainingInGCNamespaces, countTotalRemainingInGCNamespaces, gcPauseTime)
+
 	s := &Store{
 		expireAfterSecs: expireAfterSecs,
 		namespaces:      hashmap.New(),
+		LastMetrics: &Metrics{
+			registry:                          registry,
+			maxNamespacesDenom:                maxNamespacesDenomGauge,
+			namespacesTotalCount:              namespacesTotalCount,
+			namespacesGarbageCollected:        namespacesGarbageCollected,
+			keysRemainingInGCNamespaces:       keysRemainingInGCNamespaces,
+			countTotalRemainingInGCNamespaces: countTotalRemainingInGCNamespaces,
+			gcPauseTime:                       gcPauseTime,
+		},
 	}
 	s.cleanupServiceEnabled = true
+	s.LastMetrics.maxNamespacesDenom.Set(maxNamespacesDenom)
 
 	go s.runCleanup()
 
@@ -47,18 +111,20 @@ func (s *Store) runCleanup() {
 		return
 	}
 
+	start := time.Now()
+
 	defer time.AfterFunc(runDuration, s.runCleanup)
 
 	s.Lock()
 	keys := s.namespaces.Keys() // they are randomly ordered
 	s.Unlock()
 
 	hashSize := len(keys)
-	if hashSize < 3 {
-		return
-	}
 	maxNamespaces := hashSize / maxNamespacesDenom
 
+	s.LastMetrics.namespacesTotalCount.Set(float64(hashSize))
+	s.LastMetrics.namespacesGarbageCollected.Set(float64(maxNamespaces))
+
 	subtrees := make(map[string]*tree.Tree)
 
 	s.Lock()
@@ -80,18 +146,26 @@ func (s *Store) runCleanup() {
 	s.Unlock()
 
 	var subtreeKeys []string
+	var knownKeysCount int
+	var subtreeKeyTrackCount int
+	var tally int
 	for ns, subtree := range subtrees {
 		// Keys will cleanup every empty entry key
-		subtreeKeys, _ = subtree.Keys()
+		subtreeKeys, subtreeKeyTrackCount = subtree.Keys()
 		if len(subtreeKeys) == 0 {
 			// an empty subtree can be removed from the top level namespaces
 			s.Lock()
 			s.namespaces.Remove(ns)
 			s.Unlock()
 			continue
 		}
+		knownKeysCount += len(subtreeKeys)
+		tally += subtreeKeyTrackCount
 	}
 
+	s.LastMetrics.keysRemainingInGCNamespaces.Set(float64(knownKeysCount))
+	s.LastMetrics.countTotalRemainingInGCNamespaces.Set(float64(tally))
+	s.LastMetrics.gcPauseTime.Set(float64(time.Since(start).Milliseconds()))
 }
 
 func (s *Store) Put(ns, entryKey string) {