Repository: numaproj/numaflow (go)

Issue:
Implement per-partition rate calculation and display it on UI
As we migrate to metric-based rate calculation, we no longer have partition level processing rates for reduce vertex. The processing rates we delivered through daemon are ALL per vertex. Therefore we no longer support the per-partition rate feature:

![Screenshot 2023-05-24 at 5 59 23 PM](https://github.com/numaproj/numaflow/assets/13165977/e66840e8-004a-4514-9606-c71558fed3a9)

In this task, we implement per-partition rate calculation and display it on UI.

---

<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.


Pull request description:
chore: use metrics to calculate rate for reducer partitions
Fixes #746 

Showing the processing rate for each of the partitions is important as it helps us identify hot key issue. This change enables the rater to calculate rate for a single pod, such that we can retrieve per-partition rates and display them on UI.

### How was this change tested
**Before this change**

![before](https://github.com/numaproj/numaflow/assets/13165977/e421d819-6e53-440b-beab-c9996f1bae06)

**After**

![after](https://github.com/numaproj/numaflow/assets/13165977/436825c3-42b3-4c01-ab6c-da3a5047f667)



Patch:
diff --git a/pkg/daemon/server/daemon_server.go b/pkg/daemon/server/daemon_server.go
index af28f0e6..194f08a9 100644
--- a/pkg/daemon/server/daemon_server.go
+++ b/pkg/daemon/server/daemon_server.go
@@ -135,7 +135,7 @@ func (ds *daemonServer) Run(ctx context.Context) error {
 func (ds *daemonServer) newGRPCServer(
 	isbSvcClient isbsvc.ISBService,
 	wmFetchers map[string][]fetch.Fetcher,
-	rater *server.Rater) (*grpc.Server, error) {
+	rater server.Ratable) (*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:"
@@ -149,7 +149,7 @@ func (ds *daemonServer) newGRPCServer(
 	}
 	grpcServer := grpc.NewServer(sOpts...)
 	grpc_prometheus.Register(grpcServer)
-	pipelineMetadataQuery, err := service.NewPipelineMetadataQuery(isbSvcClient, ds.pipeline, wmFetchers, rater, true)
+	pipelineMetadataQuery, err := service.NewPipelineMetadataQuery(isbSvcClient, ds.pipeline, wmFetchers, rater)
 	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 c05a9f54..0d605720 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query.go
@@ -51,11 +51,7 @@ 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
+	rater             server.Ratable
 }
 
 const (
@@ -69,8 +65,7 @@ func NewPipelineMetadataQuery(
 	isbSvcClient isbsvc.ISBService,
 	pipeline *v1alpha1.Pipeline,
 	wmFetchers map[string][]fetch.Fetcher,
-	rater *server.Rater,
-	useNewRateCalculation bool) (*pipelineMetadataQuery, error) {
+	rater server.Ratable) (*pipelineMetadataQuery, error) {
 	var err error
 	ps := pipelineMetadataQuery{
 		isbSvcClient: isbSvcClient,
@@ -81,9 +76,8 @@ func NewPipelineMetadataQuery(
 			},
 			Timeout: time.Second * 3,
 		},
-		watermarkFetchers:     wmFetchers,
-		rater:                 rater,
-		useNewRateCalculation: useNewRateCalculation,
+		watermarkFetchers: wmFetchers,
+		rater:             rater,
 	}
 	if err != nil {
 		return nil, err
@@ -166,32 +160,21 @@ func (ps *pipelineMetadataQuery) GetBuffer(ctx context.Context, req *daemon.GetB
 func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daemon.GetVertexMetricsRequest) (*daemon.GetVertexMetricsResponse, error) {
 	log := logging.FromContext(ctx)
 	resp := new(daemon.GetVertexMetricsResponse)
-
 	vertexName := fmt.Sprintf("%s-%s", ps.pipeline.Name, req.GetVertex())
+
+	// Get the headless service name
 	vertex := &v1alpha1.Vertex{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: vertexName,
 		},
 	}
-	// partitions indicates replica count ~ multiple pods for a vertex here
-	obj := ps.pipeline.NumOfPartitions(req.GetVertex())
-	podNum := int64(obj)
-
-	var vertexLevelRates map[string]float64
-
-	if ps.useNewRateCalculation {
-		vertexLevelRates = ps.rater.GetRates(req.GetVertex())
-	}
-
-	// Get the headless service name
 	headlessServiceName := vertex.GetHeadlessServiceName()
 
-	metricsArr := make([]*daemon.VertexMetrics, podNum)
-	for i := int64(0); i < podNum; i++ {
-
-		processingRates := make(map[string]float64, 0)
-		pendings := make(map[string]int64, 0)
+	abstractVertex := ps.pipeline.GetVertex(req.GetVertex())
+	vertexLevelRates := ps.rater.GetRates(req.GetVertex())
 
+	metricsArr := make([]*daemon.VertexMetrics, abstractVertex.GetPartitions())
+	for i := int64(0); i < int64(abstractVertex.GetPartitions()); i++ {
 		// 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.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)
@@ -210,22 +193,8 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 				return nil, err
 			}
 
-			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()
-							}
-						}
-					}
-				}
-			}
-
+			// Get the pending messages for this partition
+			pendings := make(map[string]int64, 0)
 			if value, ok := result[metrics.VertexPendingMessages]; ok {
 				metricsList := value.GetMetric()
 				for _, metric := range metricsList {
@@ -238,22 +207,23 @@ func (ps *pipelineMetadataQuery) GetVertexMetrics(ctx context.Context, req *daem
 					}
 				}
 			}
+			vm := &daemon.VertexMetrics{
+				Pipeline: &ps.pipeline.Name,
+				Vertex:   req.Vertex,
+				Pendings: pendings,
+			}
 
-			if !ps.useNewRateCalculation {
-				metricsArr[i] = &daemon.VertexMetrics{
-					Pipeline:        &ps.pipeline.Name,
-					Vertex:          req.Vertex,
-					ProcessingRates: processingRates,
-					Pendings:        pendings,
-				}
+			// Get the processing rate for this partition
+			if abstractVertex.IsReduceUDF() {
+				// the processing rate of this ith partition is the rate of the corresponding ith pod.
+				vm.ProcessingRates = ps.rater.GetPodRates(req.GetVertex(), int(i))
 			} else {
-				metricsArr[i] = &daemon.VertexMetrics{
-					Pipeline:        &ps.pipeline.Name,
-					Vertex:          req.Vertex,
-					ProcessingRates: vertexLevelRates,
-					Pendings:        pendings,
-				}
+				// if the vertex is not a reduce udf, then the processing rate is the sum of all pods in this vertex.
+				// TODO (multi-partition) - change this to display the processing rate of each partition when we finish multi-partition support for non-reduce vertices.
+				vm.ProcessingRates = vertexLevelRates
 			}
+
+			metricsArr[i] = vm
 		}
 	}
 
diff --git a/pkg/daemon/server/service/rater/helper.go b/pkg/daemon/server/service/rater/helper.go
index 1cdb7a3c..f0055332 100644
--- a/pkg/daemon/server/service/rater/helper.go
+++ b/pkg/daemon/server/service/rater/helper.go
@@ -22,6 +22,8 @@ import (
 	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
 )
 
+const IndexNotFound = -1
+
 // 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
@@ -43,23 +45,20 @@ func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSec
 	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 {
+	startIndex := findStartIndex(lookbackSeconds, counts)
+	if startIndex == IndexNotFound {
 		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
+	if timeDiff == 0 {
+		// if the time difference is 0, we return 0 to avoid division by 0
+		// this should not happen in practice because we are using a 10s interval
+		return 0
+	}
 	for i := startIndex; i < n-1; i++ {
 		delta = delta + calculateDelta(counts[i], counts[i+1])
 	}
@@ -86,3 +85,66 @@ func calculateDelta(c1, c2 *TimestampedCounts) float64 {
 	}
 	return delta
 }
+
+// CalculatePodRate calculates the rate of a pod in the last lookback seconds
+func CalculatePodRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64, podName string) float64 {
+	n := q.Length()
+	if n <= 1 {
+		return 0
+	}
+	counts := q.Items()
+	startIndex := findStartIndex(lookbackSeconds, counts)
+	if startIndex == IndexNotFound {
+		return 0
+	}
+
+	delta := float64(0)
+	// time diff in seconds.
+	timeDiff := counts[n-1].timestamp - counts[startIndex].timestamp
+	if timeDiff == 0 {
+		// if the time difference is 0, we return 0 to avoid division by 0
+		// this should not happen in practice because we are using a 10s interval
+		return 0
+	}
+	for i := startIndex; i < n-1; i++ {
+		delta = delta + calculatePodDelta(counts[i], counts[i+1], podName)
+	}
+	return delta / float64(timeDiff)
+}
+
+func calculatePodDelta(c1, c2 *TimestampedCounts, podName string) float64 {
+	tc1 := c1.Snapshot()
+	tc2 := c2.Snapshot()
+	count1, exist1 := tc1[podName]
+	count2, exist2 := tc2[podName]
+	if !exist2 {
+		return 0
+	} else if !exist1 {
+		return count2
+	} else if count2 < count1 {
+		return count2
+	} else {
+		return count2 - count1
+	}
+}
+
+// findStartIndex finds the index of the first element in the queue that is within the lookback seconds
+// size of counts is at least 2
+func findStartIndex(lookbackSeconds int64, counts []*TimestampedCounts) int {
+	n := len(counts)
+	now := time.Now().Truncate(time.Second * 10).Unix()
+	if now-counts[n-2].timestamp > lookbackSeconds {
+		// if the second last element is already outside the lookback window, we return IndexNotFound
+		return IndexNotFound
+	}
+
+	startIndex := n - 2
+	for i := n - 2; i >= 0; i-- {
+		if now-counts[i].timestamp <= lookbackSeconds {
+			startIndex = i
+		} else {
+			break
+		}
+	}
+	return startIndex
+}
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index 0cb66eaf..98113456 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -32,6 +33,12 @@ import (
 	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
 )
 
+type Ratable interface {
+	Start(ctx context.Context) error
+	GetRates(vertexName string) map[string]float64
+	GetPodRates(vertexName string, podIndex int) map[string]float64
+}
+
 // 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
@@ -225,6 +232,27 @@ func (r *Rater) getTotalCount(vertexName, vertexType, podName string) float64 {
 // 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)
+	// calculate rates for each lookback seconds
+	for n, i := range r.buildLookbackSecondsMap(vertexName) {
+		r := CalculateRate(r.timestampedPodCounts[vertexName], i)
+		result[n] = r
+	}
+	return result
+}
+
+// GetPodRates returns the processing rates of the pod in the format of lookback second to rate mappings
+func (r *Rater) GetPodRates(vertexName string, podIndex int) map[string]float64 {
+	podName := r.pipeline.Name + "-" + vertexName + "-" + strconv.Itoa(podIndex)
+	var result = make(map[string]float64)
+	// calculate rates for each lookback seconds
+	for n, i := range r.buildLookbackSecondsMap(vertexName) {
+		r := CalculatePodRate(r.timestampedPodCounts[vertexName], i, podName)
+		result[n] = r
+	}
+	return result
+}
+
+func (r *Rater) buildLookbackSecondsMap(vertexName string) map[string]int64 {
 	// 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.
@@ -237,11 +265,5 @@ func (r *Rater) GetRates(vertexName string) map[string]float64 {
 	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
+	return lookbackSecondsMap
 }


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 fcfaca39..4e419827 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query_test.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query_test.go
@@ -41,7 +41,6 @@ type mockHttpClient struct {
 
 func (m *mockHttpClient) Get(url string) (*http.Response, error) {
 	return m.MockGet(url)
-
 }
 
 type mockIsbSvcClient struct {
@@ -72,6 +71,27 @@ func (ms *mockIsbSvcClient) CreateWatermarkFetcher(ctx context.Context, bucketNa
 	return nil, nil
 }
 
+// mock rater
+type mockRater_TestGetVertexMetrics struct {
+}
+
+func (mr *mockRater_TestGetVertexMetrics) Start(ctx context.Context) error {
+	return nil
+}
+
+func (mr *mockRater_TestGetVertexMetrics) GetRates(vertexName string) map[string]float64 {
+	res := make(map[string]float64)
+	res["default"] = 4.894736842105263
+	res["1m"] = 5.084745762711864
+	res["5m"] = 4.894736842105263
+	res["15m"] = 4.894736842105263
+	return res
+}
+
+func (mr *mockRater_TestGetVertexMetrics) GetPodRates(vertexName string, podIndex int) map[string]float64 {
+	return nil
+}
+
 func TestGetVertexMetrics(t *testing.T) {
 	pipelineName := "simple-pipeline"
 	vertexName := "cat"
@@ -81,17 +101,10 @@ func TestGetVertexMetrics(t *testing.T) {
 		Spec:       v1alpha1.PipelineSpec{Vertices: []v1alpha1.AbstractVertex{{Name: vertexName, Partitions: &vertexPartition}}},
 	}
 	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, nil, false)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, &mockRater_TestGetVertexMetrics{})
 	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.
-# TYPE vertex_processing_rate gauge
-vertex_processing_rate{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-vertex_processing_rate{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.084745762711864
-vertex_processing_rate{period="5m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-vertex_processing_rate{period="default",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-
-# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
+	metricsResponse := `# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
 # TYPE vertex_pending_messages gauge
 vertex_pending_messages{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.011
 vertex_pending_messages{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.333
@@ -156,7 +169,7 @@ func TestGetBuffer(t *testing.T) {
 	}
 
 	ms := &mockIsbSvcClient{}
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil, false)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil)
 	assert.NoError(t, err)
 
 	bufferName := "numaflow-system-simple-pipeline-cat-0"
@@ -197,7 +210,7 @@ func TestListBuffers(t *testing.T) {
 	}
 
 	ms := &mockIsbSvcClient{}
-	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil, false)
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(ms, pipeline, nil, nil)
 	assert.NoError(t, err)
 
 	req := &daemon.ListBuffersRequest{Pipeline: &pipelineName}
@@ -207,6 +220,35 @@ func TestListBuffers(t *testing.T) {
 	assert.Equal(t, len(resp.Buffers), 2)
 }
 
+// mock rater
+type mockRater_TestGetPipelineStatus struct {
+	isActivelyProcessing bool
+}
+
+func (mr *mockRater_TestGetPipelineStatus) Start(ctx context.Context) error {
+	return nil
+}
+
+func (mr *mockRater_TestGetPipelineStatus) GetRates(vertexName string) map[string]float64 {
+	res := make(map[string]float64)
+	if mr.isActivelyProcessing {
+		res["default"] = 4.894736842105263
+		res["1m"] = 5.084745762711864
+		res["5m"] = 4.894736842105263
+		res["15m"] = 4.894736842105263
+	} else {
+		res["default"] = 0
+		res["1m"] = 0
+		res["5m"] = 0
+		res["15m"] = 0
+	}
+	return res
+}
+
+func (mr *mockRater_TestGetPipelineStatus) GetPodRates(vertexName string, podIndex int) map[string]float64 {
+	return nil
+}
+
 func TestGetPipelineStatus(t *testing.T) {
 	pipelineName := "simple-pipeline"
 	pipeline := &v1alpha1.Pipeline{
@@ -220,28 +262,20 @@ func TestGetPipelineStatus(t *testing.T) {
 		},
 	}
 	client, _ := isbsvc.NewISBJetStreamSvc(pipelineName)
-	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.")}
-	ErrorPipelineResponse := daemon.PipelineStatus{Status: pointer.String("Error"), Message: pointer.String("Pipeline has an error. Vertex cat is not processing pending messages.")}
-
-	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.
-# TYPE vertex_processing_rate gauge
-vertex_processing_rate{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-vertex_processing_rate{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.084745762711864
-vertex_processing_rate{period="5m",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-vertex_processing_rate{period="default",pipeline="simple-pipeline",vertex="cat"} 4.894736842105263
-
-# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
+	metricsResponse := `# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
 # TYPE vertex_pending_messages gauge
 vertex_pending_messages{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.011
 vertex_pending_messages{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.333
 vertex_pending_messages{period="5m",pipeline="simple-pipeline",vertex="cat"} 6.002
 vertex_pending_messages{period="default",pipeline="simple-pipeline",vertex="cat"} 7.00002
 `
-	ioReader := io.NopCloser(bytes.NewReader([]byte(metricsResponse)))
+	req := &daemon.GetPipelineStatusRequest{Pipeline: &pipelineName}
 
+	// test when rater is actively processing
+	activeRater := &mockRater_TestGetPipelineStatus{isActivelyProcessing: true}
+	pipelineMetricsQueryService, err := NewPipelineMetadataQuery(client, pipeline, nil, activeRater)
+	assert.NoError(t, err)
+	ioReader := io.NopCloser(bytes.NewReader([]byte(metricsResponse)))
 	pipelineMetricsQueryService.httpClient = &mockHttpClient{
 		MockGet: func(url string) (*http.Response, error) {
 			return &http.Response{
@@ -250,29 +284,16 @@ vertex_pending_messages{period="default",pipeline="simple-pipeline",vertex="cat"
 			}, nil
 		},
 	}
-
-	req := &daemon.GetPipelineStatusRequest{Pipeline: &pipelineName}
-
 	resp, err := pipelineMetricsQueryService.GetPipelineStatus(context.Background(), req)
 	assert.NoError(t, err)
+	OKPipelineResponse := daemon.PipelineStatus{Status: pointer.String("OK"), Message: pointer.String("Pipeline has no issue.")}
 	assert.Equal(t, &OKPipelineResponse, resp.Status)
 
-	errorMetricsResponse := `# 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.
-# TYPE vertex_processing_rate gauge
-vertex_processing_rate{period="15m",pipeline="simple-pipeline",vertex="cat"} 0
-vertex_processing_rate{period="1m",pipeline="simple-pipeline",vertex="cat"} 0
-vertex_processing_rate{period="5m",pipeline="simple-pipeline",vertex="cat"} 0
-vertex_processing_rate{period="default",pipeline="simple-pipeline",vertex="cat"} 0
-
-# HELP vertex_pending_messages Average pending messages in the last period of seconds. It is the pending messages of a vertex, not a pod.
-# TYPE vertex_pending_messages gauge
-vertex_pending_messages{period="15m",pipeline="simple-pipeline",vertex="cat"} 4.011
-vertex_pending_messages{period="1m",pipeline="simple-pipeline",vertex="cat"} 5.333
-vertex_pending_messages{period="5m",pipeline="simple-pipeline",vertex="cat"} 6.002
-vertex_pending_messages{period="default",pipeline="simple-pipeline",vertex="cat"} 7.00002
-`
-	ioReader = io.NopCloser(bytes.NewReader([]byte(errorMetricsResponse)))
-
+	// test when rater is not actively processing
+	idleRater := &mockRater_TestGetPipelineStatus{isActivelyProcessing: false}
+	pipelineMetricsQueryService, err = NewPipelineMetadataQuery(client, pipeline, nil, idleRater)
+	assert.NoError(t, err)
+	ioReader = io.NopCloser(bytes.NewReader([]byte(metricsResponse)))
 	pipelineMetricsQueryService.httpClient = &mockHttpClient{
 		MockGet: func(url string) (*http.Response, error) {
 			return &http.Response{
@@ -281,8 +302,8 @@ vertex_pending_messages{period="default",pipeline="simple-pipeline",vertex="cat"
 			}, nil
 		},
 	}
-
 	resp, err = pipelineMetricsQueryService.GetPipelineStatus(context.Background(), req)
 	assert.NoError(t, err)
+	ErrorPipelineResponse := daemon.PipelineStatus{Status: pointer.String("Error"), Message: pointer.String("Pipeline has an error. Vertex cat is not processing pending messages.")}
 	assert.Equal(t, &ErrorPipelineResponse, resp.Status)
 }
diff --git a/pkg/daemon/server/service/rater/helper_test.go b/pkg/daemon/server/service/rater/helper_test.go
index 9cbb5e6d..314f1f7c 100644
--- a/pkg/daemon/server/service/rater/helper_test.go
+++ b/pkg/daemon/server/service/rater/helper_test.go
@@ -107,8 +107,17 @@ func TestUpdateCount(t *testing.T) {
 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)
+		// no data
+		assert.Equal(t, 0.0, CalculateRate(q, 10))
+		assert.Equal(t, 0.0, CalculatePodRate(q, 10, "pod1"))
+
+		// only one data
+		now := time.Now()
+		tc1 := NewTimestampedCounts(now.Truncate(time.Second*10).Unix() - 20)
+		tc1.Update("pod1", 5.0)
+		q.Append(tc1)
+		assert.Equal(t, 0.0, CalculateRate(q, 10))
+		assert.Equal(t, 0.0, CalculatePodRate(q, 10, "pod1"))
 	})
 
 	t.Run("singlePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -129,6 +138,11 @@ func TestCalculateRate(t *testing.T) {
 		assert.Equal(t, 1.0, CalculateRate(q, 15))
 		assert.Equal(t, 0.75, CalculateRate(q, 25))
 		assert.Equal(t, 0.75, CalculateRate(q, 100))
+
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod1"))
+		assert.Equal(t, 1.0, CalculatePodRate(q, 15, "pod1"))
+		assert.Equal(t, 0.75, CalculatePodRate(q, 25, "pod1"))
+		assert.Equal(t, 0.75, CalculatePodRate(q, 100, "pod1"))
 	})
 
 	t.Run("singlePod_givenCountDecreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -153,6 +167,12 @@ func TestCalculateRate(t *testing.T) {
 		assert.Equal(t, 4.0, CalculateRate(q, 25))
 		assert.Equal(t, 6.0, CalculateRate(q, 35))
 		assert.Equal(t, 6.0, CalculateRate(q, 100))
+
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod1"))
+		assert.Equal(t, 3.0, CalculatePodRate(q, 15, "pod1"))
+		assert.Equal(t, 4.0, CalculatePodRate(q, 25, "pod1"))
+		assert.Equal(t, 6.0, CalculatePodRate(q, 35, "pod1"))
+		assert.Equal(t, 6.0, CalculatePodRate(q, 100, "pod1"))
 	})
 
 	t.Run("multiplePods_givenCountIncreasesAndDecreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -176,11 +196,26 @@ func TestCalculateRate(t *testing.T) {
 		tc4.Update("pod2", 400.0)
 		q.Append(tc4)
 
+		// vertex rate
 		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))
+
+		// pod1 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod1"))
+		assert.Equal(t, 3.0, CalculatePodRate(q, 15, "pod1"))
+		assert.Equal(t, 4.0, CalculatePodRate(q, 25, "pod1"))
+		assert.Equal(t, 6.0, CalculatePodRate(q, 35, "pod1"))
+		assert.Equal(t, 6.0, CalculatePodRate(q, 100, "pod1"))
+
+		// pod2 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 15, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 25, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 35, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 100, "pod2"))
 	})
 
 	t.Run("multiplePods_givenPodsComeAndGo_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -207,14 +242,50 @@ func TestCalculateRate(t *testing.T) {
 		tc4.Update("pod100", 200.0)
 		q.Append(tc4)
 
+		// vertex rate
 		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))
+
+		// pod1 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod1"))
+		assert.Equal(t, 0.0, CalculatePodRate(q, 15, "pod1"))
+		assert.Equal(t, 2.5, CalculatePodRate(q, 25, "pod1"))
+		assert.Equal(t, 5.0, CalculatePodRate(q, 35, "pod1"))
+		assert.Equal(t, 5.0, CalculatePodRate(q, 100, "pod1"))
+
+		// pod2 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 15, "pod2"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 25, "pod2"))
+		assert.InDelta(t, 10.333, CalculatePodRate(q, 35, "pod2"), 0.001)
+		assert.InDelta(t, 10.333, CalculatePodRate(q, 100, "pod2"), 0.001)
+
+		// pod3 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod3"))
+		assert.Equal(t, 20.0, CalculatePodRate(q, 15, "pod3"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 25, "pod3"))
+		assert.InDelta(t, 6.666, CalculatePodRate(q, 100, "pod3"), 0.001)
+		assert.InDelta(t, 6.666, CalculatePodRate(q, 100, "pod3"), 0.001)
+
+		// pod4 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod4"))
+		assert.Equal(t, 0.0, CalculatePodRate(q, 15, "pod4"))
+		assert.Equal(t, 5.0, CalculatePodRate(q, 25, "pod4"))
+		assert.InDelta(t, 3.333, CalculatePodRate(q, 35, "pod4"), 0.001)
+		assert.InDelta(t, 3.333, CalculatePodRate(q, 100, "pod4"), 0.001)
+
+		// pod100 rate
+		assert.Equal(t, 0.0, CalculatePodRate(q, 5, "pod100"))
+		assert.Equal(t, 20.0, CalculatePodRate(q, 15, "pod100"))
+		assert.Equal(t, 10.0, CalculatePodRate(q, 25, "pod100"))
+		assert.InDelta(t, 6.666, CalculatePodRate(q, 35, "pod100"), 0.001)
+		assert.InDelta(t, 6.666, CalculatePodRate(q, 100, "pod100"), 0.001)
 	})
 
-	t.Run("queueOverflowed_SinglePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
+	t.Run("queueOverflowed_givenPodsComeAndGo_whenCalculateRate_thenReturnRate", func(t *testing.T) {
 		q := sharedqueue.New[*TimestampedCounts](3)
 		now := time.Now()
 
diff --git a/pkg/daemon/server/service/rater/rater_test.go b/pkg/daemon/server/service/rater/rater_test.go
index 74e720d0..4d0f71ee 100644
--- a/pkg/daemon/server/service/rater/rater_test.go
+++ b/pkg/daemon/server/service/rater/rater_test.go
@@ -120,7 +120,7 @@ func TestRater_Start(t *testing.T) {
 	}()
 	go func() {
 		for {
-			if r.GetRates("v")["1m"] <= 0 {
+			if r.GetRates("v")["1m"] <= 0 || r.GetPodRates("v", 0)["1m"] <= 0 || r.GetPodRates("v", 1)["1m"] <= 0 {
 				time.Sleep(time.Second)
 			} else {
 				succeedChan <- struct{}{}