fix: calculate processing rate for sink vertices (#1025)

Signed-off-by: Derek Wang <[email protected]>

pkg/daemon/server/service/rater/pod_tracker.go

@@ -93,6 +93,8 @@ func (pt *PodTracker) Start(ctx context.Context) error {
 						vType = "reduce"
 					} else if v.IsASource() {
 						vType = "source"
+					} else if v.IsASink() {
+						vType = "sink"
 					} else {
 						vType = "other"
 					}

test/e2e/functional_test.go

@@ -117,13 +117,20 @@ func (s *FunctionalSuite) TestCreateSimplePipeline() {
 	waitInterval := 10 * time.Second
 	succeedChan := make(chan struct{})
 	go func() {
+		vertexNames := []string{"input", "p1", "output"}
 		for {
-			m, err := client.GetVertexMetrics(context.Background(), pipelineName, "p1")
-			assert.NoError(s.T(), err)
-			assert.Equal(s.T(), pipelineName, *m[0].Pipeline)
-			oneMinRate := m[0].ProcessingRates["1m"]
-			// the rate should be around 5
-			if oneMinRate < 4 || oneMinRate > 6 {
+			accurateCount := 0
+			for _, vertexName := range vertexNames {
+				m, err := client.GetVertexMetrics(context.Background(), pipelineName, vertexName)
+				assert.NoError(s.T(), err)
+				assert.Equal(s.T(), pipelineName, *m[0].Pipeline)
+				oneMinRate := m[0].ProcessingRates["1m"]
+				// the rate should be around 5
+				if oneMinRate > 4 || oneMinRate < 6 {
+					accurateCount++
+				}
+			}
+			if accurateCount != len(vertexNames) {
 				time.Sleep(waitInterval)
 			} else {
 				succeedChan <- struct{}{}