Repository: numaproj/numaflow
Pull request #1014 (2023-09-01 18:34:17)

fix: reconcile headless services before pods
The problem: when we start a pipeline and create vertex A with 10 pods and the resource quota is 8, the vertex reconciler will keep creating new pods and start to fail reconciliation when trying to create the 9th pod. Currently in our vertex reconciler, the pod reconciliation happens before service hence the vertex reconciler will NEVER create the headless service for the vertex. 

The headless service is used for exposing metrics like processing rate, even if we have 8 pods which don't meet our specification, it shouldn't stop us from emitting metrics for the vertex.

This change moves the service creation to before pod creation to mitigate the issue.

Fixes: #1006 

<!--

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/docs/development/debugging.md b/docs/development/debugging.md
index c5cc18bf..6baa6df2 100644
--- a/docs/development/debugging.md
+++ b/docs/development/debugging.md
@@ -32,6 +32,23 @@ spec:
       to: out
 ```
 
+To enable debug logs in the daemon pod, set environment variable `NUMAFLOW_DEBUG` to `true` for the daemon pod.
+For example:
+
+```yaml
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: simple-pipeline
+spec:
+  templates:
+    daemon:
+      containerTemplate:
+        env:
+          - name: NUMAFLOW_DEBUG
+            value: "true" # DO NOT forget the double quotes!!!
+```
+
 ## Profiling
 
 If your pipeline is running with `NUMAFLOW_DEBUG` then `pprof` is enabled in the Vertex Pod. You
diff --git a/pkg/daemon/server/service/rater/pod_tracker.go b/pkg/daemon/server/service/rater/pod_tracker.go
index 45ae8d72..8a5f1a67 100644
--- a/pkg/daemon/server/service/rater/pod_tracker.go
+++ b/pkg/daemon/server/service/rater/pod_tracker.go
@@ -56,7 +56,7 @@ func NewPodTracker(ctx context.Context, p *v1alpha1.Pipeline, opts ...PodTracker
 			Timeout: time.Second,
 		},
 		activePods:      NewUniqueStringList(),
-		refreshInterval: 30 * time.Second, // Default refresh interval for updating active pod set
+		refreshInterval: 30 * time.Second, // Default refresh interval for updating the active pod set
 	}
 
 	for _, opt := range opts {
@@ -77,7 +77,7 @@ func WithRefreshInterval(d time.Duration) PodTrackerOption {
 }
 
 func (pt *PodTracker) Start(ctx context.Context) error {
-	pt.log.Infof("Starting tracking active pods for pipeline %s...", pt.pipeline.Name)
+	pt.log.Debugf("Starting tracking active pods for pipeline %s...", pt.pipeline.Name)
 	go func() {
 		ticker := time.NewTicker(pt.refreshInterval)
 		defer ticker.Stop()
@@ -115,7 +115,7 @@ func (pt *PodTracker) Start(ctx context.Context) error {
 }
 
 // LeastRecentlyUsed returns the least recently used pod from the active pod list.
-// if there is no active pods, it returns an empty string.
+// if there are no active pods, it returns an empty string.
 func (pt *PodTracker) LeastRecentlyUsed() string {
 	if e := pt.activePods.Front(); e != "" {
 		pt.activePods.MoveToBack(e)
@@ -141,17 +141,18 @@ func (pt *PodTracker) getPodKey(index int, vertexName string, vertexType string)
 
 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:2469/metrics
+	// example for 0th pod: https://simple-pipeline-in-0.simple-pipeline-in-headless.default.svc:2469/metrics
 	url := fmt.Sprintf("https://%s.%s.%s.svc:%v/metrics", podName, pt.pipeline.Name+"-"+vertexName+"-headless", pt.pipeline.Namespace, v1alpha1.VertexMetricsPort)
 	resp, err := pt.httpClient.Head(url)
 	if 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.
+		// 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.
 		pt.log.Debugf("Sending HEAD request to pod %s is unsuccessful: %v, treating the pod as inactive", podName, err)
 		return false
 	}
+	pt.log.Debugf("Sending HEAD request to pod %s is successful, treating the pod as active", podName)
 	_ = resp.Body.Close()
 	return true
 }
diff --git a/pkg/daemon/server/service/rater/rater.go b/pkg/daemon/server/service/rater/rater.go
index b05805b8..a33aca71 100644
--- a/pkg/daemon/server/service/rater/rater.go
+++ b/pkg/daemon/server/service/rater/rater.go
@@ -38,7 +38,8 @@ type Ratable interface {
 }
 
 // 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
+// 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.
@@ -107,7 +108,7 @@ func NewRater(ctx context.Context, p *v1alpha1.Pipeline, opts ...Option) *Rater
 	return &rater
 }
 
-// Function monitor() defines each of the worker's job.
+// Function monitor() defines each of the worker's jobs.
 // 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)
@@ -200,8 +201,8 @@ func (r *Rater) Start(ctx context.Context) error {
 	}
 }
 
-// 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
+// sleep function uses a select statement to check if the context is canceled before sleeping for the given duration
+// it helps ensure the sleep will be released when the context is canceled, allowing the goroutine to exit gracefully
 func sleep(ctx context.Context, duration time.Duration) {
 	select {
 	case <-ctx.Done():
@@ -257,12 +258,15 @@ func (r *Rater) getPodReadCounts(vertexName, vertexType, podName string) *PodRea
 
 // GetRates returns the processing rates of the vertex partition in the format of lookback second to rate mappings
 func (r *Rater) GetRates(vertexName, partitionName string) map[string]float64 {
+	r.log.Debugf("Getting rates for vertex %s, partition %s", vertexName, partitionName)
+	r.log.Debugf("Current timestampedPodCounts for vertex %s is: %v", vertexName, r.timestampedPodCounts[vertexName])
 	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, partitionName)
 		result[n] = r
 	}
+	r.log.Debugf("Got rates for vertex %s, partition %s: %v", vertexName, partitionName, result)
 	return result
 }
 
diff --git a/pkg/daemon/server/service/rater/timestamped_counts.go b/pkg/daemon/server/service/rater/timestamped_counts.go
index f91033e1..591edd65 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts.go
@@ -69,11 +69,10 @@ func (tc *TimestampedCounts) PodPartitionCountSnapshot() map[string]map[string]f
 	return counts
 }
 
-// ToString returns a string representation of the TimestampedCounts
+// String returns a string representation of the TimestampedCounts
 // it's used for debugging purpose
-func (tc *TimestampedCounts) ToString() string {
+func (tc *TimestampedCounts) String() string {
 	tc.lock.RLock()
 	defer tc.lock.RUnlock()
-	res := fmt.Sprintf("{timestamp: %d, podPartitionCount: %v}", tc.timestamp, tc.podPartitionCount)
-	return res
+	return fmt.Sprintf("{timestamp: %d, podPartitionCount: %v}", tc.timestamp, tc.podPartitionCount)
 }
diff --git a/pkg/reconciler/vertex/controller.go b/pkg/reconciler/vertex/controller.go
index 64889ee9..0442eae4 100644
--- a/pkg/reconciler/vertex/controller.go
+++ b/pkg/reconciler/vertex/controller.go
@@ -85,7 +85,7 @@ func (r *vertexReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 	return result, err
 }
 
-// reconcile does the real logic
+// reconcile does the real logic.
 func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (ctrl.Result, error) {
 	log := logging.FromContext(ctx)
 	vertexKey := scaling.KeyOfVertex(*vertex)
@@ -154,12 +154,66 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 		}
 	}
 
+	// Create services
+	// Note: We purposely put service reconciliation before pod,
+	// to prevent pod reconciliation failure from blocking service creation.
+	// It's ok to keep failing to scale up/down pods (e.g., due to quota),
+	// but without services, certain platform functionalities will be broken.
+	// E.g., the vertex processing rate calculation relies on the headless service to determine the number of active pods.
+	existingSvcs, err := r.findExistingServices(ctx, vertex)
+	if err != nil {
+		r.markPhaseLogEvent(vertex, log, "FindExistingSvcsFailed", err.Error(), "Failed to find existing services", zap.Error(err))
+		return ctrl.Result{}, err
+	}
+	for _, s := range vertex.GetServiceObjs() {
+		svcHash := sharedutil.MustHash(s.Spec)
+		s.Annotations = map[string]string{dfv1.KeyHash: svcHash}
+		needToCreate := false
+		if existingSvc, existing := existingSvcs[s.Name]; existing {
+			if existingSvc.GetAnnotations()[dfv1.KeyHash] != svcHash {
+				if err := r.client.Delete(ctx, &existingSvc); err != nil {
+					if !apierrors.IsNotFound(err) {
+						r.markPhaseLogEvent(vertex, log, "DelSvcFailed", err.Error(), "Failed to delete existing service", zap.String("service", existingSvc.Name), zap.Error(err))
+						return ctrl.Result{}, err
+					}
+				} else {
+					log.Infow("Deleted a stale service to recreate", zap.String("service", existingSvc.Name))
+				}
+				needToCreate = true
+			}
+			delete(existingSvcs, s.Name)
+		} else {
+			needToCreate = true
+		}
+		if needToCreate {
+			if err := r.client.Create(ctx, s); err != nil {
+				if apierrors.IsAlreadyExists(err) {
+					continue
+				}
+				r.markPhaseLogEvent(vertex, log, "CreateSvcFailed", err.Error(), "Failed to create a service", zap.String("service", s.Name), zap.Error(err))
+				return ctrl.Result{}, err
+			} else {
+				log.Infow("Succeeded to create a service", zap.String("service", s.Name))
+			}
+		}
+	}
+	for _, v := range existingSvcs { // clean up stale services
+		if err := r.client.Delete(ctx, &v); err != nil {
+			if !apierrors.IsNotFound(err) {
+				r.markPhaseLogEvent(vertex, log, "DelSvcFailed", err.Error(), "Failed to delete service not in use", zap.String("service", v.Name), zap.Error(err))
+				return ctrl.Result{}, err
+			}
+		} else {
+			log.Infow("Deleted a stale service", zap.String("service", v.Name))
+		}
+	}
+
 	pipeline := &dfv1.Pipeline{}
 	if err := r.client.Get(ctx, types.NamespacedName{Namespace: vertex.Namespace, Name: vertex.Spec.PipelineName}, pipeline); err != nil {
 		r.markPhaseLogEvent(vertex, log, "GetPipelineFailed", err.Error(), "Failed to get pipeline object", zap.Error(err))
 		return ctrl.Result{}, err
 	}
-
+	// Create pods
 	existingPods, err := r.findExistingPods(ctx, vertex)
 	if err != nil {
 		r.markPhaseLogEvent(vertex, log, "FindExistingPodFailed", err.Error(), "Failed to find existing pods", zap.Error(err))
@@ -258,55 +312,6 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 	selector, _ := labels.Parse(dfv1.KeyPipelineName + "=" + vertex.Spec.PipelineName + "," + dfv1.KeyVertexName + "=" + vertex.Spec.Name)
 	vertex.Status.Selector = selector.String()
 
-	// create services
-	existingSvcs, err := r.findExistingServices(ctx, vertex)
-	if err != nil {
-		r.markPhaseLogEvent(vertex, log, "FindExistingSvcsFailed", err.Error(), "Failed to find existing services", zap.Error(err))
-		return ctrl.Result{}, err
-	}
-	for _, s := range vertex.GetServiceObjs() {
-		svcHash := sharedutil.MustHash(s.Spec)
-		s.Annotations = map[string]string{dfv1.KeyHash: svcHash}
-		needToCreate := false
-		if existingSvc, existing := existingSvcs[s.Name]; existing {
-			if existingSvc.GetAnnotations()[dfv1.KeyHash] != svcHash {
-				if err := r.client.Delete(ctx, &existingSvc); err != nil {
-					if !apierrors.IsNotFound(err) {
-						r.markPhaseLogEvent(vertex, log, "DelSvcFailed", err.Error(), "Failed to delete existing service", zap.String("service", existingSvc.Name), zap.Error(err))
-						return ctrl.Result{}, err
-					}
-				} else {
-					log.Infow("Deleted a stale service to recreate", zap.String("service", existingSvc.Name))
-				}
-				needToCreate = true
-			}
-			delete(existingSvcs, s.Name)
-		} else {
-			needToCreate = true
-		}
-		if needToCreate {
-			if err := r.client.Create(ctx, s); err != nil {
-				if apierrors.IsAlreadyExists(err) {
-					continue
-				}
-				r.markPhaseLogEvent(vertex, log, "CreateSvcFailed", err.Error(), "Failed to create a service", zap.String("service", s.Name), zap.Error(err))
-				return ctrl.Result{}, err
-			} else {
-				log.Infow("Succeeded to create a service", zap.String("service", s.Name))
-			}
-		}
-	}
-	for _, v := range existingSvcs { // clean up stale services
-		if err := r.client.Delete(ctx, &v); err != nil {
-			if !apierrors.IsNotFound(err) {
-				r.markPhaseLogEvent(vertex, log, "DelSvcFailed", err.Error(), "Failed to delete service not in use", zap.String("service", v.Name), zap.Error(err))
-				return ctrl.Result{}, err
-			}
-		} else {
-			log.Infow("Deleted a stale service", zap.String("service", v.Name))
-		}
-	}
-
 	vertex.Status.MarkPhaseRunning()
 	return ctrl.Result{}, nil
 }
@@ -381,8 +386,8 @@ func (r *vertexReconciler) buildPodSpec(vertex *dfv1.Vertex, pl *dfv1.Pipeline,
 		})
 	}
 
-	bfs := []string{}
-	bks := []string{}
+	var bfs []string
+	var bks []string
 	// Only source vertices need to check all the pipeline buffers and buckets
 	if vertex.IsASource() {
 		bfs = append(bfs, pl.GetAllBuffers()...)
diff --git a/pkg/shared/queue/overflow_queue.go b/pkg/shared/queue/overflow_queue.go
index 29fdaa65..f81ba6dd 100644
--- a/pkg/shared/queue/overflow_queue.go
+++ b/pkg/shared/queue/overflow_queue.go
@@ -16,7 +16,10 @@ limitations under the License.
 
 package queue
 
-import "sync"
+import (
+	"fmt"
+	"sync"
+)
 
 // OverflowQueue is a thread safe queue implementation with max size, and the oldest elements automatically overflow.
 type OverflowQueue[T any] struct {
@@ -64,6 +67,16 @@ func (q *OverflowQueue[T]) Length() int {
 	return len(q.elements)
 }
 
+// String returns a string representation of the queue
+func (q *OverflowQueue[T]) String() string {
+	res := "["
+	for _, e := range q.Items() {
+		res += fmt.Sprintf("%v,", e)
+	}
+	res += "]"
+	return res
+}
+
 func reverse[T any](input []T) []T {
 	if len(input) == 0 {
 		return input


Test patch:
diff --git a/pkg/daemon/server/service/rater/timestamped_counts_test.go b/pkg/daemon/server/service/rater/timestamped_counts_test.go
index 67384872..048874d8 100644
--- a/pkg/daemon/server/service/rater/timestamped_counts_test.go
+++ b/pkg/daemon/server/service/rater/timestamped_counts_test.go
@@ -24,8 +24,10 @@ import (
 
 func TestNewTimestampedCounts(t *testing.T) {
 	tc := NewTimestampedCounts(TestTime)
+	tc.Update(&PodReadCount{"pod1", map[string]float64{"partition1": 10.0}})
 	assert.Equal(t, int64(TestTime), tc.timestamp)
-	assert.Equal(t, 0, len(tc.podPartitionCount))
+	assert.Equal(t, 1, len(tc.podPartitionCount))
+	assert.Equal(t, "{timestamp: 1620000000, podPartitionCount: map[pod1:map[partition1:10]]}", tc.String())
 }
 
 func TestTimestampedCounts_Update(t *testing.T) {