Rocksdb Metrics

cmd/kava/opendb/metrics.go

@@ -0,0 +1,156 @@
+//go:build rocksdb
+// +build rocksdb
+
+package opendb
+
+import (
+	"github.com/go-kit/kit/metrics"
+	"github.com/go-kit/kit/metrics/prometheus"
+	stdprometheus "github.com/prometheus/client_golang/prometheus"
+)
+
+type Metrics struct {
+	// Keys
+	NumberKeysWritten metrics.Gauge
+	NumberKeysRead    metrics.Gauge
+	NumberKeysUpdated metrics.Gauge
+	EstimateNumKeys   metrics.Gauge
+
+	// Files
+	NumberFileOpens  metrics.Gauge
+	NumberFileErrors metrics.Gauge
+
+	// Memory
+	BlockCacheUsage         metrics.Gauge
+	EstimateTableReadersMem metrics.Gauge
+	CurSizeAllMemTables     metrics.Gauge
+	BlockCachePinnedUsage   metrics.Gauge
+
+	// Cache
+	BlockCacheMiss        metrics.Gauge
+	BlockCacheHit         metrics.Gauge
+	BlockCacheAdd         metrics.Gauge
+	BlockCacheAddFailures metrics.Gauge
+}
+
+func (m *Metrics) send(props *properties, stats *stats) {
+	// Keys
+	m.NumberKeysWritten.Set(float64(stats.NumberKeysWritten))
+	m.NumberKeysRead.Set(float64(stats.NumberKeysRead))
+	m.NumberKeysUpdated.Set(float64(stats.NumberKeysUpdated))
+	m.EstimateNumKeys.Set(float64(props.EstimateNumKeys))
+
+	// Files
+	m.NumberFileOpens.Set(float64(stats.NumberFileOpens))
+	m.NumberFileErrors.Set(float64(stats.NumberFileErrors))
+
+	// Memory
+	m.BlockCacheUsage.Set(float64(props.BlockCacheUsage))
+	m.EstimateTableReadersMem.Set(float64(props.EstimateTableReadersMem))
+	m.CurSizeAllMemTables.Set(float64(props.CurSizeAllMemTables))
+	m.BlockCachePinnedUsage.Set(float64(props.BlockCachePinnedUsage))
+
+	// Cache
+	m.BlockCacheMiss.Set(float64(stats.BlockCacheMiss))
+	m.BlockCacheHit.Set(float64(stats.BlockCacheHit))
+	m.BlockCacheAdd.Set(float64(stats.BlockCacheAdd))
+	m.BlockCacheAddFailures.Set(float64(stats.BlockCacheAddFailures))
+}
+
+func newMetrics() *Metrics {
+	labels := make([]string, 0)
+
+	return &Metrics{
+		// Keys
+		NumberKeysWritten: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "key",
+			Name:      "number_keys_written",
+			Help:      "",
+		}, labels),
+		NumberKeysRead: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "key",
+			Name:      "number_keys_read",
+			Help:      "",
+		}, labels),
+		NumberKeysUpdated: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "key",
+			Name:      "number_keys_updated",
+			Help:      "",
+		}, labels),
+		EstimateNumKeys: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "key",
+			Name:      "estimate_num_keys",
+			Help:      "",
+		}, labels),
+
+		// Files
+		NumberFileOpens: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "file",
+			Name:      "number_file_opens",
+			Help:      "",
+		}, labels),
+		NumberFileErrors: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "file",
+			Name:      "number_file_errors",
+			Help:      "",
+		}, labels),
+
+		// Memory
+		BlockCacheUsage: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "memory",
+			Name:      "block_cache_usage",
+			Help:      "",
+		}, labels),
+		EstimateTableReadersMem: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "memory",
+			Name:      "estimate_table_readers_mem",
+			Help:      "",
+		}, labels),
+		CurSizeAllMemTables: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "memory",
+			Name:      "cur_size_all_mem_tables",
+			Help:      "",
+		}, labels),
+		BlockCachePinnedUsage: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "memory",
+			Name:      "block_cache_pinned_usage",
+			Help:      "",
+		}, labels),
+
+		// Cache
+		BlockCacheMiss: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "cache",
+			Name:      "block_cache_miss",
+			Help:      "",
+		}, labels),
+		BlockCacheHit: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "cache",
+			Name:      "block_cache_hit",
+			Help:      "",
+		}, labels),
+		BlockCacheAdd: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "cache",
+			Name:      "block_cache_add",
+			Help:      "",
+		}, labels),
+		BlockCacheAddFailures: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: "rocksdb",
+			Subsystem: "cache",
+			Name:      "block_cache_add_failures",
+			Help:      "",
+		}, labels),
+	}
+}

