mds: preprocess metrics to avoid dups

ceph/mds.go

@@ -15,14 +15,18 @@
 package ceph
 
 import (
+	"bytes"
 	"context"
+	"encoding/gob"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"os/exec"
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -262,6 +266,26 @@ 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,50 +385,56 @@ 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 {
 					m.logger.WithField("mds", mdsName).WithError(err).Error("failed parsing blocked ops description")
 					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(key.(string))
+			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
+		})
 	}
 }
 

ceph/mds_test.go

@@ -393,7 +393,7 @@ func TestMDSBlockedOps(t *testing.T) {
 			require.NoError(t, err)
 
 			for _, re := range tt.reMatch {
-				require.True(t, re.Match(buf))
+				require.True(t, re.Match(buf), string(buf))
 			}
 
 			for _, re := range tt.reUnmatch {