Repository: numaproj/numaflow
Pull request #888 (2023-07-21 15:51:14)

chore: create separate source forwarder metrics instead of exposing existing ones
Also update the rater such that for source vertex, rater uses the source metric to get read total count. Tested by creating a new pipeline locally and verified rate shows up at source vertex.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->



Patch:
diff --git a/docs/operations/metrics/metrics.md b/docs/operations/metrics/metrics.md
index 83e617f9..1afc2bf5 100644
--- a/docs/operations/metrics/metrics.md
+++ b/docs/operations/metrics/metrics.md
@@ -14,6 +14,13 @@ These metrics can be used to determine throughput of your pipeline.
 
 | Metric name                           | Metric type | Labels                                                                                                                       | Description                                                                                                      |
 |---------------------------------------|-------------|------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------|
+| `source_forwarder_read_total`         | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages read by a given Source Vertex                                              |
+| `source_forwarder_read_bytes_total`   | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes read by a given Source Vertex                                                 |
+| `source_forwarder_write_total`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages written to Inter-Step Buffer by a given Source Vertex                      |
+| `source_forwarder_write_bytes_total`  | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes written to Inter-Step Buffer by a given Source Vertex                         |
+| `source_forwarder_ack_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages acknowledged by a given Source Vertex                                      |
+| `source_forwarder_drop_total`         | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages dropped by a given Source Vertex due to a full Inter-Step Buffer Partition |
+| `source_forwarder_drop_bytes_total`   | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes dropped by a given Source Vertex due to a full Inter-Step Buffer Partition    |
 | `forwarder_read_total`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages read by a given Vertex from an Inter-Step Buffer Partition                 |
 | `forwarder_read_bytes_total`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of bytes read by a given Vertex from an Inter-Step Buffer Partition                    |
 | `forwarder_write_total`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Provides the total number of messages written to Inter-Step Buffer by a given Vertex                             |
@@ -72,12 +79,14 @@ These metrics can be used to determine throughput of your pipeline.
 
 These metrics can be used to determine the latency of your pipeline.
 
-| Metric name                               | Metric type | Labels                                                                                        | Description                                                                                    |
-|-------------------------------------------|-------------|-----------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------|
-| `forwarder_udf_processing_time`           | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's)   |
-| `forwarder_forward_chunk_processing_time` | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of the forwarder function as a whole |
-| `reduce_pnf_process_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the processing times of the reducer                       |
-| `reduce_pnf_forward_time`                 | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the forwarding times of the reducer                       |
+| Metric name                                      | Metric type | Labels                                                                                        | Description                                                                                           |
+|--------------------------------------------------|-------------|-----------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------------------------------|
+| `source_forwarder_transformer_processing_time`   | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of User Defined Source Transformer          |
+| `source_forwarder_forward_chunk_processing_time` | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of the source forwarder function as a whole |
+| `forwarder_udf_processing_time`                  | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of User Defined Functions. (UDF's)          |
+| `forwarder_forward_chunk_processing_time`        | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>` | Provides a histogram distribution of the processing times of the forwarder function as a whole        |
+| `reduce_pnf_process_time`                        | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the processing times of the reducer                              |
+| `reduce_pnf_forward_time`                        | Histogram   | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `replica=<replica-index>`         | Provides a histogram distribution of the forwarding times of the reducer                              |
 
 ### Errors
 
@@ -85,6 +94,10 @@ These metrics can be used to determine if there are any errors in the pipeline
 
 | Metric name                           | Metric type | Labels                                                                                                                       | Description                                                                   |
 |---------------------------------------|-------------|------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------------|
+| `source_forwarder_platform_error`     | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any internal errors which could stop pipeline processing            |
+| `source_forwarder_read_error`         | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while reading messages by the source forwarder           |
+| `source_forwarder_write_error`        | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while writing messages by the source forwarder           |
+| `source_forwarder_ack_error`          | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while acknowledging messages by the source forwarder     |
 | `forwarder_platform_error`            | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any internal errors which could stop pipeline processing            |
 | `forwarder_read_error`                | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while reading messages by the forwarder                  |
 | `forwarder_write_error`               | Counter     | `vertex=<vertex-name>` <br> `pipeline=<pipeline-name>` <br> `partition_name=<partition-name>`                                | Indicates any errors while writing messages by the forwarder                  |
diff --git a/pkg/daemon/server/service/rater/pod_tracker.go b/pkg/daemon/server/service/rater/pod_tracker.go
index eac26d27..b8924447 100644
--- a/pkg/daemon/server/service/rater/pod_tracker.go
+++ b/pkg/daemon/server/service/rater/pod_tracker.go
@@ -90,8 +90,10 @@ func (pt *PodTracker) Start(ctx context.Context) error {
 					var vType string
 					if v.IsReduceUDF() {
 						vType = "reduce"
+					} else if v.IsASource() {
+						vType = "source"
 					} else {
-						vType = "non_reduce"
+						vType = "other"
 					}
 					for i := 0; i < int(v.Scale.GetMaxReplicas()); i++ {
 						podName := fmt.Sprintf("%s-%s-%d", pt.pipeline.Name, v.Name, i)
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index 3872a792..1b7b1b2b 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -234,6 +234,8 @@ func (r *Rater) getPodReadCounts(vertexName, vertexType, podName string) *PodRea
 	var readTotalMetricName string
 	if vertexType == "reduce" {
 		readTotalMetricName = "reduce_isb_reader_read_total"
+	} else if vertexType == "source" {
+		readTotalMetricName = "source_forwarder_read_total"
 	} else {
 		readTotalMetricName = "forwarder_read_total"
 	}
diff --git a/pkg/forward/forward.go b/pkg/forward/forward.go
index a86027c3..59c11a1c 100644
--- a/pkg/forward/forward.go
+++ b/pkg/forward/forward.go
@@ -207,9 +207,9 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	readMessages, err := isdf.fromBufferPartition.Read(ctx, isdf.opts.readBatchSize)
 	if err != nil {
 		isdf.opts.logger.Warnw("failed to read fromBufferPartition", zap.Error(err))
-		ReadMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
+		readMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 	}
-	ReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readMessages)))
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readMessages)))
 
 	// process only if we have any read messages. There is a natural looping here if there is an internal error while
 	// reading, and we are not able to proceed.
@@ -291,7 +291,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// send to UDF only the data messages
 		for idx, m := range dataMessages {
 			// emit message size metric
-			ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(m.Payload)))
+			readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(m.Payload)))
 			// assign watermark to the message
 			m.Watermark = time.Time(processorWM)
 			// send UDF processing work to the channel
@@ -304,7 +304,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 		// context.Done() is closed.
 		wg.Wait()
 		isdf.opts.logger.Debugw("concurrent applyUDF completed", zap.Int("concurrency", isdf.opts.udfConcurrency), zap.Duration("took", time.Since(concurrentUDFProcessingStart)))
-		ConcurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
+		concurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(concurrentUDFProcessingStart).Microseconds()))
 		// UDF processing is done.
 
 		// let's figure out which vertex to send the results to.
@@ -313,7 +313,7 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 			// look for errors in udf processing, if we see even 1 error NoAck all messages
 			// then return. Handling partial retrying is not worth ATM.
 			if m.udfError != nil {
-				UdfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
+				udfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 				isdf.opts.logger.Errorw("failed to applyUDF", zap.Error(m.udfError))
 				// As there's no partial failure, non-ack all the readOffsets
 				isdf.fromBufferPartition.NoAck(ctx, readOffsets)
@@ -403,13 +403,13 @@ func (isdf *InterStepDataForward) forwardAChunk(ctx context.Context) {
 	// implicit return for posterity :-)
 	if err != nil {
 		isdf.opts.logger.Errorw("failed to ack from buffer", zap.Error(err))
-		AckMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
+		ackMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
 		return
 	}
-	AckMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
+	ackMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(readOffsets)))
 
 	// ProcessingTimes of the entire forwardAChunk
-	ForwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+	forwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 }
 
 // streamMessage streams the data messages to the next step.
@@ -437,14 +437,14 @@ func (isdf *InterStepDataForward) streamMessage(
 		// send to UDF only the data messages
 
 		// emit message size metric
-		ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).
+		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).
 			Add(float64(len(dataMessages[0].Payload)))
 		// assign watermark to the message
 		dataMessages[0].Watermark = time.Time(processorWM)
 
 		// process the UDF and get the result
 		start := time.Now()
-		UdfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
+		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 
 		writeMessageCh := make(chan isb.WriteMessage)
 		errs, ctx := errgroup.WithContext(ctx)
@@ -459,7 +459,7 @@ func (isdf *InterStepDataForward) streamMessage(
 			// add partition to the ID, this is to make sure that the ID is unique across partitions
 			writeMessage.ID = fmt.Sprintf("%s-%d-%d", dataMessages[0].ReadOffset.String(), isdf.fromBufferPartition.GetPartitionIdx(), msgIndex)
 			msgIndex += 1
-			UdfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(1))
+			udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(1))
 
 			// update toBuffers
 			if err := isdf.whereToStep(&writeMessage, messageToStep, dataMessages[0]); err != nil {
@@ -482,17 +482,17 @@ func (isdf *InterStepDataForward) streamMessage(
 		// look for errors in udf processing, if we see even 1 error NoAck all messages
 		// then return. Handling partial retrying is not worth ATM.
 		if err := errs.Wait(); err != nil {
-			UdfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
+			udfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
 				metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 			// We do not retry as we are streaming
 			if ok, _ := isdf.IsShuttingDown(); ok {
 				isdf.opts.logger.Errorw("UDF.Apply, Stop called while stuck on an internal error", zap.Error(err))
-				PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+				platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 			}
 			return nil, fmt.Errorf("failed to applyUDF, error: %w", err)
 		}
 
-		UdfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
+		udfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName,
 			metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 	} else {
 		// Even not data messages, forward the message to the edge buffer (could be multiple edges)
@@ -603,10 +603,10 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBufferPar
 					needRetry = true
 					// we retry only failed messages
 					failedMessages = append(failedMessages, msg)
-					WriteMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
+					writeMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
 					// a shutdown can break the blocking loop caused due to InternalErr
 					if ok, _ := isdf.IsShuttingDown(); ok {
-						PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+						platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 						return writeOffsets, fmt.Errorf("writeToBuffer failed, Stop called while stuck on an internal error with failed messages:%d, %v", len(failedMessages), errs)
 					}
 				}
@@ -636,10 +636,10 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBufferPar
 		}
 	}
 
-	DropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(totalCount - writeCount))
-	DropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(dropBytes)
-	WriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
-	WriteBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
+	dropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(totalCount - writeCount))
+	dropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(dropBytes)
+	writeMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
+	writeBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
 	return writeOffsets, nil
 }
 
@@ -647,12 +647,12 @@ func (isdf *InterStepDataForward) writeToBuffer(ctx context.Context, toBufferPar
 func (isdf *InterStepDataForward) concurrentApplyUDF(ctx context.Context, readMessagePair <-chan *readWriteMessagePair) {
 	for message := range readMessagePair {
 		start := time.Now()
-		UdfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
+		udfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Inc()
 		writeMessages, err := isdf.applyUDF(ctx, message.readMessage)
-		UdfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(writeMessages)))
+		udfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Add(float64(len(writeMessages)))
 		message.writeMessages = append(message.writeMessages, writeMessages...)
 		message.udfError = err
-		UdfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+		udfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.fromBufferPartition.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 	}
 }
 
@@ -671,7 +671,7 @@ func (isdf *InterStepDataForward) applyUDF(ctx context.Context, readMessage *isb
 			// this does not mean we should prohibit this from a shutdown.
 			if ok, _ := isdf.IsShuttingDown(); ok {
 				isdf.opts.logger.Errorw("UDF.Apply, Stop called while stuck on an internal error", zap.Error(err))
-				PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+				platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 				return nil, err
 			}
 			continue
@@ -698,7 +698,7 @@ func (isdf *InterStepDataForward) whereToStep(writeMessage *isb.WriteMessage, me
 		// a shutdown can break the blocking loop caused due to InternalErr
 		if ok, _ := isdf.IsShuttingDown(); ok {
 			err := fmt.Errorf("whereToStep, Stop called while stuck on an internal error, %v", err)
-			PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+			platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 			return err
 		}
 		return err
diff --git a/pkg/forward/metrics.go b/pkg/forward/metrics.go
index ca3eb97d..09f42227 100644
--- a/pkg/forward/metrics.go
+++ b/pkg/forward/metrics.go
@@ -23,123 +23,123 @@ import (
 	"github.com/numaproj/numaflow/pkg/metrics"
 )
 
-// ReadMessagesCount is used to indicate the number of messages read
-var ReadMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// readMessagesCount is used to indicate the number of messages read
+var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_total",
 	Help:      "Total number of Messages Read",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// ReadBytesCount is to indicate the number of bytes read
-var ReadBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// readBytesCount is to indicate the number of bytes read
+var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_bytes_total",
 	Help:      "Total number of bytes read",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// ReadMessagesError is used to indicate the number of errors messages read
-var ReadMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
+// readMessagesError is used to indicate the number of errors messages read
+var readMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "read_error_total",
 	Help:      "Total number of Read Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// WriteMessagesCount is used to indicate the number of messages written
-var WriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// writeMessagesCount is used to indicate the number of messages written
+var writeMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_total",
 	Help:      "Total number of Messages Written",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// WriteBytesCount is to indicate the number of bytes written
-var WriteBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// writeBytesCount is to indicate the number of bytes written
+var writeBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_bytes_total",
 	Help:      "Total number of bytes written",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// WriteMessagesError is used to indicate the number of errors messages written
-var WriteMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
+// writeMessagesError is used to indicate the number of errors messages written
+var writeMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "write_error_total",
 	Help:      "Total number of Write Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// DropMessagesCount is used to indicate the number of messages dropped
-var DropMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// dropMessagesCount is used to indicate the number of messages dropped
+var dropMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "drop_total",
 	Help:      "Total number of Messages Dropped",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// DropBytesCount is to indicate the number of bytes dropped
-var DropBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// dropBytesCount is to indicate the number of bytes dropped
+var dropBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "drop_bytes_total",
 	Help:      "Total number of Bytes Dropped",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// AckMessagesCount is used to indicate the number of  messages acknowledged
-var AckMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// ackMessagesCount is used to indicate the number of  messages acknowledged
+var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "ack_total",
 	Help:      "Total number of Messages Acknowledged",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// AckMessageError is used to indicate the errors in the number of  messages acknowledged
-var AckMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
+// ackMessageError is used to indicate the errors in the number of  messages acknowledged
+var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "ack_error_total",
 	Help:      "Total number of Acknowledged Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// UdfError is used to indicate the number of UDF errors
-var UdfError = promauto.NewCounterVec(prometheus.CounterOpts{
+// udfError is used to indicate the number of UDF errors
+var udfError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_error_total",
 	Help:      "Total number of UDF Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// PlatformError is used to indicate the number of Internal/Platform errors
-var PlatformError = promauto.NewCounterVec(prometheus.CounterOpts{
+// platformError is used to indicate the number of Internal/Platform errors
+var platformError = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "platform_error_total",
 	Help:      "Total number of platform Errors",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline})
 
-// ForwardAChunkProcessingTime is a histogram to Observe forwardAChunk Processing times as a whole
-var ForwardAChunkProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+// forwardAChunkProcessingTime is a histogram to Observe forwardAChunk Processing times as a whole
+var forwardAChunkProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Subsystem: "forwarder",
 	Name:      "forward_chunk_processing_time",
 	Help:      "Processing times of the entire forward a chunk (100 microseconds to 20 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// UdfProcessingTime is a histogram to Observe UDF Processing times as a whole
-var UdfProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+// udfProcessingTime is a histogram to Observe UDF Processing times as a whole
+var udfProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_processing_time",
 	Help:      "Processing times of UDF (100 microseconds to 15 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*15, 60),
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// ConcurrentUDFProcessingTime is a histogram to Observe UDF Processing times as a whole
-var ConcurrentUDFProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+// concurrentUDFProcessingTime is a histogram to Observe UDF Processing times as a whole
+var concurrentUDFProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Subsystem: "forwarder",
 	Name:      "concurrent_udf_processing_time",
 	Help:      "Processing times of Concurrent UDF (100 microseconds to 20 minutes)",
 	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// UdfReadMessagesCount is used to indicate the number of messages read by UDF
-var UdfReadMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// udfReadMessagesCount is used to indicate the number of messages read by UDF
+var udfReadMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_read_total",
 	Help:      "Total number of Messages Read by UDF",
 }, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
 
-// UdfWriteMessagesCount is used to indicate the number of messages written by UDF
-var UdfWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+// udfWriteMessagesCount is used to indicate the number of messages written by UDF
+var udfWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
 	Subsystem: "forwarder",
 	Name:      "udf_write_total",
 	Help:      "Total number of Messages Written by UDF",
diff --git a/pkg/sources/forward/data_forward.go b/pkg/sources/forward/data_forward.go
index 2f842cf4..8be79486 100644
--- a/pkg/sources/forward/data_forward.go
+++ b/pkg/sources/forward/data_forward.go
@@ -189,9 +189,9 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	readMessages, err := isdf.reader.Read(ctx, isdf.opts.readBatchSize)
 	if err != nil {
 		isdf.opts.logger.Warnw("failed to read from source", zap.Error(err))
-		forward.ReadMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
+		readMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
 	}
-	forward.ReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readMessages)))
+	readMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readMessages)))
 
 	// process only if we have any read messages. There is a natural looping here if there is an internal error while
 	// reading, and we are not able to proceed.
@@ -238,7 +238,7 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	// send to transformer only the data messages
 	for idx, m := range readMessages {
 		// emit message size metric
-		forward.ReadBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(m.Payload)))
+		readBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(m.Payload)))
 		// assign watermark to the message
 		m.Watermark = time.Time(processorWM)
 		// send transformer processing work to the channel
@@ -251,7 +251,7 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	// context.Done() is closed.
 	wg.Wait()
 	isdf.opts.logger.Debugw("concurrent applyTransformer completed", zap.Int("concurrency", isdf.opts.transformerConcurrency), zap.Duration("took", time.Since(concurrentTransformerProcessingStart)))
-	forward.ConcurrentUDFProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(concurrentTransformerProcessingStart).Microseconds()))
+	concurrentTransformerProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(concurrentTransformerProcessingStart).Microseconds()))
 	// transformer processing is done.
 
 	// publish source watermark and assign IsLate attribute based on new event time.
@@ -285,7 +285,7 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 		// look for errors in transformer processing, if we see even 1 error NoAck all messages
 		// then return. Handling partial retrying is not worth ATM.
 		if m.transformerError != nil {
-			forward.UdfError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
+			transformerError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
 			isdf.opts.logger.Errorw("failed to apply source transformer", zap.Error(m.transformerError))
 			// As there's no partial failure, non-ack all the readOffsets
 			isdf.reader.NoAck(ctx, readOffsets)
@@ -351,13 +351,13 @@ func (isdf *DataForward) forwardAChunk(ctx context.Context) {
 	// implicit return for posterity :-)
 	if err != nil {
 		isdf.opts.logger.Errorw("failed to ack from source", zap.Error(err))
-		forward.AckMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readOffsets)))
+		ackMessageError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readOffsets)))
 		return
 	}
-	forward.AckMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readOffsets)))
+	ackMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(readOffsets)))
 
 	// ProcessingTimes of the entire forwardAChunk
-	forward.ForwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+	forwardAChunkProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 }
 
 // ackFromBuffer acknowledges an array of offsets back to reader and is a blocking call or until shutdown has been initiated.
@@ -457,10 +457,10 @@ func (isdf *DataForward) writeToBuffer(ctx context.Context, toBufferPartition is
 					needRetry = true
 					// we retry only failed messages
 					failedMessages = append(failedMessages, msg)
-					forward.WriteMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
+					writeMessagesError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Inc()
 					// a shutdown can break the blocking loop caused due to InternalErr
 					if ok, _ := isdf.IsShuttingDown(); ok {
-						forward.PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+						platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 						return writeOffsets, fmt.Errorf("writeToBuffer failed, Stop called while stuck on an internal error with failed messages:%d, %v", len(failedMessages), errs)
 					}
 				}
@@ -490,10 +490,10 @@ func (isdf *DataForward) writeToBuffer(ctx context.Context, toBufferPartition is
 		}
 	}
 
-	forward.DropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(totalCount - writeCount))
-	forward.DropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(dropBytes)
-	forward.WriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
-	forward.WriteBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
+	dropMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(totalCount - writeCount))
+	dropBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(dropBytes)
+	writeMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(float64(writeCount))
+	writeBytesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: toBufferPartition.GetName()}).Add(writeBytes)
 	return writeOffsets, nil
 }
 
@@ -501,12 +501,12 @@ func (isdf *DataForward) writeToBuffer(ctx context.Context, toBufferPartition is
 func (isdf *DataForward) concurrentApplyTransformer(ctx context.Context, readMessagePair <-chan *readWriteMessagePair) {
 	for message := range readMessagePair {
 		start := time.Now()
-		forward.UdfReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
+		transformerReadMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Inc()
 		writeMessages, err := isdf.applyTransformer(ctx, message.readMessage)
-		forward.UdfWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(writeMessages)))
+		transformerWriteMessagesCount.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Add(float64(len(writeMessages)))
 		message.writeMessages = append(message.writeMessages, writeMessages...)
 		message.transformerError = err
-		forward.UdfProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(start).Microseconds()))
+		transformerProcessingTime.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName, metrics.LabelPartitionName: isdf.reader.GetName()}).Observe(float64(time.Since(start).Microseconds()))
 	}
 }
 
@@ -525,7 +525,7 @@ func (isdf *DataForward) applyTransformer(ctx context.Context, readMessage *isb.
 			// this does not mean we should prohibit this from a shutdown.
 			if ok, _ := isdf.IsShuttingDown(); ok {
 				isdf.opts.logger.Errorw("Transformer.Apply, Stop called while stuck on an internal error", zap.Error(err))
-				forward.PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+				platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 				return nil, err
 			}
 			continue
@@ -552,7 +552,7 @@ func (isdf *DataForward) whereToStep(writeMessage *isb.WriteMessage, messageToSt
 		// a shutdown can break the blocking loop caused due to InternalErr
 		if ok, _ := isdf.IsShuttingDown(); ok {
 			err := fmt.Errorf("whereToStep, Stop called while stuck on an internal error, %v", err)
-			forward.PlatformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
+			platformError.With(map[string]string{metrics.LabelVertex: isdf.vertexName, metrics.LabelPipeline: isdf.pipelineName}).Inc()
 			return err
 		}
 		return err
diff --git a/pkg/sources/forward/metrics.go b/pkg/sources/forward/metrics.go
new file mode 100644
index 00000000..f9a73056
--- /dev/null
+++ b/pkg/sources/forward/metrics.go
@@ -0,0 +1,146 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package forward
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/numaproj/numaflow/pkg/metrics"
+)
+
+// readMessagesCount is used to indicate the number of messages read
+var readMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "read_total",
+	Help:      "Total number of Messages Read",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// readBytesCount is to indicate the number of bytes read
+var readBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "read_bytes_total",
+	Help:      "Total number of bytes read",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// readMessagesError is used to indicate the number of errors messages read
+var readMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "read_error_total",
+	Help:      "Total number of Read Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// writeMessagesCount is used to indicate the number of messages written
+var writeMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "write_total",
+	Help:      "Total number of Messages Written",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// writeBytesCount is to indicate the number of bytes written
+var writeBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "write_bytes_total",
+	Help:      "Total number of bytes written",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// writeMessagesError is used to indicate the number of errors messages written
+var writeMessagesError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "write_error_total",
+	Help:      "Total number of Write Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// dropMessagesCount is used to indicate the number of messages dropped
+var dropMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "drop_total",
+	Help:      "Total number of Messages Dropped",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// dropBytesCount is to indicate the number of bytes dropped
+var dropBytesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "drop_bytes_total",
+	Help:      "Total number of Bytes Dropped",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// ackMessagesCount is used to indicate the number of  messages acknowledged
+var ackMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "ack_total",
+	Help:      "Total number of Messages Acknowledged",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// ackMessageError is used to indicate the errors in the number of  messages acknowledged
+var ackMessageError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "ack_error_total",
+	Help:      "Total number of Acknowledged Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// transformerError is used to indicate the number of source transformer errors
+var transformerError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "transformer_error_total",
+	Help:      "Total number of source transformer Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// platformError is used to indicate the number of Internal/Platform errors
+var platformError = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "platform_error_total",
+	Help:      "Total number of platform Errors",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline})
+
+// forwardAChunkProcessingTime is a histogram to Observe forwardAChunk Processing times as a whole
+var forwardAChunkProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "source_forwarder",
+	Name:      "forward_chunk_processing_time",
+	Help:      "Processing times of the entire forward a chunk (100 microseconds to 20 minutes)",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// transformerProcessingTime is a histogram to Observe Source Transformer Processing times as a whole
+var transformerProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "source_forwarder",
+	Name:      "transformer_processing_time",
+	Help:      "Processing times of source transformer (100 microseconds to 15 minutes)",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*15, 60),
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// concurrentTransformerProcessingTime is a histogram to Observe Source Transformer Processing times as a whole
+var concurrentTransformerProcessingTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Subsystem: "source_forwarder",
+	Name:      "concurrent_transformer_processing_time",
+	Help:      "Processing times of Concurrent source transformer (100 microseconds to 20 minutes)",
+	Buckets:   prometheus.ExponentialBucketsRange(100, 60000000*20, 60),
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// transformerReadMessagesCount is used to indicate the number of messages read by source transformer
+var transformerReadMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "transformer_read_total",
+	Help:      "Total number of Messages Read by source transformer",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
+
+// transformerWriteMessagesCount is used to indicate the number of messages written by source transformer
+var transformerWriteMessagesCount = promauto.NewCounterVec(prometheus.CounterOpts{
+	Subsystem: "source_forwarder",
+	Name:      "transformer_write_total",
+	Help:      "Total number of Messages Written by source transformer",
+}, []string{metrics.LabelVertex, metrics.LabelPipeline, metrics.LabelPartitionName})
diff --git a/pkg/sources/forward/options.go b/pkg/sources/forward/options.go
index f912b29f..bfdddf5e 100644
--- a/pkg/sources/forward/options.go
+++ b/pkg/sources/forward/options.go
@@ -64,7 +64,7 @@ func WithReadBatchSize(f int64) Option {
 	}
 }
 
-// WithTransformerConcurrency sets concurrency for UDF processing
+// WithTransformerConcurrency sets concurrency for source transformer processing
 func WithTransformerConcurrency(f int) Option {
 	return func(o *options) error {
 		o.transformerConcurrency = f
diff --git a/pkg/sources/forward/shutdown.go b/pkg/sources/forward/shutdown.go
index 5d0e47b3..e8017c85 100644
--- a/pkg/sources/forward/shutdown.go
+++ b/pkg/sources/forward/shutdown.go
@@ -66,7 +66,7 @@ func (isdf *DataForward) Stop() {
 // ForceStop sets up the force shutdown flag.
 func (isdf *DataForward) ForceStop() {
 	// call stop (what if we have an enthusiastic shutdown that forces first)
-	// e.g. I know I have written a wrong UDF, so shutdown ASAP
+	// e.g. I know I have written a wrong source transformer, so shutdown ASAP
 	isdf.Stop()
 	isdf.Shutdown.rwlock.Lock()
 	defer isdf.Shutdown.rwlock.Unlock()


Test patch:
diff --git a/pkg/daemon/server/service/rater/rater_test.go b/pkg/daemon/server/service/rater/rater_test.go
index 839f6e37..4d9026cb 100644
--- a/pkg/daemon/server/service/rater/rater_test.go
+++ b/pkg/daemon/server/service/rater/rater_test.go
@@ -46,6 +46,8 @@ func (m *raterMockHttpClient) Get(url string) (*http.Response, error) {
 		m.podOneCount = m.podOneCount + 20
 		resp := &http.Response{
 			StatusCode: 200,
+			// the test uses an abstract vertex without specifying vertex type, meaning it's neither source nor reduce,
+			// hence the default forwarder metric name "forwarder_read_total" is used to retrieve the metric
 			Body: io.NopCloser(bytes.NewReader([]byte(fmt.Sprintf(`
 # HELP forwarder_read_total Total number of Messages Read
 # TYPE forwarder_read_total counter
diff --git a/pkg/forward/forward_test.go b/pkg/forward/forward_test.go
index 47fd7a0d..75f5b4ba 100644
--- a/pkg/forward/forward_test.go
+++ b/pkg/forward/forward_test.go
@@ -1523,7 +1523,7 @@ func validateMetrics(t *testing.T, batchSize int64) {
 		forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
 	`
 
-	err := testutil.CollectAndCompare(ReadMessagesCount, strings.NewReader(metadata+expected), "forwarder_read_total")
+	err := testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "forwarder_read_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
@@ -1545,7 +1545,7 @@ func validateMetrics(t *testing.T, batchSize int64) {
 	`
 	}
 
-	err = testutil.CollectAndCompare(WriteMessagesCount, strings.NewReader(writeMetadata+writeExpected), "forwarder_write_total")
+	err = testutil.CollectAndCompare(writeMessagesCount, strings.NewReader(writeMetadata+writeExpected), "forwarder_write_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
@@ -1558,16 +1558,16 @@ func validateMetrics(t *testing.T, batchSize int64) {
 		forwarder_ack_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%d", batchSize) + `
 	`
 
-	err = testutil.CollectAndCompare(AckMessagesCount, strings.NewReader(ackMetadata+ackExpected), "forwarder_ack_total")
+	err = testutil.CollectAndCompare(ackMessagesCount, strings.NewReader(ackMetadata+ackExpected), "forwarder_ack_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
 }
 
 func metricsReset() {
-	ReadMessagesCount.Reset()
-	WriteMessagesCount.Reset()
-	AckMessagesCount.Reset()
+	readMessagesCount.Reset()
+	writeMessagesCount.Reset()
+	ackMessagesCount.Reset()
 }
 
 // buildPublisherMap builds OTStore and publisher for each toBuffer
diff --git a/pkg/sources/forward/data_forward_test.go b/pkg/sources/forward/data_forward_test.go
index 5ee1f5e6..1c3e14eb 100644
--- a/pkg/sources/forward/data_forward_test.go
+++ b/pkg/sources/forward/data_forward_test.go
@@ -1162,58 +1162,58 @@ func (f myForwardApplyTransformerErrTest) ApplyMapStream(_ context.Context, _ *i
 
 func validateMetrics(t *testing.T, batchSize int64) {
 	metadata := `
-		# HELP forwarder_read_total Total number of Messages Read
-		# TYPE forwarder_read_total counter
+		# HELP source_forwarder_read_total Total number of Messages Read
+		# TYPE source_forwarder_read_total counter
 		`
 	expected := `
-		forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		source_forwarder_read_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
 	`
 
-	err := testutil.CollectAndCompare(forward.ReadMessagesCount, strings.NewReader(metadata+expected), "forwarder_read_total")
+	err := testutil.CollectAndCompare(readMessagesCount, strings.NewReader(metadata+expected), "source_forwarder_read_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
 
 	writeMetadata := `
-		# HELP forwarder_write_total Total number of Messages Written
-		# TYPE forwarder_write_total counter
+		# HELP source_forwarder_write_total Total number of Messages Written
+		# TYPE source_forwarder_write_total counter
 		`
 	var writeExpected string
 	if batchSize > 1 {
 		writeExpected = `
-		forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
-		forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
+		source_forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
+		source_forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)/2) + `
 	`
 	} else {
 		writeExpected = `
-		forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
-		forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(0)) + `
+		source_forwarder_write_total{partition_name="to1-1",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(batchSize)) + `
+		source_forwarder_write_total{partition_name="to1-2",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%f", float64(0)) + `
 	`
 	}
 
-	err = testutil.CollectAndCompare(forward.WriteMessagesCount, strings.NewReader(writeMetadata+writeExpected), "forwarder_write_total")
+	err = testutil.CollectAndCompare(writeMessagesCount, strings.NewReader(writeMetadata+writeExpected), "source_forwarder_write_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
 
 	ackMetadata := `
-		# HELP forwarder_ack_total Total number of Messages Acknowledged
-		# TYPE forwarder_ack_total counter
+		# HELP source_forwarder_ack_total Total number of Messages Acknowledged
+		# TYPE source_forwarder_ack_total counter
 		`
 	ackExpected := `
-		forwarder_ack_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%d", batchSize) + `
+		source_forwarder_ack_total{partition_name="from",pipeline="testPipeline",vertex="testVertex"} ` + fmt.Sprintf("%d", batchSize) + `
 	`
 
-	err = testutil.CollectAndCompare(forward.AckMessagesCount, strings.NewReader(ackMetadata+ackExpected), "forwarder_ack_total")
+	err = testutil.CollectAndCompare(ackMessagesCount, strings.NewReader(ackMetadata+ackExpected), "source_forwarder_ack_total")
 	if err != nil {
 		t.Errorf("unexpected collecting result:\n%s", err)
 	}
 }
 
 func metricsReset() {
-	forward.ReadMessagesCount.Reset()
-	forward.WriteMessagesCount.Reset()
-	forward.AckMessagesCount.Reset()
+	readMessagesCount.Reset()
+	writeMessagesCount.Reset()
+	ackMessagesCount.Reset()
 }
 
 // buildPublisherMap builds OTStore and publisher for each toBuffer
diff --git a/test/e2e-suite-1/functional_test.go b/test/e2e-suite-1/functional_test.go
index 30c03166..24a05cde 100644
--- a/test/e2e-suite-1/functional_test.go
+++ b/test/e2e-suite-1/functional_test.go
@@ -1,3 +1,5 @@
+//go:build test
+
 /*
 Copyright 2022 The Numaproj Authors.
 
@@ -277,7 +279,6 @@ wmLoop:
 				if err != nil {
 					assert.Fail(s.T(), err.Error())
 				}
-				println(edgeWM)
 				// Watermark propagation can delay, we consider the test as passed as long as the retrieved watermark is greater than the event time of the first message
 				// and less than the current time.
 				assert.True(s.T(), edgeWM >= time.Date(2021, 1, 18, 21, 54, 42, 123000000, time.UTC).UnixMilli() && edgeWM < currentTime)