cmd/kava/opendb/opendb_rocksdb.go

@@ -23,10 +23,12 @@ package opendb
 import (
 	"errors"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"runtime"
 	"strings"
+	"time"
 
 	"github.com/cosmos/cosmos-sdk/server/types"
 	"github.com/linxGnu/grocksdb"
@@ -136,10 +138,31 @@ func newRocksDBWithOptions(name string, dir string, dbOpts, cfOpts *grocksdb.Opt
 		return nil, fmt.Errorf("failed to create db path: %w", err)
 	}
 
+	dbOpts.EnableStatistics()
+	//cfOpts.EnableStatistics()
 	db, _, err := grocksdb.OpenDbColumnFamilies(dbOpts, dbPath, []string{defaultColumnFamilyName}, []*grocksdb.Options{cfOpts})
 	if err != nil {
 		return nil, err
 	}
+
+	keyMetrics := newMetrics()
+
+	go func() {
+		ticker := time.NewTicker(time.Second * 15)
+		for {
+			select {
+			case <-ticker.C:
+				props, stats, err := getPropsAndStats(db)
+				if err != nil {
+					log.Fatal(err)
+					continue
+				}
+
+				keyMetrics.send(props, stats)
+			}
+		}
+	}()
+
 	ro := grocksdb.NewDefaultReadOptions()
 	wo := grocksdb.NewDefaultWriteOptions()
 	woSync := grocksdb.NewDefaultWriteOptions()
@@ -168,3 +191,24 @@ func newDefaultOptions() *grocksdb.Options {
 
 	return opts
 }
+
+func getPropsAndStats(db *grocksdb.DB) (*properties, *stats, error) {
+	propsLoader := newPropsLoader(db)
+	props, err := propsLoader.load()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	metricMap, err := parseSerializedStats(props.OptionsStatistics)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	statLoader := newStatLoader(metricMap)
+	stats, err := statLoader.load()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return props, stats, nil
+}

cmd/kava/opendb/props_loader.go

@@ -0,0 +1,73 @@
+//go:build rocksdb
+// +build rocksdb
+
+package opendb
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/linxGnu/grocksdb"
+	"github.com/pkg/errors"
+)
+
+type propsLoader struct {
+	db        *grocksdb.DB
+	errorMsgs []string
+}
+
+func newPropsLoader(db *grocksdb.DB) *propsLoader {
+	return &propsLoader{
+		db:        db,
+		errorMsgs: make([]string, 0),
+	}
+}
+
+func (l *propsLoader) load() (*properties, error) {
+	props := &properties{
+		BaseLevel:               l.getIntProperty("rocksdb.base-level"),
+		BlockCacheCapacity:      l.getIntProperty("rocksdb.block-cache-capacity"),
+		BlockCachePinnedUsage:   l.getIntProperty("rocksdb.block-cache-pinned-usage"),
+		BlockCacheUsage:         l.getIntProperty("rocksdb.block-cache-usage"),
+		CurSizeActiveMemTable:   l.getIntProperty("rocksdb.cur-size-active-mem-table"),
+		CurSizeAllMemTables:     l.getIntProperty("rocksdb.cur-size-all-mem-tables"),
+		EstimateLiveDataSize:    l.getIntProperty("rocksdb.estimate-live-data-size"),
+		EstimateNumKeys:         l.getIntProperty("rocksdb.estimate-num-keys"),
+		EstimateTableReadersMem: l.getIntProperty("rocksdb.estimate-table-readers-mem"),
+		LiveSSTFilesSize:        l.getIntProperty("rocksdb.live-sst-files-size"),
+		SizeAllMemTables:        l.getIntProperty("rocksdb.size-all-mem-tables"),
+		OptionsStatistics:       l.db.GetProperty("rocksdb.options-statistics"),
+	}
+
+	if len(l.errorMsgs) != 0 {
+		errorMsg := strings.Join(l.errorMsgs, ";")
+		return nil, errors.New(errorMsg)
+	}
+
+	return props, nil
+}
+
+func (l *propsLoader) getIntProperty(propName string) uint64 {
+	value, ok := l.db.GetIntProperty(propName)
+	if !ok {
+		l.errorMsgs = append(l.errorMsgs, fmt.Sprintf("can't get %v int property", propName))
+		return 0
+	}
+
+	return value
+}
+
+type properties struct {
+	BaseLevel               uint64
+	BlockCacheCapacity      uint64
+	BlockCachePinnedUsage   uint64
+	BlockCacheUsage         uint64
+	CurSizeActiveMemTable   uint64
+	CurSizeAllMemTables     uint64
+	EstimateLiveDataSize    uint64
+	EstimateNumKeys         uint64
+	EstimateTableReadersMem uint64
+	LiveSSTFilesSize        uint64
+	SizeAllMemTables        uint64
+	OptionsStatistics       string
+}

