Repository: numaproj/numaflow
Pull request #1169 (2023-10-05 20:57:30)

chore: API cluster summary improvement
Get all the pipelines and isbsvcs at once instead of individual calls.
fixes:#1161

Example Response payload
```
{
    "data": [
        {
            "namespace": "numaflow-system",
            "pipelineSummary": {
                "active": {
                    "Healthy": 1,
                    "Warning": 0,
                    "Critical": 0
                },
                "inactive": 0
            },
            "isbServiceSummary": {
                "active": {
                    "Healthy": 1,
                    "Warning": 0,
                    "Critical": 0
                },
                "inactive": 0
            }
        },
        {
            "namespace": "default",
            "pipelineSummary": {
                "active": {
                    "Healthy": 2,
                    "Warning": 0,
                    "Critical": 0
                },
                "inactive": 1
            },
            "isbServiceSummary": {
                "active": {
                    "Healthy": 2,
                    "Warning": 0,
                    "Critical": 0
                },
                "inactive": 0
            }
        }
    ]
}
```


Patch:
diff --git a/cmd/commands/root.go b/cmd/commands/root.go
index 9702c583..35569cb9 100644
--- a/cmd/commands/root.go
+++ b/cmd/commands/root.go
@@ -48,5 +48,5 @@ func init() {
 	rootCmd.AddCommand(NewWebhookCommand())
 	rootCmd.AddCommand(NewSideInputsInitCommand())
 	rootCmd.AddCommand(NewSideInputsManagerCommand())
-	rootCmd.AddCommand(NewSideInputsWatcherCommand())
+	rootCmd.AddCommand(NewSideInputsSynchronizerCommand())
 }
diff --git a/cmd/commands/side_inputs_watcher.go b/cmd/commands/side_inputs_synchronizer.go
similarity index 88%
rename from cmd/commands/side_inputs_watcher.go
rename to cmd/commands/side_inputs_synchronizer.go
index 7be5889e..5bda7ae9 100644
--- a/cmd/commands/side_inputs_watcher.go
+++ b/cmd/commands/side_inputs_synchronizer.go
@@ -28,15 +28,15 @@ import (
 	"github.com/numaproj/numaflow/pkg/sideinputs/synchronizer"
 )
 
