Repository: numaproj/numaflow
Pull request #781 (2023-06-08 10:12:23)

fix: duplicate ui served from gin Router
Currently, on refresh we return index.html (default HTML file in your browser when visiting any directory on a web server) content twice when we don't set the baseHref like in this response. This caused all the calls in the UI to be run twice (meaning duplicate logs and API calls). This PR fixes the routing.
![image](https://github.com/numaproj/numaflow/assets/49195734/c4cd597c-1fb8-4ce3-bb21-6ca3b39ce2c0)


Also added a safety check to the Breadcrumbs component, to provide correct navigation for trailing slashes.
(`Unknown -> Namespaces > default (simple-pipeline)`)
<img width="1403" alt="image" src="https://github.com/numaproj/numaflow/assets/49195734/89c8fc4e-5975-4b77-90cb-95f487a2c230">


Patch:
diff --git a/pkg/daemon/server/service/rater/options.go b/pkg/daemon/server/service/rater/options.go
index faa8bf35..4a798b51 100644
--- a/pkg/daemon/server/service/rater/options.go
+++ b/pkg/daemon/server/service/rater/options.go
@@ -28,14 +28,15 @@ 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
+	// The task interval is 5 seconds, which means each task need to be picked up by a worker every 5 seconds.
+	// Hence, a worker needs to finish processing 1 task in 0.5 second.
+	// Translating to numaflow language, for a 200-pod pipeline, a worker needs to finish scraping 1 pod in 0.5 second, which is a reasonable number.
 	return &options{
 		workers: 20,
-		// 10 seconds
-		taskInterval: 10000,
+		// ensure that each task is picked up at least once within a CountWindow by defining taskInterval as half of CountWindow.
+		// if a CountWindow misses one pod, when calculating the delta with the next window, for that specific pod,
+		// we will count the total processed count as delta, which is wrong and eventually leads to incorrect high processing rate.
+		taskInterval: int(CountWindow.Milliseconds() / 2),
 	}
 }
 
diff --git a/pkg/daemon/server/service/rater/pod_tracker.go b/pkg/daemon/server/service/rater/pod_tracker.go
index 0684f740..949759bd 100644
--- a/pkg/daemon/server/service/rater/pod_tracker.go
+++ b/pkg/daemon/server/service/rater/pod_tracker.go
@@ -125,6 +125,10 @@ func (pt *PodTracker) isActive(vertexName, podName string) bool {
 	// 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 {
+		// during performance test (100 pods per vertex), we never saw a false negative, meaning every time isActive returns false,
+		// it truly means the pod doesn't exist.
+		// in reality, we can imagine that a pod can be active but the Head request times out for some reason and returns an incorrect false,
+		// if we ever observe such case, we can think about adding retry here.
 		return false
 	}
 	return true
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index 98113456..151d7796 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -81,8 +81,8 @@ func NewRater(ctx context.Context, p *v1alpha1.Pipeline, opts ...Option) *Rater
 
 	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)
+		// maintain the total counts of the last 30 minutes(1800 seconds) since we support 1m, 5m, 15m lookback seconds.
+		rater.timestampedPodCounts[v.Name] = sharedqueue.New[*TimestampedCounts](int(1800 / CountWindow.Seconds()))
 	}
 
 	for _, opt := range opts {
@@ -125,8 +125,7 @@ func (r *Rater) monitorOnePod(ctx context.Context, key string, worker int) error
 	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)
+			log.Debugf("Failed retrieving total count for pod %s", podName)
 		}
 	} else {
 		log.Debugf("Pod %s does not exist, updating it with CountNotAvailable...", podName)
@@ -224,6 +223,7 @@ func (r *Rater) getTotalCount(vertexName, vertexType, podName string) float64 {
 			metricsList := value.GetMetric()
 			return metricsList[0].Counter.GetValue()
 		} else {
+			r.log.Errorf("failed getting the read total metric, the metric is not available.")
 			return CountNotAvailable
 		}
 	}
diff --git a/pkg/daemon/server/service/rater/timestamped_counts.go b/pkg/daemon/server/service/rater/timestamped_counts.go
index 99ef9b2f..ffa07567 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts.go
@@ -17,6 +17,7 @@ limitations under the License.
 package server
 
 import (
+	"fmt"
 	"sync"
 )
 
@@ -45,7 +46,14 @@ func (tc *TimestampedCounts) Update(podName string, count float64) {
 	tc.lock.Lock()
 	defer tc.lock.Unlock()
 	if count == CountNotAvailable {
-		delete(tc.podCounts, podName)
+		// we choose to skip updating when count is not available for the pod, instead of removing the pod from the map.
+		// imagine if the getTotalCount call fails to scrape the count metric, and it's NOT because the pod is down.
+		// in this case getTotalCount returns CountNotAvailable.
+		// if we remove the pod from the map and then the next scrape successfully gets the count, we can reach a state that in the timestamped counts,
+		// for this single pod, at t1, count is 123456, at t2, the map doesn't contain this pod and t3, count is 123457.
+		// when calculating the rate, as we sum up deltas among timestamps, we will get 123457 total delta instead of the real delta 1.
+		// one occurrence of such case can lead to extremely high rate and mess up the autoscaling.
+		// hence we'd rather keep the count as it is to avoid wrong rate calculation.
 		return
 	}
 	tc.podCounts[podName] = count
@@ -62,3 +70,12 @@ func (tc *TimestampedCounts) Snapshot() map[string]float64 {
 	}
 	return counts
 }
