chore: Enhance to support metrics prefix for OpEx metric name (#42)

Signed-off-by: sarabala1979 <[email protected]>
numaproj · Aug 6, 2024 · 0e8e3a2 · 0e8e3a2
1 parent 7b68e65
commit 0e8e3a2
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 9 deletions.
diff --git a/prometheus-pusher/main.go b/prometheus-pusher/main.go
@@ -25,6 +25,7 @@ const (
 	METRICS_LABELS         = "METRICS_LABELS"
 	METRICS_NAME           = "METRICS_NAME"
 	EXCLUDE_METRIC_LABELS  = "EXCLUDE_METRICS_LABELS"
+	OPEX_METRIC_PREFIX     = "OPEX_METRIC_PREFIX"
 )
 
 type prometheusSink struct {
@@ -198,10 +199,14 @@ func main() {
 	skipFailedStr := os.Getenv(SKIP_VALIDATION_FAILED)
 	labels := parseStringToMap(os.Getenv(METRICS_LABELS))
 	excludeLabels := parseStringToSlice(os.Getenv(EXCLUDE_METRIC_LABELS))
+	opexMetricsPrefix := os.Getenv(os.Getenv(OPEX_METRIC_PREFIX))
 	metricName := os.Getenv(METRICS_NAME)
 	if metricName == "" {
 		metricName = "namespace_app_rollouts_unified_anomaly"
 	}
+	if opexMetricsPrefix == "" {
+		opexMetricsPrefix = "numaflow_prom_sink"
+	}
 	var metricPort int
 	var ignoreMetricsTs, enableMsgTransformer bool
 	meticslabels := numaflag.MapFlag{}
@@ -224,7 +229,7 @@ func main() {
 	ps := prometheusSink{logger: logger, skipFailed: skipFailed, labels: labels, excludeLabels: excludeLabels,
 		ignoreMetricsTs: ignoreMetricsTs, metricsName: metricName, enableMsgTransformer: enableMsgTransformer}
 
-	ps.metrics = NewMetricsServer(labels)
+	ps.metrics = NewMetricsServer(labels, opexMetricsPrefix)
 	go ps.metrics.startMetricServer(metricPort)
 	ps.logger.Infof("Metrics publisher initialized with port=%d", metricPort)
 	err = sinksdk.NewServer(&ps).Start(context.Background())

diff --git a/prometheus-pusher/main_test.go b/prometheus-pusher/main_test.go
@@ -64,7 +64,7 @@ func TestPusher(t *testing.T) {
 
 	pgwOK := httptest.NewServer(
 		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-			dec := expfmt.NewDecoder(r.Body, expfmt.FmtProtoDelim)
+			dec := expfmt.NewDecoder(r.Body, expfmt.ResponseFormat(r.Header))
 
 			var mf io_prometheus_client.MetricFamily
 			dec.Decode(&mf)

diff --git a/prometheus-pusher/metrics.go b/prometheus-pusher/metrics.go
@@ -16,32 +16,33 @@ type MetricsPublisher struct {
 	metricsTotalSkipped     prometheus.Counter
 	metricsAnomalyGenerated *prometheus.CounterVec
 	labels                  map[string]string
+	opexMetricPrefix        string
 }
 
 func (mp *MetricsPublisher) registerMertics() {
 	mp.metricsTotalPushed = promauto.NewCounter(prometheus.CounterOpts{
-		Name:        "total_metrics_pushed",
+		Name:        mp.opexMetricPrefix + "_" + "total_metrics_pushed",
 		Help:        "The total number of metrics pushed",
 		ConstLabels: mp.labels,
 	})
 	mp.metricsTotalSuccess = promauto.NewCounter(prometheus.CounterOpts{
-		Name:        "total_metrics_success",
+		Name:        mp.opexMetricPrefix + "_" + "total_metrics_success",
 		Help:        "The total number of metrics successfully pushed",
 		ConstLabels: mp.labels,
 	})
 	mp.metricsTotalFailed = promauto.NewCounter(prometheus.CounterOpts{
-		Name:        "total_metrics_failed",
+		Name:        mp.opexMetricPrefix + "_" + "total_metrics_failed",
 		Help:        "The total number of metrics failed push",
 		ConstLabels: mp.labels,
 	})
 	mp.metricsTotalSkipped = promauto.NewCounter(prometheus.CounterOpts{
-		Name:        "total_metrics_skipped",
+		Name:        mp.opexMetricPrefix + "_" + "total_metrics_skipped",
 		Help:        "The total number of metrics skipped",
 		ConstLabels: mp.labels,
 	})
 
 	mp.metricsAnomalyGenerated = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name:        "total_anomaly_generated",
+		Name:        mp.opexMetricPrefix + "_" + "total_anomaly_generated",
 		Help:        "The total count of anomaly score generator",
 		ConstLabels: mp.labels,
 	}, []string{"namespace", "app", "metrics"})
@@ -64,9 +65,10 @@ func (mp *MetricsPublisher) IncreaseAnomalyGenerated(namespace, app, metricName
 	mp.metricsAnomalyGenerated.WithLabelValues(namespace, app, metricName).Inc()
 }
 
-func NewMetricsServer(labels map[string]string) *MetricsPublisher {
+func NewMetricsServer(labels map[string]string, prefix string) *MetricsPublisher {
 	metricsPublisher := &MetricsPublisher{}
 	metricsPublisher.labels = labels
+	metricsPublisher.opexMetricPrefix = prefix
 	metricsPublisher.registerMertics()
 	return metricsPublisher
 

diff --git a/prometheus-pusher/metrics_test.go b/prometheus-pusher/metrics_test.go
@@ -7,7 +7,7 @@ import (
 )
 
 func TestMetricsPublisher(t *testing.T) {
-	mp := NewMetricsServer(map[string]string{"label": "val1", "label2": "val2"})
+	mp := NewMetricsServer(map[string]string{"label": "val1", "label2": "val2"}, "prefix")
 	mp.IncreaseTotalPushed()
 	mp.IncreaseTotalSuccess()
 	mp.IncreaseTotalSkipped()