From cd7d3995ea42640dfe818a5b26c7af346aad022f Mon Sep 17 00:00:00 2001
From: Vaibhav Bhembre <vbhembre@coreweave.com>
Date: Mon, 15 Apr 2024 17:38:08 -0400
Subject: [PATCH] mds: preprocess metrics to avoid dups

---
 ceph/mds.go | 83 ++++++++++++++++++++++++++++++++++++-----------------
 1 file changed, 57 insertions(+), 26 deletions(-)

diff --git a/ceph/mds.go b/ceph/mds.go
index 3a4150c..5b50531 100644
--- a/ceph/mds.go
+++ b/ceph/mds.go
@@ -15,7 +15,9 @@
 package ceph
 
 import (
+	"bytes"
 	"context"
+	"encoding/gob"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -23,6 +25,8 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -262,6 +266,27 @@ type mdsStatus struct {
 	Uptime             float64 `json:"uptime"`
 }
 
+type mdsLabels struct {
+	FSName    string
+	MDSName   string
+	State     string
+	OpType    string
+	FSOpType  string
+	FlagPoint string
+	Inode     string
+}
+
+func (ml mdsLabels) Hash() string {
+	var b bytes.Buffer
+	gob.NewEncoder(&b).Encode(ml)
+	return b.String()
+}
+
+func (ml *mdsLabels) UnHash(hash string) error {
+	return gob.NewDecoder(strings.NewReader(hash)).Decode(ml)
+
+}
+
 type mdsSlowOp struct {
 	Ops []struct {
 		// Custom fields for easy parsing by caller.
@@ -361,7 +386,11 @@ func (m *MDSCollector) collectMDSSlowOps() {
 			return
 		}
 
+		var metricMap sync.Map
+
 		for _, op := range mso.Ops {
+			var ml mdsLabels
+
 			if op.TypeData.OpType == "client_request" {
 				opd, err := extractOpFromDescription(op.Description)
 				if err != nil {
@@ -369,42 +398,44 @@ func (m *MDSCollector) collectMDSSlowOps() {
 					continue
 				}
 
-				select {
-				case m.ch <- prometheus.MustNewConstMetric(
-					m.MDSBlockedOps,
-					prometheus.CounterValue,
-					1,
-					mss.FsName,
-					mdsName,
-					mss.State,
-					op.TypeData.OpType,
-					opd.fsOpType,
-					op.TypeData.FlagPoint,
-					opd.inode,
-				):
-				default:
-				}
-
-				continue
+				ml.FSOpType = opd.fsOpType
+				ml.Inode = opd.inode
 			}
 
+			ml.FSName = mss.FsName
+			ml.MDSName = mdsName
+			ml.State = mss.State
+			ml.OpType = op.TypeData.OpType
+			ml.FlagPoint = op.TypeData.FlagPoint
+
+			cnt, _ := metricMap.LoadOrStore(ml.Hash(), new(int32))
+			v := cnt.(*int32)
+			atomic.AddInt32(v, 1)
+		}
+
+		metricMap.Range(func(key, value any) bool {
+			var ml mdsLabels
+			ml.UnHash(fmt.Sprint(key))
+			v := value.(*int32)
+
 			select {
 			case m.ch <- prometheus.MustNewConstMetric(
 				m.MDSBlockedOps,
 				prometheus.CounterValue,
-				1,
-				mss.FsName,
-				mdsName,
-				mss.State,
-				op.TypeData.OpType,
-				"",
-				op.TypeData.FlagPoint,
-				"",
+				float64(*v),
+				ml.FSName,
+				ml.MDSName,
+				ml.State,
+				ml.OpType,
+				ml.FSOpType,
+				ml.FlagPoint,
+				ml.Inode,
 			):
 			default:
 			}
-		}
 
+			return true
+		})
 	}
 }