-func NewSideInputsWatcherCommand() *cobra.Command {
+func NewSideInputsSynchronizerCommand() *cobra.Command {
 	var (
 		isbSvcType      string
 		sideInputsStore string
 		sideInputs      []string
 	)
 	command := &cobra.Command{
-		Use:   "side-inputs-watcher",
-		Short: "Start the Side Inputs Watcher",
+		Use:   "side-inputs-synchronizer",
+		Short: "Start the Side Inputs Synchronizer",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			pipelineName, defined := os.LookupEnv(dfv1.EnvPipelineName)
 			if !defined {
@@ -47,7 +47,7 @@ func NewSideInputsWatcherCommand() *cobra.Command {
 				return fmt.Errorf("no side inputs are defined for this vertex")
 			}
 
-			logger := logging.NewLogger().Named("side-inputs-watcher").With("pipeline", pipelineName)
+			logger := logging.NewLogger().Named("side-inputs-synchronizer").With("pipeline", pipelineName)
 			ctx := logging.WithLogger(signals.SetupSignalHandler(), logger)
 			sideInputsWatcher := synchronizer.NewSideInputsSynchronizer(dfv1.ISBSvcType(isbSvcType), pipelineName, sideInputsStore, sideInputs)
 			return sideInputsWatcher.Start(ctx)
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index 094cbd5e..64a992f1 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -71,7 +71,7 @@ const (
 	CtrUdtransformer     = "transformer"
 	CtrUdSideInput       = "udsi"
 	CtrInitSideInputs    = "init-side-inputs"
-	CtrSideInputsWatcher = "side-inputs-watcher"
+	CtrSideInputsWatcher = "side-inputs-synchronizer"
 
 	// components
 	ComponentISBSvc           = "isbsvc"
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index 271a8871..805e5ab7 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -277,7 +277,7 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) {
 			Image:           req.Image,
 			ImagePullPolicy: req.PullPolicy,
 			Resources:       standardResources,
-			Args:            []string{"side-inputs-watcher", "--isbsvc-type=" + string(req.ISBSvcType), "--side-inputs-store=" + req.SideInputsStoreName, "--side-inputs=" + strings.Join(v.Spec.SideInputs, ",")},
+			Args:            []string{"side-inputs-synchronizer", "--isbsvc-type=" + string(req.ISBSvcType), "--side-inputs-store=" + req.SideInputsStoreName, "--side-inputs=" + strings.Join(v.Spec.SideInputs, ",")},
 		}
 		sideInputsWatcher.Env = append(sideInputsWatcher.Env, v.commonEnvs()...)
 		if x := v.Spec.SideInputsContainerTemplate; x != nil {
diff --git a/pkg/shared/kvs/jetstream/kv_store.go b/pkg/shared/kvs/jetstream/kv_store.go
index ce52ea30..49aeaee8 100644
--- a/pkg/shared/kvs/jetstream/kv_store.go
+++ b/pkg/shared/kvs/jetstream/kv_store.go
@@ -202,7 +202,7 @@ func (jss *jetStreamStore) Watch(ctx context.Context) (<-chan kvs.KVEntry, <-cha
 					continue
 				}
 				if value == nil {
-					jss.log.Infow("watcher initialization and subscription got nil value")
+					jss.log.Infow("Watcher initialization and subscription got nil value")
 					continue
 				}
 				jss.previousFetchTime = value.Created()
diff --git a/pkg/sideinputs/initializer/initializer.go b/pkg/sideinputs/initializer/initializer.go
index d0bc9bb1..be0051dc 100644
--- a/pkg/sideinputs/initializer/initializer.go
+++ b/pkg/sideinputs/initializer/initializer.go
@@ -110,9 +110,9 @@ func startSideInputInitializer(ctx context.Context, store kvs.KVStorer, mountPat
 			m[value.Key()] = value.Value()
 			// Wait for the data to be ready in the side input store, and then copy it to the disk
 			if gotAllSideInputVals(sideInputs, m) {
-				for sideInput := range m {
+				for _, sideInput := range sideInputs {
 					p := path.Join(mountPath, sideInput)
-					log.Info("Initializing Side Input data for %q", p)
+					log.Infof("Initializing Side Input data for %q", sideInput)
 					err := utils.UpdateSideInputFile(ctx, p, m[sideInput])
 					if err != nil {
 						return fmt.Errorf("failed to update Side Input value, %w", err)
@@ -132,9 +132,6 @@ func startSideInputInitializer(ctx context.Context, store kvs.KVStorer, mountPat
 // gotAllSideInputVals checks if values for all side-inputs
 // have been received from the KV bucket
 func gotAllSideInputVals(sideInputs []string, m map[string][]byte) bool {
-	if len(sideInputs) != len(m) {
-		return false
-	}
 	for _, sideInput := range sideInputs {
 		if _, ok := m[sideInput]; !ok {
 			return false
diff --git a/pkg/sideinputs/synchronizer/synchronizer.go b/pkg/sideinputs/synchronizer/synchronizer.go
index 86b1a814..8b2d359b 100644
--- a/pkg/sideinputs/synchronizer/synchronizer.go
+++ b/pkg/sideinputs/synchronizer/synchronizer.go
@@ -29,6 +29,7 @@ import (
 	"github.com/numaproj/numaflow/pkg/shared/kvs"
 	"github.com/numaproj/numaflow/pkg/shared/kvs/jetstream"
 	"github.com/numaproj/numaflow/pkg/shared/logging"
+	sharedutil "github.com/numaproj/numaflow/pkg/shared/util"
 	"github.com/numaproj/numaflow/pkg/sideinputs/utils"
 )
 
@@ -62,7 +63,7 @@ func (sis *sideInputsSynchronizer) Start(ctx context.Context) error {
 	)
 
 	log := logging.FromContext(ctx)
-	log.Infow("Starting Side Inputs Watcher", zap.Strings("sideInputs", sis.sideInputs))
+	log.Infow("Starting Side Inputs Synchronizer", zap.Strings("sideInputs", sis.sideInputs))
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -85,26 +86,29 @@ func (sis *sideInputsSynchronizer) Start(ctx context.Context) error {
 	default:
 		return fmt.Errorf("unrecognized isbsvc type %q", sis.isbSvcType)
 	}
-	go startSideInputSynchronizer(ctx, sideInputStore, dfv1.PathSideInputsMount)
+	go startSideInputSynchronizer(ctx, sideInputStore, sis.sideInputs, dfv1.PathSideInputsMount)
 	<-ctx.Done()
 	return nil
 }
 
 // startSideInputSynchronizer watches the Side Input KV store for any changes
 // and writes the updated value to the mount volume.
-func startSideInputSynchronizer(ctx context.Context, watch kvs.KVStorer, mountPath string) {
+func startSideInputSynchronizer(ctx context.Context, watch kvs.KVStorer, sideInputs []string, mountPath string) {
 	log := logging.FromContext(ctx)
 	watchCh, stopped := watch.Watch(ctx)
 	for {
 		select {
 		case <-stopped:
-			log.Info("Side Input watcher stopped")
+			log.Info("Side Input Synchronizer stopped")
 			return
 		case value := <-watchCh:
 			if value == nil {
 				log.Warnw("nil value received from Side Input watcher")
 				continue
 			}
+			if !sharedutil.StringSliceContains(sideInputs, value.Key()) {
+				continue
+			}
 			log.Infow("Side Input value received ",
 				zap.String("key", value.Key()), zap.String("value", string(value.Value())))
 			p := path.Join(mountPath, value.Key())
diff --git a/server/apis/v1/handler.go b/server/apis/v1/handler.go
index 7628ba7a..a5b89383 100644
--- a/server/apis/v1/handler.go
+++ b/server/apis/v1/handler.go
@@ -94,57 +94,63 @@ func (h *handler) ListNamespaces(c *gin.Context) {
 
 // GetClusterSummary summarizes information of all the namespaces in a cluster and wrapped the result in a list.
 func (h *handler) GetClusterSummary(c *gin.Context) {
-	namespaces, err := getAllNamespaces(h)
+	type namespaceSummary struct {
+		pipelineSummary PipelineSummary
+		isbsvcSummary   IsbServiceSummary
+	}
+	var namespaceSummaryMap = make(map[string]namespaceSummary)
+
+	// get pipeline summary
+	pipelineList, err := h.numaflowClient.Pipelines("").List(context.Background(), metav1.ListOptions{})
 	if err != nil {
 		h.respondWithError(c, fmt.Sprintf("Failed to fetch cluster summary, %s", err.Error()))
 		return
 	}
-	var clusterSummary ClusterSummaryResponse
-	// TODO(API): need a more efficient solution
-	// Loop over the namespaces to get status
-	for _, ns := range namespaces {
-		// Fetch pipeline summary
-		pipelines, err := getPipelines(h, ns)
+	for _, pipeline := range pipelineList.Items {
+		var summary namespaceSummary
+		if value, ok := namespaceSummaryMap[pipeline.Namespace]; ok {
+			summary = value
+		}
+		status, err := getPipelineStatus(&pipeline)
 		if err != nil {
 			h.respondWithError(c, fmt.Sprintf("Failed to fetch cluster summary, %s", err.Error()))
 			return
 		}
-
-		var pipeSummary PipelineSummary
-		var pipeActiveSummary ActiveStatus
-		// Loop over the pipelines and get the status
-		for _, pl := range pipelines {
-			if pl.Status == PipelineStatusInactive {
-				pipeSummary.Inactive++
-			} else {
-				pipeActiveSummary.increment(pl.Status)
-
-			}
+		if status == PipelineStatusInactive {
+			summary.pipelineSummary.Inactive++
+		} else {
+			summary.pipelineSummary.Active.increment(status)
 		}
-		pipeSummary.Active = pipeActiveSummary
+		namespaceSummaryMap[pipeline.Namespace] = summary
+	}
 
-		// Fetch ISB service summary
-		isbSvcs, err := getIsbServices(h, ns)
-		if err != nil {
-			h.respondWithError(c, fmt.Sprintf("Failed to fetch cluster summary, %s", err.Error()))
-			return
+	// get isbsvc summary
+	isbsvcList, err := h.numaflowClient.InterStepBufferServices("").List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		h.respondWithError(c, fmt.Sprintf("Failed to fetch cluster summary, %s", err.Error()))
+		return
+	}
+	for _, isbsvc := range isbsvcList.Items {
+		var summary namespaceSummary
+		if value, ok := namespaceSummaryMap[isbsvc.Namespace]; ok {
+			summary = value
 		}
-
-		var isbSummary IsbServiceSummary
-		var isbActiveSummary ActiveStatus
-		// loop over the ISB services and get the status
-		for _, isb := range isbSvcs {
-			if isb.Status == ISBServiceStatusInactive {
-				isbSummary.Inactive++
-			} else {
-				isbActiveSummary.increment(isb.Status)
-			}
+		// TODO(API) : Get the current status of the ISB service
+		status := ISBServiceStatusHealthy
+		if status == ISBServiceStatusInactive {
+			summary.isbsvcSummary.Inactive++
+		} else {
+			summary.isbsvcSummary.Active.increment(status)
 		}
-		isbSummary.Active = isbActiveSummary
-		clusterSummary = append(clusterSummary, NewClusterSummary(ns, pipeSummary, isbSummary))
+		namespaceSummaryMap[isbsvc.Namespace] = summary
 	}
-	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, clusterSummary))
 
+	// get cluster summary
+	var clusterSummary ClusterSummaryResponse
+	for name, summary := range namespaceSummaryMap {
+		clusterSummary = append(clusterSummary, NewClusterSummary(name, summary.pipelineSummary, summary.isbsvcSummary))
+	}
+	c.JSON(http.StatusOK, NewNumaflowAPIResponse(nil, clusterSummary))
 }
 
 // CreatePipeline is used to create a given pipeline


Test patch:
diff --git a/pkg/sideinputs/synchronizer/synchronizer_test.go b/pkg/sideinputs/synchronizer/synchronizer_test.go
index e07bc6ab..9338929e 100644
--- a/pkg/sideinputs/synchronizer/synchronizer_test.go
+++ b/pkg/sideinputs/synchronizer/synchronizer_test.go
@@ -93,7 +93,7 @@ func TestSideInputsValueUpdates(t *testing.T) {
 
 	bucketName := keyspace
 	sideInputStore, _ := jetstream.NewKVJetStreamKVStore(ctx, bucketName, nc)
-	go startSideInputSynchronizer(ctx, sideInputStore, mountPath)
+	go startSideInputSynchronizer(ctx, sideInputStore, sideInputs, mountPath)
 	for x := range sideInputs {
 		_, err = kv.Put(sideInputs[x], []byte(dataTest[x]))
 		if err != nil {