cmd/kava/opendb/stat_parser.go

@@ -0,0 +1,87 @@
+//go:build rocksdb
+// +build rocksdb
+
+package opendb
+
+import (
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+type stat struct {
+	name  string
+	props map[string]string
+}
+
+func parseSerializedStats(serializedStats string) (map[string]*stat, error) {
+	stats := make(map[string]*stat, 0)
+
+	serializedStatList := strings.Split(serializedStats, "\n")
+	serializedStatList = serializedStatList[:len(serializedStatList)-1]
+	for _, serializedStat := range serializedStatList {
+		stat, err := parseSerializedStat(serializedStat)
+		if err != nil {
+			return nil, err
+		}
+
+		stats[stat.name] = stat
+	}
+
+	return stats, nil
+}
+
+func parseSerializedStat(serializedStat string) (*stat, error) {
+	tokens := strings.Split(serializedStat, " ")
+	tokensNum := len(tokens)
+	if err := validateTokens(tokens); err != nil {
+		return nil, errors.Wrap(err, "tokens are invalid")
+	}
+
+	props := make(map[string]string)
+	for idx := 1; idx < tokensNum; idx += 3 {
+		key := tokens[idx]
+		sep := tokens[idx+1]
+		value := tokens[idx+2]
+
+		if err := validateStatProperty(key, value, sep); err != nil {
+			return nil, errors.Wrap(err, "invalid stat property")
+		}
+
+		props[key] = value
+	}
+
+	return &stat{
+		name:  tokens[0],
+		props: props,
+	}, nil
+}
+
+func validateTokens(tokens []string) error {
+	tokensNum := len(tokens)
+	if tokensNum < 4 {
+		return errors.Errorf("invalid number of tokens: %v, tokens: %v", tokensNum, tokens)
+	}
+	if (tokensNum-1)%3 != 0 {
+		return errors.Errorf("invalid number of tokens: %v, tokens: %v", tokensNum, tokens)
+	}
+	if tokens[0] == "" {
+		return errors.Errorf("stat name shouldn't be empty")
+	}
+
+	return nil
+}
+
+func validateStatProperty(key, value, sep string) error {
+	if key == "" {
+		return errors.Errorf("key shouldn't be empty")
+	}
+	if sep != ":" {
+		return errors.Errorf("separator should be :")
+	}
+	if value == "" {
+		return errors.Errorf("value shouldn't be empty")
+	}
+
+	return nil
+}