Repository: numaproj/numaflow
Pull request #1302 (2023-10-31 15:10:39)

chore: update rater package name
* rename the package from server to rater.
* add doc.go
* let pod tracker own parsing the pod key to pod information, the rater doesn't need to know how to parse a key using separator 

<!--

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/pkg/daemon/server/service/pipeline_metrics_query.go b/pkg/daemon/server/service/pipeline_metrics_query.go
index 22c63d44..5df96c1d 100644
--- a/pkg/daemon/server/service/pipeline_metrics_query.go
+++ b/pkg/daemon/server/service/pipeline_metrics_query.go
@@ -31,7 +31,7 @@ import (
 
 	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/apis/proto/daemon"
-	server "github.com/numaproj/numaflow/pkg/daemon/server/service/rater"
+	rater "github.com/numaproj/numaflow/pkg/daemon/server/service/rater"
 	"github.com/numaproj/numaflow/pkg/isbsvc"
 	"github.com/numaproj/numaflow/pkg/metrics"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
@@ -50,7 +50,7 @@ type pipelineMetadataQuery struct {
 	pipeline          *v1alpha1.Pipeline
 	httpClient        metricsHttpClient
 	watermarkFetchers map[v1alpha1.Edge][]fetch.UXFetcher
-	rater             server.Ratable
+	rater             rater.Ratable
 }
 
 const (
@@ -64,7 +64,7 @@ func NewPipelineMetadataQuery(
 	isbSvcClient isbsvc.ISBService,
 	pipeline *v1alpha1.Pipeline,
 	wmFetchers map[v1alpha1.Edge][]fetch.UXFetcher,
-	rater server.Ratable) (*pipelineMetadataQuery, error) {
+	rater rater.Ratable) (*pipelineMetadataQuery, error) {
 	var err error
 	ps := pipelineMetadataQuery{
 		isbSvcClient: isbSvcClient,
diff --git a/pkg/daemon/server/service/rater/doc.go b/pkg/daemon/server/service/rater/doc.go
new file mode 100644
index 00000000..aa1060e0
--- /dev/null
+++ b/pkg/daemon/server/service/rater/doc.go
@@ -0,0 +1,19 @@
+/*
+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 rater provides the functionality to calculate the processing rate of each vertex partition.
+// The processing rate is calculated based on the metric forwarder_data_read_total.
+package rater
diff --git a/pkg/daemon/server/service/rater/helper.go b/pkg/daemon/server/service/rater/helper.go
index f680eb94..929ac99d 100644
--- a/pkg/daemon/server/service/rater/helper.go
+++ b/pkg/daemon/server/service/rater/helper.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"time"
diff --git a/pkg/daemon/server/service/rater/options.go b/pkg/daemon/server/service/rater/options.go
index 4a798b51..03fdc53d 100644
--- a/pkg/daemon/server/service/rater/options.go
+++ b/pkg/daemon/server/service/rater/options.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 type options struct {
 	// Number of workers working on collecting counts of processed messages.
diff --git a/pkg/daemon/server/service/rater/pod_tracker.go b/pkg/daemon/server/service/rater/pod_tracker.go
index 4fb2fc49..c6936309 100644
--- a/pkg/daemon/server/service/rater/pod_tracker.go
+++ b/pkg/daemon/server/service/rater/pod_tracker.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"context"
@@ -30,10 +30,10 @@ import (
 	"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.
+// podInfoSeparator is used as a separator to split the pod key
+// to get the pipeline name, vertex name, and pod index.
 // "*" is chosen because it is not allowed in all the above fields.
-const PodInfoSeparator = "*"
+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
@@ -131,9 +131,28 @@ func (pt *PodTracker) GetActivePodsCount() int {
 	return pt.activePods.Length()
 }
 
+// podInfo represents the information of a pod that is used for tracking the processing rate
+type podInfo struct {
+	pipelineName string
+	vertexName   string
+	podName      string
+}
+
+func (pt *PodTracker) GetPodInfo(key string) (*podInfo, error) {
+	pi := strings.Split(key, podInfoSeparator)
+	if len(pi) != 3 {
+		return nil, fmt.Errorf("invalid key %q", key)
+	}
+	return &podInfo{
+		pipelineName: pi[0],
+		vertexName:   pi[1],
+		podName:      strings.Join([]string{pi[0], pi[1], pi[2]}, "-"),
+	}, nil
+}
+
 func (pt *PodTracker) getPodKey(index int, vertexName string) string {
 	// 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.
-	return strings.Join([]string{pt.pipeline.Name, vertexName, fmt.Sprintf("%d", index)}, PodInfoSeparator)
+	return strings.Join([]string{pt.pipeline.Name, vertexName, fmt.Sprintf("%d", index)}, podInfoSeparator)
 }
 
 func (pt *PodTracker) isActive(vertexName, podName string) bool {
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index d3060cd1..c7d232ea 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -14,14 +14,13 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"context"
 	"crypto/tls"
 	"fmt"
 	"net/http"
-	"strings"
 	"time"
 
 	"github.com/prometheus/common/expfmt"
@@ -38,6 +37,8 @@ type Ratable interface {
 	GetRates(vertexName, partitionName string) map[string]float64
 }
 
+var _ Ratable = (*Rater)(nil)
+
 // 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
@@ -133,24 +134,22 @@ func (r *Rater) monitor(ctx context.Context, id int, keyCh <-chan string) {
 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) != 3 {
-		return fmt.Errorf("invalid key %q", key)
+	podInfo, err := r.podTracker.GetPodInfo(key)
+	if err != nil {
+		return err
 	}
-	vertexName := podInfo[1]
-	podName := strings.Join([]string{podInfo[0], podInfo[1], podInfo[2]}, "-")
 	var podReadCount *PodReadCount
 	if r.podTracker.IsActive(key) {
-		podReadCount = r.getPodReadCounts(vertexName, podName)
+		podReadCount = r.getPodReadCounts(podInfo.vertexName, podInfo.podName)
 		if podReadCount == nil {
-			log.Debugf("Failed retrieving total podReadCount for pod %s", podName)
+			log.Debugf("Failed retrieving total podReadCount for pod %s", podInfo.podName)
 		}
 	} else {
-		log.Debugf("Pod %s does not exist, updating it with nil...", podName)
+		log.Debugf("Pod %s does not exist, updating it with nil...", podInfo.podName)
 		podReadCount = nil
 	}
 	now := time.Now().Add(CountWindow).Truncate(CountWindow).Unix()
-	UpdateCount(r.timestampedPodCounts[vertexName], now, podReadCount)
+	UpdateCount(r.timestampedPodCounts[podInfo.vertexName], now, podReadCount)
 	return nil
 }
 
diff --git a/pkg/daemon/server/service/rater/timestamped_counts.go b/pkg/daemon/server/service/rater/timestamped_counts.go
index 207771b4..436deeee 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"fmt"
diff --git a/pkg/daemon/server/service/rater/uniq_str_list.go b/pkg/daemon/server/service/rater/uniq_str_list.go
index 2131c510..8bc13f10 100644
--- a/pkg/daemon/server/service/rater/uniq_str_list.go
+++ b/pkg/daemon/server/service/rater/uniq_str_list.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"container/list"
diff --git a/pkg/forward/doc.go b/pkg/forward/doc.go
index 8e8c01c3..a9141c6d 100644
--- a/pkg/forward/doc.go
+++ b/pkg/forward/doc.go
@@ -11,5 +11,5 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package forward is used for creating a data forwarder for the map UDF and sink vertex.
+// Package forward is used for creating a data forwarder for the map UDF vertex.
 package forward
diff --git a/ui/src/components/pages/Pipeline/partials/Graph/index.tsx b/ui/src/components/pages/Pipeline/partials/Graph/index.tsx
index c372a223..d1461642 100644
--- a/ui/src/components/pages/Pipeline/partials/Graph/index.tsx
+++ b/ui/src/components/pages/Pipeline/partials/Graph/index.tsx
@@ -174,10 +174,22 @@ const Flow = (props: FlowProps) => {
     undefined
   );
   const [statusPayload, setStatusPayload] = useState<any>(undefined);
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
   const [timerDateStamp, setTimerDateStamp] = useState<any>(undefined);
-  const [timer, setTimer] = useState<any>(undefined);
+  const [timer, setTimer] = useState<number | undefined>(undefined);
 
+  const handleTimer = useCallback(() => {
+    if (timer) {
+      clearInterval(timer);
+    }
+    const dateString = new Date().toISOString();
+    const time = timeAgo(dateString);
+    setTimerDateStamp(time);
+    const pauseTimer = setInterval(() => {
+      const time = timeAgo(dateString);
+      setTimerDateStamp(time);
+    }, 1000);
+    setTimer(pauseTimer);
+  }, [timer]);
   const handlePlayClick = useCallback(() => {
     handleTimer();
     setStatusPayload({
@@ -187,7 +199,7 @@ const Flow = (props: FlowProps) => {
         },
       },
     });
-  }, []);
+  }, [handleTimer]);
 
   const handlePauseClick = useCallback(() => {
     handleTimer();
@@ -198,18 +210,7 @@ const Flow = (props: FlowProps) => {
         },
       },
     });
-  }, []);
-
-  const handleTimer = useCallback(() => {
-    const dateString = new Date().toISOString();
-    const time = timeAgo(dateString);
-    setTimerDateStamp(time);
-    const pauseTimer = setInterval(() => {
-      const time = timeAgo(dateString);
-      setTimerDateStamp(time);
-    }, 1000);
-    setTimer(pauseTimer);
-  }, []);
+  }, [handleTimer]);
 
   useEffect(() => {
     const patchStatus = async () => {
@@ -245,14 +246,12 @@ const Flow = (props: FlowProps) => {
       statusPayload?.spec?.lifecycle?.desiredPhase === PAUSED &&
       data?.pipeline?.status?.phase === PAUSED
     ) {
-      clearInterval(timer);
       setStatusPayload(undefined);
     }
     if (
       statusPayload?.spec?.lifecycle?.desiredPhase === RUNNING &&
       data?.pipeline?.status?.phase === RUNNING
     ) {
-      clearInterval(timer);
       setStatusPayload(undefined);
     }
   }, [data]);


Test patch:
diff --git a/pkg/daemon/server/service/rater/goleak_test.go b/pkg/daemon/server/service/rater/goleak_test.go
index fb234536..58abf6f6 100644
--- a/pkg/daemon/server/service/rater/goleak_test.go
+++ b/pkg/daemon/server/service/rater/goleak_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"testing"
diff --git a/pkg/daemon/server/service/rater/helper_test.go b/pkg/daemon/server/service/rater/helper_test.go
index e2f00388..e420bcc0 100644
--- a/pkg/daemon/server/service/rater/helper_test.go
+++ b/pkg/daemon/server/service/rater/helper_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"testing"
diff --git a/pkg/daemon/server/service/rater/pod_tracker_test.go b/pkg/daemon/server/service/rater/pod_tracker_test.go
index 203645e9..99974345 100644
--- a/pkg/daemon/server/service/rater/pod_tracker_test.go
+++ b/pkg/daemon/server/service/rater/pod_tracker_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"bytes"
@@ -123,3 +123,20 @@ func TestPodTracker_Start(t *testing.T) {
 	assert.Equal(t, true, tracker.IsActive("p*v*4"))
 	assert.Equal(t, false, tracker.IsActive("p*v*5"))
 }
+
+func TestPodTracker_GetPodInfo(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*30)
+	defer cancel()
+	tracker := NewPodTracker(ctx, nil, WithRefreshInterval(time.Second))
+	// error scenario - more than 3 fields
+	podInfo, err := tracker.GetPodInfo("p*v*0*1")
+	assert.Nilf(t, podInfo, "podInfo should be nil")
+	assert.ErrorContains(t, err, "invalid key")
+
+	// common scenario - get the pod info
+	podInfo, err = tracker.GetPodInfo("p*v*0")
+	assert.Nilf(t, err, "error should be nil")
+	assert.Equal(t, "p", podInfo.pipelineName)
+	assert.Equal(t, "v", podInfo.vertexName)
+	assert.Equal(t, "p-v-0", podInfo.podName)
+}
diff --git a/pkg/daemon/server/service/rater/rater_test.go b/pkg/daemon/server/service/rater/rater_test.go
index b33dbf8e..e6f6fae0 100644
--- a/pkg/daemon/server/service/rater/rater_test.go
+++ b/pkg/daemon/server/service/rater/rater_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"bytes"
diff --git a/pkg/daemon/server/service/rater/timestamped_counts_test.go b/pkg/daemon/server/service/rater/timestamped_counts_test.go
index 048874d8..4220ab98 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts_test.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"testing"
diff --git a/pkg/daemon/server/service/rater/uniq_str_list_test.go b/pkg/daemon/server/service/rater/uniq_str_list_test.go
index 99ccb861..94f958e0 100644
--- a/pkg/daemon/server/service/rater/uniq_str_list_test.go
+++ b/pkg/daemon/server/service/rater/uniq_str_list_test.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package server
+package rater
 
 import (
 	"testing"