+
+// ToString returns a string representation of the TimestampedCounts
+// it's used for debugging purpose
+func (tc *TimestampedCounts) ToString() string {
+	tc.lock.RLock()
+	defer tc.lock.RUnlock()
+	res := fmt.Sprintf("{timestamp: %d, podCount: %v}", tc.timestamp, tc.podCounts)
+	return res
+}
diff --git a/server/cmd/start.go b/server/cmd/start.go
index e645268e..0ce2ba66 100644
--- a/server/cmd/start.go
+++ b/server/cmd/start.go
@@ -42,9 +42,11 @@ func Start(insecure bool, port int, namespaced bool, managedNamespace string, ba
 	router.Use(gin.LoggerWithConfig(gin.LoggerConfig{SkipPaths: []string{"/livez"}}))
 	router.RedirectTrailingSlash = true
 	router.Use(static.Serve(baseHref, static.LocalFile("./ui/build", true)))
-	router.NoRoute(func(c *gin.Context) {
-		c.File("./ui/build/index.html")
-	})
+	if baseHref != "/" {
+		router.NoRoute(func(c *gin.Context) {
+			c.File("./ui/build/index.html")
+		})
+	}
 	routes.Routes(router, routes.SystemInfo{ManagedNamespace: managedNamespace, Namespaced: namespaced})
 	router.Use(UrlRewrite(router))
 	server := http.Server{
diff --git a/ui/src/components/common/Breadcrumbs/index.tsx b/ui/src/components/common/Breadcrumbs/index.tsx
index d4ae013a..ff15621a 100644
--- a/ui/src/components/common/Breadcrumbs/index.tsx
+++ b/ui/src/components/common/Breadcrumbs/index.tsx
@@ -1,5 +1,5 @@
 import React, { useMemo } from "react";
-import { useLocation, Link } from "react-router-dom";
+import { Link, useLocation } from "react-router-dom";
 import MUIBreadcrumbs from "@mui/material/Breadcrumbs";
 import Typography from "@mui/material/Typography";
 
@@ -22,6 +22,8 @@ export function Breadcrumbs() {
       );
     }
     const pathParts = pathname.split("/");
+    // safety check for trailing slash
+    if (pathname.charAt(pathname.length - 1) === "/") pathParts.pop();
     switch (pathParts.length) {
       case 5: // pipeline view
         return [
diff --git a/ui/src/index.tsx b/ui/src/index.tsx
index 08151da8..1df95da2 100644
--- a/ui/src/index.tsx
+++ b/ui/src/index.tsx
@@ -21,7 +21,6 @@ const theme = createTheme({
       main: "#0077C5",
     },
   },
-
 });
 
 root.render(


Test patch:
diff --git a/pkg/daemon/server/service/rater/helper_test.go b/pkg/daemon/server/service/rater/helper_test.go
index 314f1f7c..1c7abb51 100644
--- a/pkg/daemon/server/service/rater/helper_test.go
+++ b/pkg/daemon/server/service/rater/helper_test.go
@@ -53,7 +53,7 @@ func TestUpdateCount(t *testing.T) {
 		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod2"])
 	})
 
-	t.Run("givenTimeExistsPodExistsCountNotAvailable_whenUpdate_thenRemovePod", func(t *testing.T) {
+	t.Run("givenTimeExistsPodExistsCountNotAvailable_whenUpdate_thenNotUpdatePod", func(t *testing.T) {
 		q := sharedqueue.New[*TimestampedCounts](1800)
 		tc := NewTimestampedCounts(TestTime)
 		tc.Update("pod1", 10.0)
@@ -62,7 +62,8 @@ func TestUpdateCount(t *testing.T) {
 		UpdateCount(q, TestTime, "pod1", CountNotAvailable)
 
 		assert.Equal(t, 1, q.Length())
-		assert.Equal(t, 0, len(q.Items()[0].podCounts))
+		assert.Equal(t, 1, len(q.Items()[0].podCounts))
+		assert.Equal(t, 10.0, q.Items()[0].podCounts["pod1"])
 	})
 
 	t.Run("givenTimeExistsPodNotExistsCountNotAvailable_whenUpdate_thenNoUpdate", func(t *testing.T) {
diff --git a/pkg/daemon/server/service/rater/timestamped_counts_test.go b/pkg/daemon/server/service/rater/timestamped_counts_test.go
index 2a602665..d5a8c494 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts_test.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts_test.go
@@ -38,7 +38,9 @@ func TestTimestampedCounts_Update(t *testing.T) {
 	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))
+	assert.Equal(t, 2, len(tc.podCounts))
+	assert.Equal(t, 20, int(tc.podCounts["pod1"]))
+	assert.Equal(t, 30, int(tc.podCounts["pod2"]))
 }
 
 func TestTimestampedCounts_Snapshot(t *testing.T) {