PMM-12848 Refactoring currentop_collector. (#886)

* PMM-12848 Refactoring currentop_collector.

* PMM-12848 fix linter.

* PMM-12848 fix linter.

* PMM-12848 fix linter.

Author: BupycHuk

exporter/currentop_collector.go

@@ -66,6 +66,7 @@ func (d *currentopCollector) collect(ch chan<- prometheus.Metric) {
 	client := d.base.client
 	slowtime, err := time.ParseDuration(d.currentopslowtime)
 	if err != nil {
+		logger.Errorf("Failed to parse slowtime: %s", err)
 		ch <- prometheus.NewInvalidMetric(prometheus.NewInvalidDesc(err), err)
 		return
 	}
@@ -89,6 +90,7 @@ func (d *currentopCollector) collect(ch chan<- prometheus.Metric) {
 
 	var r primitive.M
 	if err := res.Decode(&r); err != nil {
+		logger.Errorf("Failed to decode currentOp response: %s", err)
 		ch <- prometheus.NewInvalidMetric(prometheus.NewInvalidDesc(err), err)
 		return
 	}
@@ -99,10 +101,16 @@ func (d *currentopCollector) collect(ch chan<- prometheus.Metric) {
 	inprog, ok := r["inprog"].(primitive.A)
 
 	if !ok {
+		logger.Errorf("Invalid type primitive.A assertion for 'inprog': %T", r["inprog"])
 		ch <- prometheus.NewInvalidMetric(prometheus.NewInvalidDesc(ErrInvalidOrMissingInprogEntry),
 			ErrInvalidOrMissingInprogEntry)
 	}
 
+	labels := d.topologyInfo.baseLabels()
+	ln := []string{"opid", "op", "desc", "database", "collection", "ns"}
+	const name = "mongodb_currentop_query_uptime"
+	pd := prometheus.NewDesc(name, " mongodb_currentop_query_uptime currentop_query", ln, labels)
+
 	for _, bsonMap := range inprog {
 
 		bsonMapElement, ok := bsonMap.(primitive.M)
@@ -137,18 +145,8 @@ func (d *currentopCollector) collect(ch chan<- prometheus.Metric) {
 			continue
 		}
 
-		labels := d.topologyInfo.baseLabels()
-		labels["opid"] = strconv.Itoa(int(opid))
-		labels["op"] = op
-		labels["desc"] = desc
-		labels["database"] = db
-		labels["collection"] = collection
-		labels["ns"] = namespace
-
-		m := primitive.M{"uptime": microsecs_running}
+		lv := []string{strconv.Itoa(int(opid)), op, desc, db, collection, namespace}
 
-		for _, metric := range makeMetrics("currentop_query", m, labels, d.compatibleMode) {
-			ch <- metric
-		}
+		ch <- prometheus.MustNewConstMetric(pd, prometheus.GaugeValue, float64(microsecs_running), lv...)
 	}
 }

exporter/currentop_collector_test.go

@@ -17,7 +17,6 @@ package exporter
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -31,32 +30,31 @@ import (
 )
 
 func TestCurrentopCollector(t *testing.T) {
-	// It seems like this test needs the queries to continue running so that current oplog is not empty.
-	// TODO: figure out how to restore this test.
-	t.Skip()
-
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
 	var wg sync.WaitGroup
 
 	client := tu.DefaultTestClient(ctx, t)
 
 	database := client.Database("testdb")
-	database.Drop(ctx)
+	_ = database.Drop(ctx)
 
 	defer func() {
 		err := database.Drop(ctx)
 		assert.NoError(t, err)
 	}()
+	ch := make(chan struct{})
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		for i := 0; i < 300; i++ {
-			coll := fmt.Sprintf("testcol_%02d", i)
-			_, err := database.Collection(coll).InsertOne(ctx, bson.M{"f1": 1, "f2": "2"})
+		coll := "testcol_01"
+		for j := 0; j < 100; j++ { //nolint:intrange // false positive
+			_, err := database.Collection(coll).InsertOne(ctx, bson.M{"f1": j, "f2": "2"})
 			assert.NoError(t, err)
 		}
+		ch <- struct{}{}
+		_, _ = database.Collection(coll).Find(ctx, bson.M{"$where": "function() {return sleep(100)}"})
 	}()
 
 	ti := labelsGetterMock{}
@@ -76,6 +74,10 @@ func TestCurrentopCollector(t *testing.T) {
 		"mongodb_currentop_query_uptime",
 	}
 
+	<-ch
+
+	time.Sleep(1 * time.Second)
+
 	count := testutil.CollectAndCount(c, filter...)
 	assert.True(t, count > 0)
 	wg.Wait()