Repository: numaproj/numaflow
Pull request #743 (2023-05-23 17:11:51)

feat: use metrics to calculate vertex processing rate
@whynowy @numaproj/numaflow-dev , team, although I am publishing it as a draft PR, this change is code-wise ready to be reviewed. The functionality has been tested using a simple pipeline as follow.

![Screenshot 2023-05-16 at 11 42 12 AM](https://github.com/numaproj/numaflow/assets/13165977/c52b53ec-f312-4fa3-986f-7d734f08fda9)

Please review it and leave comments. While I collect feedbacks from PR review, I will be manually testing it with more use cases like reduce, auto-scaling, running on IKS etc.

### More test notes
* Case 1: 1k TPS

![Screenshot 2023-05-23 at 3 13 54 PM](https://github.com/numaproj/numaflow/assets/13165977/89293b1c-a280-48f3-bd90-0cf0bdbecbdc)

* Case 2: auto-scaling

Verified with high throughput, vertex p1 can scale up and down according to the traffic change.

* Case 3: more pods within a vertex

![Screenshot 2023-05-23 at 4 06 55 PM](https://github.com/numaproj/numaflow/assets/13165977/73d853ea-2b84-4c98-a97e-3115a7f30f9f)

* Case 4: reduce pipeline

![Screenshot 2023-05-23 at 4 44 56 PM](https://github.com/numaproj/numaflow/assets/13165977/7d427104-151a-44fb-885b-646b27233407)


<!--

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.

-->


KeranYang: Addressed early comments.

I found one issue during manual tests: the processing rate for reduce vertex showing on UI is incorrect. I created a separate issue to dive deeper into it. https://github.com/numaproj/numaflow/issues/746

This PR is now officially ready for review
whynowy: As a follow up, please also clean up the `rate()` function in ISB readers and writers.
KeranYang: > As a follow up, please also clean up the rate() function in ISB readers and writers.

@whynowy yes, will do when I create PR to remove the `useNewRateCalculation` flag.

Patch:
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index 58430c99..af28f0e6 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -33,6 +33,8 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 
+	server "github.com/numaproj/numaflow/pkg/daemon/server/service/rater"
+
 	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
 	"github.com/numaproj/numaflow/pkg/daemon/server/service"
@@ -87,6 +89,9 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 		}
 	}()
 
+	// rater is used to calculate the processing rate for each of the vertices
+	rater := server.NewRater(ctx, ds.pipeline)
+
 	// Start listener
 	var conn net.Listener
 	var listerErr error
@@ -102,7 +107,7 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 	}
 
 	tlsConfig := &tls.Config{Certificates: []tls.Certificate{*cer}, MinVersion: tls.VersionTLS12}
-	grpcServer, err := ds.newGRPCServer(isbSvcClient, wmFetchers)
+	grpcServer, err := ds.newGRPCServer(isbSvcClient, wmFetchers, rater)
 	if err != nil {
 		return fmt.Errorf("failed to create grpc server: %w", err)
 	}
@@ -119,11 +124,18 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 	go func() { _ = tcpm.Serve() }()
 
 	log.Infof("Daemon server started successfully on %s", address)
+	// Start the rater
+	if err := rater.Start(ctx); err != nil {
+		return fmt.Errorf("failed to start the rater: %w", err)
+	}
 	<-ctx.Done()
 	return nil
 }
 
