fix: incorrect side inputs initialization ready check logic

Signed-off-by: Derek Wang <[email protected]>
numaproj · Oct 5, 2023 · 85eddd1 · 85eddd1
1 parent 26b6472
commit 85eddd1
Show file tree

Hide file tree

Showing 6 changed files with 9 additions and 12 deletions.
diff --git 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 → cmd/commands/side_inputs_synchronizer.go b/cmd/commands/side_inputs_watcher.go → 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
@@ -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
@@ -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
@@ -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
@@ -112,7 +112,7 @@ func startSideInputInitializer(ctx context.Context, store kvs.KVStorer, mountPat
 			if gotAllSideInputVals(sideInputs, m) {
 				for sideInput := range m {
 					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