wip: source metrics [cleanup]

Signed-off-by: adarsh0728 <[email protected]>

.codecov.yml

@@ -17,4 +17,4 @@ coverage:
     project:
       default:
         # allow test coverage to drop by 2%, assume that it's typically due to CI problems
-        threshold: 2
+        threshold: "2"

pkg/metrics/metrics.go

@@ -32,8 +32,7 @@ const (
 	LabelVertexType         = "vertex_type"
 	LabelPartitionName      = "partition_name"
 	LabelMonoVertexName     = "mvtx_name"
-
-	LabelReason = "reason"
+	LabelReason             = "reason"
 )
 
 // Generic forwarder metrics

pkg/sources/generator/tickgen.go

@@ -31,7 +31,6 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	"github.com/numaproj/numaflow/pkg/sources/sourcer"
 )
@@ -199,7 +198,6 @@ loop:
 		// we implement Read With Wait semantics
 		select {
 		case r := <-mg.srcChan:
-			tickgenSourceReadCount.With(map[string]string{metrics.LabelVertex: mg.vertexName, metrics.LabelPipeline: mg.pipelineName}).Inc()
 			msgs = append(msgs, mg.newReadMessage(r.key, r.data, r.offset, r.ts))
 		case <-timeout:
 			break loop
@@ -240,7 +238,6 @@ func (mg *memGen) newWorker(ctx context.Context, rate int) func(chan time.Time,
 			case <-ctx.Done():
 				return
 			case ts := <-tickChan:
-				tickgenSourceCount.With(map[string]string{metrics.LabelVertex: mg.vertexName, metrics.LabelPipeline: mg.pipelineName}).Inc()
 				// we would generate all the keys in a round robin fashion
 				// even if there are multiple pods, all the pods will generate same keys in the same order.
 				// TODO: alternatively, we could also think about generating a subset of keys per pod.

pkg/sources/http/http.go

@@ -33,7 +33,6 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedtls "github.com/numaproj/numaflow/pkg/shared/tls"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
@@ -211,7 +210,6 @@ loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-h.messages:
-			httpSourceReadCount.With(map[string]string{metrics.LabelVertex: h.vertexName, metrics.LabelPipeline: h.pipelineName}).Inc()
 			msgs = append(msgs, m)
 		case <-timeout:
 			h.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", h.readTimeout), zap.Int("read", len(msgs)))

pkg/sources/jetstream/jetstream.go

@@ -30,7 +30,6 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/sources/sourcer"
@@ -401,7 +400,6 @@ loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-ns.messages:
-			jetstreamSourceReadCount.With(map[string]string{metrics.LabelVertex: ns.vertexName, metrics.LabelPipeline: ns.pipelineName}).Inc()
 			msgs = append(msgs, m)
 		case <-timeout:
 			ns.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", ns.readTimeout), zap.Int("read", len(msgs)))

pkg/sources/kafka/metrics.go

@@ -23,13 +23,6 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 )
 
-// kafkaSourceReadCount is used to indicate the number of messages read
-var kafkaSourceReadCount = promauto.NewCounterVec(prometheus.CounterOpts{
-	Subsystem: "kafka_source",
-	Name:      "read_total",
-	Help:      "Total number of messages Read",
-}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
-
 // kafkaSourceAckCount is used to indicate the number of messages Acknowledged
 var kafkaSourceAckCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "kafka_source",

pkg/sources/kafka/reader.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strconv"
 	"sync"
 	"time"
 
@@ -159,11 +158,6 @@ loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-ks.handler.messages:
-			kafkaSourceReadCount.With(map[string]string{
-				metrics.LabelVertex:        ks.vertexName,
-				metrics.LabelPipeline:      ks.pipelineName,
-				metrics.LabelPartitionName: strconv.Itoa(int(m.Partition)),
-			}).Inc()
 			msgs = append(msgs, ks.toReadMessage(m))
 		case <-timeout:
 			// log that timeout has happened and don't return an error

pkg/sources/nats/nats.go

@@ -27,7 +27,6 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
 	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/sources/sourcer"
@@ -192,7 +191,6 @@ loop:
 	for i := int64(0); i < count; i++ {
 		select {
 		case m := <-ns.messages:
-			natsSourceReadCount.With(map[string]string{metrics.LabelVertex: ns.vertexName, metrics.LabelPipeline: ns.pipelineName}).Inc()
 			msgs = append(msgs, m)
 		case <-timeout:
 			ns.logger.Debugw("Timed out waiting for messages to read.", zap.Duration("waited", ns.readTimeout), zap.Int("read", len(msgs)))