-func (ds *daemonServer) newGRPCServer(isbSvcClient isbsvc.ISBService, wmFetchers map[string][]fetch.Fetcher) (*grpc.Server, error) {
+func (ds *daemonServer) newGRPCServer(
+	isbSvcClient isbsvc.ISBService,
+	wmFetchers map[string][]fetch.Fetcher,
+	rater *server.Rater) (*grpc.Server, error) {
 	// "Prometheus histograms are a great way to measure latency distributions of your RPCs.
 	// However, since it is a bad practice to have metrics of high cardinality the latency monitoring metrics are disabled by default.
 	// To enable them please call the following in your server initialization code:"
@@ -137,7 +149,7 @@ func (ds *daemonServer) newGRPCServer(isbSvcClient isbsvc.ISBService, wmFetchers
 	}
 	grpcServer := grpc.NewServer(sOpts...)
 	grpc_prometheus.Register(grpcServer)
-	pipelineMetadataQuery, err := service.NewPipelineMetadataQuery(isbSvcClient, ds.pipeline, wmFetchers)
+	pipelineMetadataQuery, err := service.NewPipelineMetadataQuery(isbSvcClient, ds.pipeline, wmFetchers, rater, true)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/daemon/server/service/pipeline_metrics_query.go b/pkg/daemon/server/service/pipeline_metrics_query.go
index 8258a71b..181c4048 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query.go
@@ -29,6 +29,8 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
 
+	server "github.com/numaproj/numaflow/pkg/daemon/server/service/rater"
+
 	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
 	"github.com/numaproj/numaflow/pkg/isbsvc"
@@ -49,6 +51,11 @@ type pipelineMetadataQuery struct {
 	pipeline          *v1alpha1.Pipeline
 	httpClient        metricsHttpClient
 	watermarkFetchers map[string][]fetch.Fetcher
+	rater             *server.Rater
+
+	// temporary flag to help with testing new rate calculation approach
+	// TODO - remove this flag once we fully migrate to new rate calculation
+	useNewRateCalculation bool
 }
 
 const (
@@ -58,18 +65,25 @@ const (
 )
 
 // NewPipelineMetadataQuery returns a new instance of pipelineMetadataQuery
-func NewPipelineMetadataQuery(isbSvcClient isbsvc.ISBService, pipeline *v1alpha1.Pipeline, wmFetchers map[string][]fetch.Fetcher) (*pipelineMetadataQuery, error) {
+func NewPipelineMetadataQuery(
+	isbSvcClient isbsvc.ISBService,
+	pipeline *v1alpha1.Pipeline,
+	wmFetchers map[string][]fetch.Fetcher,
+	rater *server.Rater,
+	useNewRateCalculation bool) (*pipelineMetadataQuery, error) {
 	var err error
 	ps := pipelineMetadataQuery{
-		isbSvcClient:      isbSvcClient,
-		pipeline:          pipeline,
-		watermarkFetchers: wmFetchers,
+		isbSvcClient: isbSvcClient,
+		pipeline:     pipeline,
 		httpClient: &http.Client{
 			Transport: &http.Transport{
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 			},
 			Timeout: time.Second * 3,
 		},
+		watermarkFetchers:     wmFetchers,
+		rater:                 rater,
+		useNewRateCalculation: useNewRateCalculation,
 	}
 	if err != nil {
 		return nil, err
@@ -169,6 +183,12 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 		podNum = int64(*obj[0].DeprecatedParallelism)
 	}
 
+	var vertexLevelRates map[string]float64
+
+	if ps.useNewRateCalculation {
+		vertexLevelRates = ps.rater.GetRates(req.GetVertex())
+	}
+
 	// Get the headless service name
 	headlessServiceName := vertex.GetHeadlessServiceName()
 
@@ -179,7 +199,7 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 		pendings := make(map[string]int64, 0)
 
 		// We can query the metrics endpoint of the (i)th pod to obtain this value.
-		// example for 0th pod : https://simple-pipeline-in-0.simple-pipeline-in-headless.svc.cluster.local:2469/metrics
+		// example for 0th pod : https://simple-pipeline-in-0.simple-pipeline-in-headless.default.svc.cluster.local:2469/metrics
 		url := fmt.Sprintf("https://%s-%v.%s.%s.svc.cluster.local:%v/metrics", vertexName, i, headlessServiceName, ps.pipeline.Namespace, v1alpha1.VertexMetricsPort)
 		if res, err := ps.httpClient.Get(url); err != nil {
 			log.Debugf("Error reading the metrics endpoint, it might be because of vertex scaling down to 0: %f", err.Error())
@@ -196,15 +216,17 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 				return nil, err
 			}
 
-			// Check if the resultant metrics list contains the processingRate, if it does look for the period label
-			if value, ok := result[metrics.VertexProcessingRate]; ok {
-				metricsList := value.GetMetric()
-				for _, metric := range metricsList {
-					labels := metric.GetLabel()
-					for _, label := range labels {
-						if label.GetName() == metrics.LabelPeriod {
-							lookback := label.GetValue()
-							processingRates[lookback] = metric.Gauge.GetValue()
+			if !ps.useNewRateCalculation {
+				// Check if the resultant metrics list contains the processingRate, if it does look for the period label
+				if value, ok := result[metrics.VertexProcessingRate]; ok {
+					metricsList := value.GetMetric()
+					for _, metric := range metricsList {
+						labels := metric.GetLabel()
+						for _, label := range labels {
+							if label.GetName() == metrics.LabelPeriod {
+								lookback := label.GetValue()
+								processingRates[lookback] = metric.Gauge.GetValue()
+							}
 						}
 					}
 				}
@@ -223,11 +245,20 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 				}
 			}
 
-			metricsArr[i] = &daemon.VertexMetrics{
-				Pipeline:        &ps.pipeline.Name,
-				Vertex:          req.Vertex,
-				ProcessingRates: processingRates,
-				Pendings:        pendings,
+			if !ps.useNewRateCalculation {
+				metricsArr[i] = &daemon.VertexMetrics{
+					Pipeline:        &ps.pipeline.Name,
+					Vertex:          req.Vertex,
+					ProcessingRates: processingRates,
+					Pendings:        pendings,
+				}
+			} else {
+				metricsArr[i] = &daemon.VertexMetrics{
+					Pipeline:        &ps.pipeline.Name,
+					Vertex:          req.Vertex,
+					ProcessingRates: vertexLevelRates,
+					Pendings:        pendings,
+				}
 			}
 		}
 	}
@@ -259,16 +290,26 @@ func (ps *pipelineMetadataQuery) GetPipelineStatus(ctx context.Context, req *dae
 
 		// may need to revisit later, another concern could be that the processing rate is too slow instead of just 0
 		for _, vertexMetrics := range vertexResp.VertexMetrics {
-			if pending, ok := vertexMetrics.GetPendings()["default"]; ok {
-				if processingRate, ok := vertexMetrics.GetProcessingRates()["default"]; ok {
-					if pending > 0 && processingRate == 0 {
-						resp.Status = &daemon.PipelineStatus{
-							Status:  pointer.String(PipelineStatusError),
-							Message: pointer.String(fmt.Sprintf("Pipeline has an error. Vertex %s is not processing pending messages.", vertex.Name)),
-						}
-						return resp, nil
-					}
+			var pending int64
+			var processingRate float64
+			if p, ok := vertexMetrics.GetPendings()["default"]; ok {
+				pending = p
+			} else {
+				continue
+			}
+
+			if p, ok := vertexMetrics.GetProcessingRates()["default"]; ok {
+				processingRate = p
+			} else {
+				continue
+			}
+
+			if pending > 0 && processingRate == 0 {
+				resp.Status = &daemon.PipelineStatus{
+					Status:  pointer.String(PipelineStatusError),
+					Message: pointer.String(fmt.Sprintf("Pipeline has an error. Vertex %s is not processing pending messages.", vertex.Name)),
 				}
+				return resp, nil
 			}
 		}
 	}
diff --git a/pkg/daemon/server/service/rater/helper.go b/pkg/daemon/server/service/rater/helper.go
new file mode 100644
index 00000000..1cdb7a3c
--- /dev/null
+++ b/pkg/daemon/server/service/rater/helper.go
@@ -0,0 +1,88 @@
+/*
+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 server
+
+import (
+	"time"
+
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
+)
+
+// UpdateCount updates the count of processed messages for a pod at a given time
+func UpdateCount(q *sharedqueue.OverflowQueue[*TimestampedCounts], time int64, podName string, count float64) {
+	// find the element matching the input timestamp and update it
+	for _, i := range q.Items() {
+		if i.timestamp == time {
+			i.Update(podName, count)
+			return
+		}
+	}
+	// if we cannot find a matching element, it means we need to add a new timestamped count to the queue
+	tc := NewTimestampedCounts(time)
+	tc.Update(podName, count)
+	q.Append(tc)
+}
+
+// CalculateRate calculates the rate of the vertex in the last lookback seconds
+func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) float64 {
+	n := q.Length()
+	if n <= 1 {
+		return 0
+	}
+	now := time.Now().Truncate(CountWindow).Unix()
+	counts := q.Items()
+	var startIndex int
+	startCountInfo := counts[n-2]
+	if now-startCountInfo.timestamp > lookbackSeconds {
+		return 0
+	}
+	for i := n - 2; i >= 0; i-- {
+		if now-counts[i].timestamp <= lookbackSeconds {
+			startIndex = i
+		} else {
+			break
+		}
+	}
+	delta := float64(0)
+	// time diff in seconds.
+	timeDiff := counts[n-1].timestamp - counts[startIndex].timestamp
+	for i := startIndex; i < n-1; i++ {
+		delta = delta + calculateDelta(counts[i], counts[i+1])
+	}
+	return delta / float64(timeDiff)
+}
+
+func calculateDelta(c1, c2 *TimestampedCounts) float64 {
+	tc1 := c1.Snapshot()
+	tc2 := c2.Snapshot()
+	delta := float64(0)
+	// Iterate over the podCounts of the second TimestampedCounts
+	for pod, count2 := range tc2 {
+		// If the pod also exists in the first TimestampedCounts
+		if count1, ok := tc1[pod]; ok {
+			// If the count has decreased, it means the pod restarted
+			if count2 < count1 {
+				delta += count2
+			} else { // If the count has increased or stayed the same
+				delta += count2 - count1
+			}
+		} else { // If the pod only exists in the second TimestampedCounts, it's a new pod
+			delta += count2
+		}
+	}
+	return delta
+}
diff --git a/pkg/daemon/server/service/rater/options.go b/pkg/daemon/server/service/rater/options.go
new file mode 100644
index 00000000..faa8bf35
--- /dev/null
+++ b/pkg/daemon/server/service/rater/options.go
@@ -0,0 +1,52 @@
+/*
+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 server
+
+type options struct {
+	// Number of workers working on collecting counts of processed messages.
+	workers int
+	// Time in milliseconds, each element in the work queue will be picked up in an interval of this period of time.
+	taskInterval int
+}
+
+type Option func(*options)
+
+func defaultOptions() *options {
+	// A simple example of how these numbers work together:
+	// Assuming we have 200 tasks, we have 20 workers, each worker will be responsible for approximately 10 tasks during one iteration.
+	// The task interval is 10 seconds, which means each task need to be picked up by a worker every 10 seconds.
+	// Hence a worker needs to finish processing 1 task in 1 second.
+	// Translating to numaflow language, for a 200-pod pipeline, a worker needs to finish scraping 1 pod in 1 second, which is a reasonable number.
+	// TODO - need to adjust the numbers once we have a better idea of the performance
+	return &options{
+		workers: 20,
+		// 10 seconds
+		taskInterval: 10000,
+	}
+}
+
+func WithWorkers(n int) Option {
+	return func(o *options) {
+		o.workers = n
+	}
+}
+
+func WithTaskInterval(n int) Option {
+	return func(o *options) {
+		o.taskInterval = n
+	}
+}
diff --git a/pkg/daemon/server/service/rater/pod_tracker.go b/pkg/daemon/server/service/rater/pod_tracker.go
new file mode 100644
index 00000000..0684f740
--- /dev/null
+++ b/pkg/daemon/server/service/rater/pod_tracker.go
@@ -0,0 +1,131 @@
+/*
+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 server
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+)
+
+// PodInfoSeparator is used as a separator such that the worker can split the pod key
+// to get the pipeline name, vertex name, pod index and vertex type.
+// "*" is chosen because it is not allowed in all the above fields.
+const PodInfoSeparator = "*"
+
+// PodTracker maintains a set of active pods for a pipeline
+// It periodically sends http requests to pods to check if they are still active
+type PodTracker struct {
+	pipeline        *v1alpha1.Pipeline
+	log             *zap.SugaredLogger
+	httpClient      metricsHttpClient
+	activePods      *UniqueStringList
+	refreshInterval time.Duration
+}
+
+func NewPodTracker(ctx context.Context, p *v1alpha1.Pipeline, opts ...PodTrackerOption) *PodTracker {
+	pt := PodTracker{
+		pipeline: p,
+		log:      logging.FromContext(ctx).Named("PodTracker"),
+		httpClient: &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			},
+			Timeout: time.Second,
+		},
+		activePods:      NewUniqueStringList(),
+		refreshInterval: 30 * time.Second, // Default refresh interval for updating active pod set
+	}
+
+	for _, opt := range opts {
+		if opt != nil {
+			opt(&pt)
+		}
+	}
+	return &pt
+}
+
+type PodTrackerOption func(*PodTracker)
+
+// WithRefreshInterval sets how often to refresh the rate metrics.
+func WithRefreshInterval(d time.Duration) PodTrackerOption {
+	return func(r *PodTracker) {
+		r.refreshInterval = d
+	}
+}
+
+func (pt *PodTracker) Start(ctx context.Context) error {
+	pt.log.Infof("Starting tracking active pods for pipeline %s...", pt.pipeline.Name)
+	go func() {
+		ticker := time.NewTicker(pt.refreshInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				for _, v := range pt.pipeline.Spec.Vertices {
+					var vType string
+					if v.IsReduceUDF() {
+						vType = "reduce"
+					} else {
+						vType = "non_reduce"
+					}
+					for i := 0; i < int(v.Scale.GetMaxReplicas()); i++ {
+						podName := fmt.Sprintf("%s-%s-%d", pt.pipeline.Name, v.Name, i)
+						// podKey is used as a unique identifier for the pod, it is used by worker to determine the count of processed messages of the pod.
+						podKey := strings.Join([]string{pt.pipeline.Name, v.Name, fmt.Sprintf("%d", i), vType}, PodInfoSeparator)
+						if pt.isActive(v.Name, podName) {
+							pt.activePods.PushBack(podKey)
+						} else {
+							pt.activePods.Remove(podKey)
+							// we assume all the pods are ordered with continuous indices, hence as we keep increasing the index, if we don't find one, we can stop looking.
+							// the assumption holds because when we scale down, we always scale down from the last pod.
+							// there can be a case when a pod in the middle crashes, causing us missing counting the following pods.
+							// such case is rare and if it happens, it can lead to lower rate then the real one. It is acceptable because it will recover when the crashed pod is restarted.
+							break
+						}
+					}
+				}
+				pt.log.Debugf("Finished updating the active pod set: %v", pt.activePods.ToString())
+			}
+		}
+	}()
+	return nil
+}
+
+func (pt *PodTracker) GetActivePods() *UniqueStringList {
+	return pt.activePods
+}
+
+func (pt *PodTracker) isActive(vertexName, podName string) bool {
+	// using the vertex headless service to check if a pod exists or not.
+	// example for 0th pod : https://simple-pipeline-in-0.simple-pipeline-in-headless.default.svc.cluster.local:2469/metrics
+	url := fmt.Sprintf("https://%s.%s.%s.svc.cluster.local:%v/metrics", podName, pt.pipeline.Name+"-"+vertexName+"-headless", pt.pipeline.Namespace, v1alpha1.VertexMetricsPort)
+	if _, err := pt.httpClient.Head(url); err != nil {
+		return false
+	}
+	return true
+}
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
new file mode 100644
index 00000000..0cb66eaf
--- /dev/null
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -0,0 +1,247 @@
+/*
+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 server
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/prometheus/common/expfmt"
+	"go.uber.org/zap"
+
+	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
+)
+
+// CountWindow is the time window for which we maintain the timestamped counts, currently 10 seconds
+// e.g. if the current time is 12:00:07, the retrieved count will be tracked in the 12:00:00-12:00:10 time window using 12:00:10 as the timestamp
+const CountWindow = time.Second * 10
+
+// metricsHttpClient interface for the GET/HEAD call to metrics endpoint.
+// Had to add this an interface for testing
+type metricsHttpClient interface {
+	Get(url string) (*http.Response, error)
+	Head(url string) (*http.Response, error)
+}
+
+// fixedLookbackSeconds always maintain rate metrics for the following lookback seconds (1m, 5m, 15m)
+var fixedLookbackSeconds = map[string]int64{"1m": 60, "5m": 300, "15m": 900}
+
+// Rater is a struct that maintains information about the processing rate of each vertex.
+// It monitors the number of processed messages for each pod in a vertex and calculates the rate.
+type Rater struct {
+	pipeline   *v1alpha1.Pipeline
+	httpClient metricsHttpClient
+	log        *zap.SugaredLogger
+	podTracker *PodTracker
+	// timestampedPodCounts is a map between vertex name and a queue of timestamped counts for that vertex
+	timestampedPodCounts map[string]*sharedqueue.OverflowQueue[*TimestampedCounts]
+	options              *options
+}
+
+func NewRater(ctx context.Context, p *v1alpha1.Pipeline, opts ...Option) *Rater {
+	rater := Rater{
+		pipeline: p,
+		httpClient: &http.Client{
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			},
+			Timeout: time.Second * 1,
+		},
+		log:                  logging.FromContext(ctx).Named("Rater"),
+		timestampedPodCounts: make(map[string]*sharedqueue.OverflowQueue[*TimestampedCounts]),
+		options:              defaultOptions(),
+	}
+
+	rater.podTracker = NewPodTracker(ctx, p)
+	for _, v := range p.Spec.Vertices {
+		// maintain the total counts of the last 30 minutes since we support 1m, 5m, 15m lookback seconds.
+		rater.timestampedPodCounts[v.Name] = sharedqueue.New[*TimestampedCounts](1800)
+	}
+
+	for _, opt := range opts {
+		if opt != nil {
+			opt(rater.options)
+		}
+	}
+	return &rater
+}
+
+// Function monitor() defines each of the worker's job.
+// It waits for keys in the channel, and starts a monitoring job
+func (r *Rater) monitor(ctx context.Context, id int, keyCh <-chan string) {
+	r.log.Infof("Started monitoring worker %v", id)
+	for {
+		select {
+		case <-ctx.Done():
+			r.log.Infof("Stopped monitoring worker %v", id)
+			return
+		case key := <-keyCh:
+			if err := r.monitorOnePod(ctx, key, id); err != nil {
+				r.log.Errorw("Failed to monitor a pod", zap.String("pod", key), zap.Error(err))
+			}
+		}
+	}
+}
+
+func (r *Rater) monitorOnePod(ctx context.Context, key string, worker int) error {
+	log := logging.FromContext(ctx).With("worker", fmt.Sprint(worker)).With("podKey", key)
+	log.Debugf("Working on key: %s", key)
+	podInfo := strings.Split(key, PodInfoSeparator)
+	if len(podInfo) != 4 {
+		return fmt.Errorf("invalid key %q", key)
+	}
+	vertexName := podInfo[1]
+	vertexType := podInfo[3]
+	podName := strings.Join([]string{podInfo[0], podInfo[1], podInfo[2]}, "-")
+	var count float64
+	activePods := r.podTracker.GetActivePods()
+	if activePods.Contains(key) {
+		count = r.getTotalCount(vertexName, vertexType, podName)
+		if count == CountNotAvailable {
+			log.Debugf("Pod %s does not exist, removing it from the active pod list...", podName)
+			activePods.Remove(key)
+		}
+	} else {
+		log.Debugf("Pod %s does not exist, updating it with CountNotAvailable...", podName)
+		count = CountNotAvailable
+	}
+	now := time.Now().Add(CountWindow).Truncate(CountWindow).Unix()
+	UpdateCount(r.timestampedPodCounts[vertexName], now, podName, count)
+	return nil
+}
+
+func (r *Rater) Start(ctx context.Context) error {
+	r.log.Infof("Starting rater...")
+	keyCh := make(chan string)
+	ctx, cancel := context.WithCancel(logging.WithLogger(ctx, r.log))
+	defer cancel()
+
+	go func() {
+		err := r.podTracker.Start(ctx)
+		if err != nil {
+			r.log.Errorw("Failed to start pod tracker", zap.Error(err))
+		}
+	}()
+
+	// Worker group
+	for i := 1; i <= r.options.workers; i++ {
+		go r.monitor(ctx, i, keyCh)
+	}
+
+	// Function assign() moves an element in the list from the front to the back,
+	// and send to the channel so that it can be picked up by a worker.
+	assign := func() {
+		activePods := r.podTracker.GetActivePods()
+		if e := activePods.Front(); e != "" {
+			activePods.MoveToBack(e)
+			keyCh <- e
+			return
+		}
+	}
+
+	// Following for loop keeps calling assign() function to assign monitoring tasks to the workers.
+	// It makes sure each element in the list will be assigned every N milliseconds.
+	for {
+		select {
+		case <-ctx.Done():
+			r.log.Info("Shutting down monitoring job assigner")
+			return nil
+		default:
+			assign()
+			// Make sure each of the key will be assigned at least every taskInterval milliseconds.
+			sleep(ctx, time.Millisecond*time.Duration(func() int {
+				l := r.podTracker.GetActivePods().Length()
+				if l == 0 {
+					return r.options.taskInterval
+				}
+				result := r.options.taskInterval / l
+				if result > 0 {
+					return result
+				}
+				return 1
+			}()))
+		}
+	}
+}
+
+// sleep function uses a select statement to check if the context is cancelled before sleeping for the given duration
+// it helps ensure the sleep will be released when the context is cancelled, allowing the goroutine to exit gracefully
+func sleep(ctx context.Context, duration time.Duration) {
+	select {
+	case <-ctx.Done():
+	case <-time.After(duration):
+	}
+}
+
+// getTotalCount returns the total number of messages read by the pod
+func (r *Rater) getTotalCount(vertexName, vertexType, podName string) float64 {
+	// scrape the read total metric from pod metric port
+	url := fmt.Sprintf("https://%s.%s.%s.svc.cluster.local:%v/metrics", podName, r.pipeline.Name+"-"+vertexName+"-headless", r.pipeline.Namespace, v1alpha1.VertexMetricsPort)
+	if res, err := r.httpClient.Get(url); err != nil {
+		r.log.Errorf("failed reading the metrics endpoint, %v", err.Error())
+		return CountNotAvailable
+	} else {
+		textParser := expfmt.TextParser{}
+		result, err := textParser.TextToMetricFamilies(res.Body)
+		if err != nil {
+			r.log.Errorf("failed parsing to prometheus metric families, %v", err.Error())
+			return CountNotAvailable
+		}
+		var readTotalMetricName string
+		if vertexType == "reduce" {
+			readTotalMetricName = "reduce_isb_reader_read_total"
+		} else {
+			readTotalMetricName = "forwarder_read_total"
+		}
+		if value, ok := result[readTotalMetricName]; ok && value != nil && len(value.GetMetric()) > 0 {
+			metricsList := value.GetMetric()
+			return metricsList[0].Counter.GetValue()
+		} else {
+			return CountNotAvailable
+		}
+	}
+}
+
+// GetRates returns the processing rates of the vertex in the format of lookback second to rate mappings
+func (r *Rater) GetRates(vertexName string) map[string]float64 {
+	var result = make(map[string]float64)
+	// get the user-specified lookback seconds from the pipeline spec
+	var userSpecifiedLookBackSeconds int64
+	// TODO - we can keep a local copy of vertex to lookback seconds mapping to avoid iterating the pipeline spec all the time.
+	for _, v := range r.pipeline.Spec.Vertices {
+		if v.Name == vertexName {
+			userSpecifiedLookBackSeconds = int64(v.Scale.GetLookbackSeconds())
+		}
+	}
+	lookbackSecondsMap := map[string]int64{"default": userSpecifiedLookBackSeconds}
+	for k, v := range fixedLookbackSeconds {
+		lookbackSecondsMap[k] = v
+	}
+
+	// calculate rates for each lookback seconds
+	for n, i := range lookbackSecondsMap {
+		r := CalculateRate(r.timestampedPodCounts[vertexName], i)
+		result[n] = r
+	}
+	return result
+}
diff --git a/pkg/daemon/server/service/rater/timestamped_counts.go b/pkg/daemon/server/service/rater/timestamped_counts.go
new file mode 100644
index 00000000..99ef9b2f
--- /dev/null
+++ b/pkg/daemon/server/service/rater/timestamped_counts.go
@@ -0,0 +1,64 @@
+/*
+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 server
+
+import (
+	"sync"
+)
+
+// CountNotAvailable indicates the rater is not able to collect the count of processed messages for a pod
+// normally it is because the pod is not running
+const CountNotAvailable = -1
+
+// TimestampedCounts track the total count of processed messages for a list of pods at a given timestamp
+type TimestampedCounts struct {
+	// timestamp in seconds, is the time when the count is recorded
+	timestamp int64
+	// podName to count mapping
+	podCounts map[string]float64
+	lock      *sync.RWMutex
+}
+
+func NewTimestampedCounts(t int64) *TimestampedCounts {
+	return &TimestampedCounts{
+		timestamp: t,
+		podCounts: make(map[string]float64),
+		lock:      new(sync.RWMutex),
+	}
+}
+
+func (tc *TimestampedCounts) Update(podName string, count float64) {
+	tc.lock.Lock()
+	defer tc.lock.Unlock()
+	if count == CountNotAvailable {
+		delete(tc.podCounts, podName)
+		return
+	}
+	tc.podCounts[podName] = count
+}
+
+// Snapshot returns a copy of the podName to count mapping
+// it's used to ensure the returned map is not modified by other goroutines
+func (tc *TimestampedCounts) Snapshot() map[string]float64 {
+	tc.lock.RLock()
+	defer tc.lock.RUnlock()
+	counts := make(map[string]float64)
+	for k, v := range tc.podCounts {
+		counts[k] = v
+	}
+	return counts
+}
diff --git a/pkg/daemon/server/service/rater/uniq_str_list.go b/pkg/daemon/server/service/rater/uniq_str_list.go
new file mode 100644
index 00000000..599feff9
--- /dev/null
+++ b/pkg/daemon/server/service/rater/uniq_str_list.go
@@ -0,0 +1,102 @@
+/*
+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 server
+
+import (
+	"container/list"
+	"sync"
+)
+
+// UniqueStringList is a list of strings that only allows unique values.
+// the underlying list is a doubly linked list.
+type UniqueStringList struct {
+	l    *list.List
+	m    map[string]*list.Element
+	lock *sync.RWMutex
+}
+
+func NewUniqueStringList() *UniqueStringList {
+	return &UniqueStringList{
+		l:    list.New(),
+		m:    make(map[string]*list.Element),
+		lock: new(sync.RWMutex),
+	}
+}
+
+// PushBack adds a value to the back of the list, if the value doesn't exist in the list.
+func (l *UniqueStringList) PushBack(value string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	if _, ok := l.m[value]; !ok {
+		l.m[value] = l.l.PushBack(value)
+	}
+}
+
+// MoveToBack moves the element to the back of the list, if the value exists in the list.
+func (l *UniqueStringList) MoveToBack(value string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	if e, ok := l.m[value]; ok {
+		l.l.MoveToBack(e)
+	}
+}
+
+// Front returns the first string of list l or empty string if the list is empty.
+func (l *UniqueStringList) Front() string {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	if l.Length() == 0 {
+		return ""
+	}
+	return l.l.Front().Value.(string)
+}
+
+// Length returns the number of elements of list l.
+func (l *UniqueStringList) Length() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return l.l.Len()
+}
+
+// Contains returns true if the value exists in the list.
+func (l *UniqueStringList) Contains(value string) bool {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	_, ok := l.m[value]
+	return ok
+}
+
+// Remove removes the string from the list, if it exists in the list.
+func (l *UniqueStringList) Remove(value string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	if e, ok := l.m[value]; ok {
+		l.l.Remove(e)
+		delete(l.m, value)
+	}
+}
+
+// ToString returns a comma separated string of the list values.
+func (l *UniqueStringList) ToString() string {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	var s string
+	for e := l.l.Front(); e != nil; e = e.Next() {
+		s += e.Value.(string) + ","
+	}
+	return s
+}
diff --git a/pkg/shared/logging/log.go b/pkg/shared/logging/log.go
index a2af7489..cc1226d2 100644
--- a/pkg/shared/logging/log.go
+++ b/pkg/shared/logging/log.go
@@ -21,6 +21,7 @@ import (
 	"os"
 
 	zap "go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 )
 
 // NewLogger returns a new zap.SugaredLogger
@@ -33,6 +34,7 @@ func NewLogger() *zap.SugaredLogger {
 		config = zap.NewProductionConfig()
 	}
 	// Config customization goes here if any
+	config.EncoderConfig.EncodeTime = zapcore.RFC3339NanoTimeEncoder
 	config.OutputPaths = []string{"stdout"}
 	logger, err := config.Build()
 	if err != nil {


Test patch:
diff --git a/pkg/daemon/server/service/pipeline_metrics_query_test.go b/pkg/daemon/server/service/pipeline_metrics_query_test.go
index e73bb3a8..5aa59346 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query_test.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query_test.go
@@ -78,7 +78,7 @@ func TestGetVertexMetrics(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{Name: pipelineName},
 	}
 	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, nil, false)
 	assert.NoError(t, err)
 
 	metricsResponse := `# HELP vertex_processing_rate Message processing rate in the last period of seconds, tps. It represents the rate of a vertex instead of a pod.
@@ -153,7 +153,7 @@ func TestGetBuffer(t *testing.T) {
 	}
 
 	ms := &mockIsbSvcClient{}
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil, false)
 	assert.NoError(t, err)
 
 	bufferName := "numaflow-system-simple-pipeline-cat-0"
@@ -194,7 +194,7 @@ func TestListBuffers(t *testing.T) {
 	}
 
 	ms := &mockIsbSvcClient{}
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil, false)
 	assert.NoError(t, err)
 
 	req := &daemon.ListBuffersRequest{Pipeline: &pipelineName}
@@ -217,7 +217,7 @@ func TestGetPipelineStatus(t *testing.T) {
 		},
 	}
 	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, nil, false)
 	assert.NoError(t, err)
 
 	OKPipelineResponse := daemon.PipelineStatus{Status: pointer.String("OK"), Message: pointer.String("Pipeline has no issue.")}
diff --git a/pkg/daemon/server/service/rater/helper_test.go b/pkg/daemon/server/service/rater/helper_test.go
new file mode 100644
index 00000000..9cbb5e6d
--- /dev/null
+++ b/pkg/daemon/server/service/rater/helper_test.go
@@ -0,0 +1,247 @@
+/*
+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 server
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
+)
+
+const TestTime = 1620000000
+
+func TestUpdateCount(t *testing.T) {
+	t.Run("givenTimeExistsPodExistsCountAvailable_whenUpdate_thenUpdatePodCount", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 10.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime, "pod1", 20.0)
+
+		assert.Equal(t, 1, q.Length())
+		assert.Equal(t, 20.0, q.Items()[0].podCounts["pod1"])
+	})
+
+	t.Run("givenTimeExistsPodNotExistsCountAvailable_whenUpdate_thenAddPodCount", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 20.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime, "pod2", 10.0)
+
+		assert.Equal(t, 1, q.Length())
+		assert.Equal(t, 20.0, q.Items()[0].podCounts["pod1"])
+		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod2"])
+	})
+
+	t.Run("givenTimeExistsPodExistsCountNotAvailable_whenUpdate_thenRemovePod", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 10.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime, "pod1", CountNotAvailable)
+
+		assert.Equal(t, 1, q.Length())
+		assert.Equal(t, 0, len(q.Items()[0].podCounts))
+	})
+
+	t.Run("givenTimeExistsPodNotExistsCountNotAvailable_whenUpdate_thenNoUpdate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 10.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime, "pod2", CountNotAvailable)
+
+		assert.Equal(t, 1, q.Length())
+		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod1"])
+	})
+
+	t.Run("givenTimeNotExistsCountAvailable_whenUpdate_thenUpdateNewTimeWithPod", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 10.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime+1, "pod1", 20.0)
+
+		assert.Equal(t, 2, q.Length())
+		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod1"])
+		assert.Equal(t, 20.0, q.Items()[1].podCounts["pod1"])
+	})
+
+	t.Run("givenTimeNotExistsCountNotAvailable_whenUpdate_thenAddEmptyItem", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		tc := NewTimestampedCounts(TestTime)
+		tc.Update("pod1", 10.0)
+		q.Append(tc)
+
+		UpdateCount(q, TestTime+1, "pod2", CountNotAvailable)
+
+		assert.Equal(t, 2, q.Length())
+		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod1"])
+		assert.Equal(t, 0, len(q.Items()[1].podCounts))
+	})
+}
+
+func TestCalculateRate(t *testing.T) {
+	t.Run("givenCollectedTimeLessThanTwo_whenCalculateRate_thenReturnZero", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		rate := CalculateRate(q, 10)
+		assert.Equal(t, 0.0, rate)
+	})
+
+	t.Run("singlePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		now := time.Now()
+
+		tc1 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 20)
+		tc1.Update("pod1", 5.0)
+		q.Append(tc1)
+		tc2 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 10)
+		tc2.Update("pod1", 10.0)
+		q.Append(tc2)
+		tc3 := NewTimestampedCounts(now.Truncate(CountWindow).Unix())
+		tc3.Update("pod1", 20.0)
+		q.Append(tc3)
+
+		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, 1.0, CalculateRate(q, 15))
+		assert.Equal(t, 0.75, CalculateRate(q, 25))
+		assert.Equal(t, 0.75, CalculateRate(q, 100))
+	})
+
+	t.Run("singlePod_givenCountDecreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		now := time.Now()
+
+		tc1 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 30)
+		tc1.Update("pod1", 200.0)
+		q.Append(tc1)
+		tc2 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 20)
+		tc2.Update("pod1", 100.0)
+		q.Append(tc2)
+		tc3 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 10)
+		tc3.Update("pod1", 50.0)
+		q.Append(tc3)
+		tc4 := NewTimestampedCounts(now.Truncate(CountWindow).Unix())
+		tc4.Update("pod1", 80.0)
+		q.Append(tc4)
+
+		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, 3.0, CalculateRate(q, 15))
+		assert.Equal(t, 4.0, CalculateRate(q, 25))
+		assert.Equal(t, 6.0, CalculateRate(q, 35))
+		assert.Equal(t, 6.0, CalculateRate(q, 100))
+	})
+
+	t.Run("multiplePods_givenCountIncreasesAndDecreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		now := time.Now()
+
+		tc1 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 30)
+		tc1.Update("pod1", 200.0)
+		tc1.Update("pod2", 100.0)
+		q.Append(tc1)
+		tc2 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 20)
+		tc2.Update("pod1", 100.0)
+		tc2.Update("pod2", 200.0)
+		q.Append(tc2)
+		tc3 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 10)
+		tc3.Update("pod1", 50.0)
+		tc3.Update("pod2", 300.0)
+		q.Append(tc3)
+		tc4 := NewTimestampedCounts(now.Truncate(CountWindow).Unix())
+		tc4.Update("pod1", 80.0)
+		tc4.Update("pod2", 400.0)
+		q.Append(tc4)
+
+		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, 13.0, CalculateRate(q, 15))
+		assert.Equal(t, 14.0, CalculateRate(q, 25))
+		assert.Equal(t, 16.0, CalculateRate(q, 35))
+		assert.Equal(t, 16.0, CalculateRate(q, 100))
+	})
+
+	t.Run("multiplePods_givenPodsComeAndGo_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](1800)
+		now := time.Now()
+
+		tc1 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 30)
+		tc1.Update("pod1", 200.0)
+		tc1.Update("pod2", 90.0)
+		tc1.Update("pod3", 50.0)
+		q.Append(tc1)
+		tc2 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 20)
+		tc2.Update("pod1", 100.0)
+		tc2.Update("pod2", 200.0)
+		q.Append(tc2)
+		tc3 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 10)
+		tc3.Update("pod1", 50.0)
+		tc3.Update("pod2", 300.0)
+		tc3.Update("pod4", 100.0)
+		q.Append(tc3)
+		tc4 := NewTimestampedCounts(now.Truncate(CountWindow).Unix())
+		tc4.Update("pod2", 400.0)
+		tc4.Update("pod3", 200.0)
+		tc4.Update("pod100", 200.0)
+		q.Append(tc4)
+
+		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, 50.0, CalculateRate(q, 15))
+		assert.Equal(t, 37.5, CalculateRate(q, 25))
+		assert.Equal(t, 32.0, CalculateRate(q, 35))
+		assert.Equal(t, 32.0, CalculateRate(q, 100))
+	})
+
+	t.Run("queueOverflowed_SinglePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+		q := sharedqueue.New[*TimestampedCounts](3)
+		now := time.Now()
+
+		tc1 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 30)
+		tc1.Update("pod1", 200.0)
+		tc1.Update("pod2", 90.0)
+		tc1.Update("pod3", 50.0)
+		q.Append(tc1)
+		tc2 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 20)
+		tc2.Update("pod1", 100.0)
+		tc2.Update("pod2", 200.0)
+		q.Append(tc2)
+		tc3 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 10)
+		tc3.Update("pod1", 50.0)
+		tc3.Update("pod2", 300.0)
+		tc3.Update("pod4", 100.0)
+		q.Append(tc3)
+		tc4 := NewTimestampedCounts(now.Truncate(CountWindow).Unix())
+		tc4.Update("pod2", 400.0)
+		tc4.Update("pod3", 200.0)
+		tc4.Update("pod100", 200.0)
+		q.Append(tc4)
+
+		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, 50.0, CalculateRate(q, 15))
+		assert.Equal(t, 37.5, CalculateRate(q, 25))
+		assert.Equal(t, 37.5, CalculateRate(q, 35))
+		assert.Equal(t, 37.5, CalculateRate(q, 100))
+	})
+}
diff --git a/pkg/daemon/server/service/rater/rater_test.go b/pkg/daemon/server/service/rater/rater_test.go
new file mode 100644
index 00000000..74e720d0
--- /dev/null
+++ b/pkg/daemon/server/service/rater/rater_test.go
@@ -0,0 +1,139 @@
+/*
+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 server
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/goleak"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+)
+
+type mockHttpClient struct {
+	podOneCount int64
+	podTwoCount int64
+	lock        *sync.RWMutex
+}
+
+func (m *mockHttpClient) Get(url string) (*http.Response, error) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if url == "https://p-v-0.p-v-headless.default.svc.cluster.local:2469/metrics" {
+		m.podOneCount = m.podOneCount + 20
+		resp := &http.Response{
+			StatusCode: 200,
+			Body: io.NopCloser(bytes.NewReader([]byte(fmt.Sprintf(`
+# HELP forwarder_read_total Total number of Messages Read
+# TYPE forwarder_read_total counter
+forwarder_read_total{buffer="input",pipeline="simple-pipeline",vertex="input"} %d
+`, m.podOneCount))))}
+		return resp, nil
+	} else if url == "https://p-v-1.p-v-headless.default.svc.cluster.local:2469/metrics" {
+		m.podTwoCount = m.podTwoCount + 60
+		resp := &http.Response{
+			StatusCode: 200,
+			Body: io.NopCloser(bytes.NewReader([]byte(fmt.Sprintf(`
+# HELP forwarder_read_total Total number of Messages Read
+# TYPE forwarder_read_total counter
+forwarder_read_total{buffer="input",pipeline="simple-pipeline",vertex="input"} %d
+`, m.podTwoCount))))}
+		return resp, nil
+	} else {
+		return nil, nil
+	}
+}
+
+func (m *mockHttpClient) Head(url string) (*http.Response, error) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	if url == "https://p-v-0.p-v-headless.default.svc.cluster.local:2469/metrics" {
+		return &http.Response{
+			StatusCode: 200,
+			Body:       io.NopCloser(bytes.NewReader([]byte(``)))}, nil
+	} else if url == "https://p-v-1.p-v-headless.default.svc.cluster.local:2469/metrics" {
+		return &http.Response{
+			StatusCode: 200,
+			Body:       io.NopCloser(bytes.NewReader([]byte(``)))}, nil
+	} else {
+		return nil, fmt.Errorf("unknown url: %s", url)
+	}
+}
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}
+
+// TestRater_Start tests the rater by mocking the http client
+// we mock the metrics endpoint of the pods and increment the read count by 20 for pod one, and 60 for pod two,
+// then we verify that the rate calculator is able to calculate a positive rate for the vertex
+// note: this test doesn't test the accuracy of the calculated rate, the calculation is tested by helper_test.go
+func TestRater_Start(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*29)
+	defer cancel()
+	pipeline := &v1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "p",
+			Namespace: "default",
+		},
+		Spec: v1alpha1.PipelineSpec{
+			Vertices: []v1alpha1.AbstractVertex{
+				{Name: "v"},
+			},
+		},
+	}
+	r := NewRater(ctx, pipeline)
+	podTracker := NewPodTracker(ctx, pipeline, WithRefreshInterval(time.Second*5))
+	podTracker.httpClient = &mockHttpClient{podOneCount: 0, podTwoCount: 0, lock: &sync.RWMutex{}}
+	r.httpClient = &mockHttpClient{podOneCount: 0, podTwoCount: 0, lock: &sync.RWMutex{}}
+	r.podTracker = podTracker
+
+	timer := time.NewTimer(60 * time.Second)
+	succeedChan := make(chan struct{})
+	go func() {
+		if err := r.Start(ctx); err != nil {
+			log.Fatalf("failed to start rater: %v", err)
+		}
+	}()
+	go func() {
+		for {
+			if r.GetRates("v")["1m"] <= 0 {
+				time.Sleep(time.Second)
+			} else {
+				succeedChan <- struct{}{}
+				break
+			}
+		}
+	}()
+	select {
+	case <-succeedChan:
+		time.Sleep(time.Second)
+		break
+	case <-timer.C:
+		t.Fatalf("timed out waiting for rate to be calculated")
+	}
+	timer.Stop()
+}
diff --git a/pkg/daemon/server/service/rater/timestamped_counts_test.go b/pkg/daemon/server/service/rater/timestamped_counts_test.go
new file mode 100644
index 00000000..2a602665
--- /dev/null
+++ b/pkg/daemon/server/service/rater/timestamped_counts_test.go
@@ -0,0 +1,50 @@
+/*
+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 server
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewTimestampedCounts(t *testing.T) {
+	tc := NewTimestampedCounts(1620000000)
+	assert.Equal(t, int64(1620000000), tc.timestamp)
+	assert.Equal(t, 0, len(tc.podCounts))
+}
+
+func TestTimestampedCounts_Update(t *testing.T) {
+	tc := NewTimestampedCounts(1620000000)
+	tc.Update("pod1", 10.0)
+	assert.Equal(t, 10.0, tc.podCounts["pod1"])
+	tc.Update("pod1", 20.0)
+	assert.Equal(t, 20.0, tc.podCounts["pod1"])
+	tc.Update("pod2", 30.0)
+	assert.Equal(t, 30.0, tc.podCounts["pod2"])
+	assert.Equal(t, 2, len(tc.podCounts))
+	tc.Update("pod1", CountNotAvailable)
+	assert.Equal(t, 1, len(tc.podCounts))
+}
+
+func TestTimestampedCounts_Snapshot(t *testing.T) {
+	tc := NewTimestampedCounts(1620000000)
+	tc.Update("pod1", 10.0)
+	tc.Update("pod2", 20.0)
+	tc.Update("pod3", 30.0)
+	assert.Equal(t, map[string]float64{"pod1": 10.0, "pod2": 20.0, "pod3": 30.0}, tc.Snapshot())
+}
diff --git a/pkg/daemon/server/service/rater/uniq_str_list_test.go b/pkg/daemon/server/service/rater/uniq_str_list_test.go
new file mode 100644
index 00000000..99ccb861
--- /dev/null
+++ b/pkg/daemon/server/service/rater/uniq_str_list_test.go
@@ -0,0 +1,69 @@
+/*
+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 server
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUniqStrList(t *testing.T) {
+	l := NewUniqueStringList()
+	assert.Equal(t, 0, l.Length())
+	assert.Equal(t, "", l.Front())
+
+	// Add a new string
+	l.PushBack("a")
+	assert.Equal(t, 1, l.Length())
+	assert.Equal(t, "a", l.Front())
+	assert.Equal(t, true, l.Contains("a"))
+	assert.Equal(t, false, l.Contains("b"))
+
+	// Add a new string
+	l.PushBack("b")
+	assert.Equal(t, 2, l.Length())
+	assert.Equal(t, "a", l.Front())
+	assert.Equal(t, true, l.Contains("a"))
+	assert.Equal(t, true, l.Contains("b"))
+
+	// Add a new string
+	l.PushBack("c")
+	assert.Equal(t, 3, l.Length())
+	assert.Equal(t, "a", l.Front())
+
+	// Move "a" to back of the list
+	l.MoveToBack("a")
+	assert.Equal(t, 3, l.Length())
+	assert.Equal(t, "b", l.Front())
+
+	// Remove "b"
+	l.Remove("b")
+	assert.Equal(t, 2, l.Length())
+	assert.Equal(t, "c", l.Front())
+	assert.Equal(t, false, l.Contains("b"))
+	assert.Equal(t, true, l.Contains("a"))
+	assert.Equal(t, true, l.Contains("c"))
+
+	// Operate on a non-existing string
+	l.MoveToBack("non-existing")
+	assert.Equal(t, 2, l.Length())
+	assert.Equal(t, "c", l.Front())
+	assert.Equal(t, false, l.Contains("non-existing"))
+	assert.Equal(t, true, l.Contains("a"))
+	assert.Equal(t, true, l.Contains("c"))
+}