OM-92 & OM-94 (#92)

added support for index-pressure
added support for stats and configs at XDR-DC and XDR-DC-Namespace level
moved prometheus push logic to a common function

common.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/gobwas/glob"
 	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
 
 	goversion "github.com/hashicorp/go-version"
 )
@@ -37,17 +38,25 @@ const (
 	CTX_LATENCIES  ContextType = "latencies"
 )
 
-const STORAGE_ENGINE = "storage-engine_"
-
 // below constant represent the labels we send along with metrics to Prometheus or something
-const METRIC_LABEL_CLUSTER_NAME = "cluster_name"
-const METRIC_LABEL_SERVICE = "service"
-const METRIC_LABEL_NS = "ns"
-const METRIC_LABEL_SET = "set"
-const METRIC_LABEL_LE = "le"
-const METRIC_LABEL_DC_NAME = "dc"
-const METRIC_LABEL_SINDEX = "sindex"
-const METRIC_LABEL_USER = "user"
+const (
+	METRIC_LABEL_CLUSTER_NAME = "cluster_name"
+	METRIC_LABEL_SERVICE      = "service"
+	METRIC_LABEL_NS           = "ns"
+	METRIC_LABEL_SET          = "set"
+	METRIC_LABEL_LE           = "le"
+	METRIC_LABEL_DC_NAME      = "dc"
+	METRIC_LABEL_INDEX        = "index"
+	METRIC_LABEL_SINDEX       = "sindex"
+	METRIC_LABEL_USER         = "user"
+)
+
+// constants used to identify type of metrics
+const (
+	STORAGE_ENGINE = "storage-engine"
+	INDEX_TYPE     = "index-type"
+	SINDEX_TYPE    = "sindex-type"
+)
 
 func makeMetric(namespace, name string, t metricType, constLabels map[string]string, labels ...string) promMetric {
 	promDesc := prometheus.NewDesc(
@@ -424,3 +433,21 @@ func getMetricType(pContext ContextType, pRawMetricName string) metricType {
 
 	return mtCounter
 }
+
+// This is a common utility, used by all the watchers to push metric to prometheus
+func pushToPrometheus(asMetric AerospikeStat, pv float64, labels []string, labelValues []string,
+	ch chan<- prometheus.Metric) {
+
+	if asMetric.isAllowed {
+		// handle any panic from prometheus, this may occur when prom encounters a config/stat with special characters
+		defer func() {
+			if r := recover(); r != nil {
+				log.Tracef("%s recovered from panic while handling stat %s", string(asMetric.context), asMetric.name)
+			}
+		}()
+
+		desc, valueType := asMetric.makePromMetric(labels...)
+		ch <- prometheus.MustNewConstMetric(desc, valueType, pv, labelValues...)
+
+	}
+}

gauge_stats.go

@@ -34,7 +34,6 @@ func initGaugeStats(pGaugeStatsFile string, pGaugeStats *GaugeStats) {
 
 	log.Infof("Loading Gauge Stats file %s", pGaugeStatsFile)
 
-	// fmt.Println("Loading Gauge Stats file ", configFile)
 	blob, err := os.ReadFile(pGaugeStatsFile)
 	if err != nil {
 		log.Fatalln(err)

gauge_stats_list.toml

@@ -55,9 +55,7 @@ sindex_gauge_stats = [
 #
 
 node_gauge_stats = [
-	"batch_index_created_buffers",
-	"batch_index_huge_buffers",
-	"batch_index_proto_compression_ratio",
+	"batch_index_proto_compression_ratio", 
 	"batch_index_proto_uncompressed_pct",
 	"batch_index_queue",
 	"batch_index_unused_buffers",
@@ -69,9 +67,6 @@ node_gauge_stats = [
 	"cluster_max_compatibility_id",	
 	"cluster_min_compatibility_id",
 	"cluster_size",
-	"dlog_free_pct",
-	"dlog_logged",	
-	"dlog_used_objects",
 	"fabric_bulk_recv_rate",
 	"fabric_bulk_send_rate",
 	"fabric_connections",
@@ -111,8 +106,9 @@ node_gauge_stats = [
 	"tree_gc_queue",
 	"tsvc_queue",
 	#
-	# TODO: Do we need the below XDR Stats
-	#	
+	# 4.x XDR stats
+	"dlog_free_pct",
+	"dlog_used_objects",
 	"xdr_active_failed_node_sessions",
 	"xdr_active_link_down_sessions",
 	"xdr_global_lastshiptime",
@@ -124,8 +120,6 @@ node_gauge_stats = [
 	"xdr_read_respq_used",
 	"xdr_read_txnq_used_pct",
 	"xdr_read_txnq_used",
-	"xdr_relogged_incoming",
-	"xdr_relogged_outgoing",
 	"xdr_ship_compression_avg_pct",
 	"xdr_ship_inflight_objects",
 	"xdr_ship_latency_avg",
@@ -138,7 +132,6 @@ node_gauge_stats = [
 # Namespace: below section define all Namespace stats which are treated as Gauges 
 #
 namespace_gauge_stats =[
-
 	"appeals_rx_active",
 	"appeals_tx_active",
 	"appeals_tx_remaining",
@@ -171,11 +164,7 @@ namespace_gauge_stats =[
 	"memory_used_index_bytes",
 	"memory_used_set_index_bytes",
 	"memory_used_sindex_bytes",
-	"migrate_record_receives",
-	"migrate_record_retransmits",
-	"migrate_records_skipped",
-	"migrate_records_transmitted",
-	"migrate_rx_instances",
+	"migrate_rx_instances",   
 	"migrate_rx_partitions_active",
 	"migrate_rx_partitions_initial",
 	"migrate_rx_partitions_remaining",
@@ -214,22 +203,16 @@ namespace_gauge_stats =[
 	"scan_proto_uncompressed_pct",
 	"shadow_write_q",
 	"stop_writes",
-
 	"storage-engine.device.defrag_q",
-	"storage-engine.device.defrag_reads",
-	"storage-engine.device.defrag_writes",
 	"storage-engine.device.free_wblocks",
 	"storage-engine.device.shadow_write_q",
 	"storage-engine.device.used_bytes",
 	"storage-engine.device.write_q",
 	"storage-engine.file.defrag_q",
-	"storage-engine.file.defrag_reads",
-	"storage-engine.file.defrag_writes",
 	"storage-engine.file.free_wblocks",
 	"storage-engine.file.shadow_write_q",
 	"storage-engine.file.used_bytes",
-	"storage-engine.file.write_q",
-
+	"storage-engine.file.write_q",	
 	"tombstones",
 	"truncate_lut",
 	"unavailable_partitions",

gauge_stats_test.go

@@ -2,22 +2,16 @@ package main
 
 import (
 	"fmt"
-	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestGetGaugesNotEmpty(t *testing.T) {
-
-	// this is to force-reload the config in the NamespaceWatcher, this is a check on this param in NamespaceWatcher implementation
-	os.Setenv(TESTCASE_MODE, TESTCASE_MODE_TRUE)
-
 	fmt.Println("initializing GaugeMetrics ... TestGetGaugesNotEmpty")
 
 	// Initialize and validate Gauge config
 	gaugeList := new(GaugeStats)
-
 	initGaugeStats(METRICS_CONFIG_FILE, gaugeList)
 
 	nslist := gaugeList.NamespaceStats
@@ -27,10 +21,6 @@ func TestGetGaugesNotEmpty(t *testing.T) {
 }
 
 func TestGetGaugesCounts(t *testing.T) {
-
-	// this is to force-reload the config in the NamespaceWatcher, this is a check on this param in NamespaceWatcher implementation
-	os.Setenv(TESTCASE_MODE, TESTCASE_MODE_TRUE)
-
 	fmt.Println("initializing GaugeMetrics ... TestGetGaugesCounts")
 
 	// Initialize and validate Gauge config
@@ -39,10 +29,10 @@ func TestGetGaugesCounts(t *testing.T) {
 	initGaugeStats(METRICS_CONFIG_FILE, gaugeList)
 
 	glist := gaugeList.NamespaceStats
-	assert.Equal(t, len(glist), 96)
+	assert.Equal(t, len(glist), 88)
 
 	glist = gaugeList.NodeStats
-	assert.Equal(t, len(glist), 74)
+	assert.Equal(t, len(glist), 69)
 
 	glist = gaugeList.SetsStats
 	assert.Equal(t, len(glist), 7)
@@ -56,9 +46,7 @@ func TestGetGaugesCounts(t *testing.T) {
 }
 
 func TestIsAGaugeTrue(t *testing.T) {
-
-	// this is to force-reload the config in the NamespaceWatcher, this is a check on this param in NamespaceWatcher implementation
-	os.Setenv(TESTCASE_MODE, TESTCASE_MODE_TRUE)
+	fmt.Println("initializing GaugeMetrics ... TestIsAGaugeTrue")
 
 	// Initialize and validate Gauge config
 	gaugeList := new(GaugeStats)
@@ -86,9 +74,6 @@ func TestIsAGaugeTrue(t *testing.T) {
 
 func TestNoGaugeExists(t *testing.T) {
 
-	// this is to force-reload the config in the NamespaceWatcher, this is a check on this param in NamespaceWatcher implementation
-	os.Setenv(TESTCASE_MODE, TESTCASE_MODE_TRUE)
-
 	fmt.Println("initializing GaugeMetrics ... TestNoGaugeExists")
 
 	// Initialize and validate Gauge config

helper_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -13,13 +14,9 @@ var LABELS_APE_TOML = "tests/labels_ape.toml"
 var NS_ALLOWLIST_APE_TOML = "tests/ns_allowlist_ape.toml"
 var NS_BLOCKLIST_APE_TOML = "tests/ns_blocklist_ape.toml"
 
-var TESTCASE_MODE = "TESTCASE_MODE"
-var TESTCASE_MODE_TRUE = "true"
-var TESTCASE_MODE_FALSE = "false"
-
 var METRICS_CONFIG_FILE = "gauge_stats_list.toml"
 
-// var g_ns_metric_allow_list = []string{"aerospike_namespace_master_objects", "aerospike_namespace_memory_used_bytes"}
+var MOCK_TEST_DATA_FILE = "tests/mock_test_data.txt"
 
 func extractNamespaceFromLabel(label string) string {
 	// [name:"cluster_name" value:""  name:"ns" value:"bar"  name:"service" value:"" ]
@@ -30,6 +27,14 @@ func extractNamespaceFromLabel(label string) string {
 	return nsFromLabel
 }
 
+func stringifyLabel(label string) string {
+	labelReplacerFunc := strings.NewReplacer(".", "_", "-", "_", " ", "_", "[", "_", "]", "_", "\"", "_", ":", "_", "/", "_")
+	hypenReplacerFunc := strings.NewReplacer("_", "")
+
+	// return labelReplacerFunc.Replace(label)
+	return hypenReplacerFunc.Replace(labelReplacerFunc.Replace(label))
+}
+
 func extractLabelNameValueFromFullLabel(fullLabel string, reqName string) string {
 
 	// Example Given Original: [name:"cluster_name" value:"null"  name:"service" value:"172.17.0.3:3000" ]
@@ -256,3 +261,35 @@ func extractOperationFromMetric(metricName string) string {
 
 	return operationName
 }
+
+func copyConfigLabels() map[string]string {
+	cfgLabels := config.AeroProm.MetricLabels
+
+	returnLabelMap := make(map[string]string)
+	for key, value := range cfgLabels {
+		returnLabelMap[key] = value
+	}
+
+	return returnLabelMap
+}
+
+func createLabelByNames(labelsMap map[string]string) string {
+
+	arr_label_names := []string{}
+	createdLabelString := ""
+
+	for key := range labelsMap {
+		arr_label_names = append(arr_label_names, strings.TrimSpace(key))
+	}
+
+	// sort the keys
+	sort.Strings(arr_label_names)
+
+	for idx := range arr_label_names {
+		keyName := arr_label_names[idx]
+		value := labelsMap[keyName]
+		createdLabelString = createdLabelString + constructLabelElement(keyName, strings.TrimSpace(value))
+	}
+
+	return strings.TrimSpace(createdLabelString)
+}