update: reworked watching workflow execution to keep watching until the workflow is done or failed.

pkg/workflow_execution.go

@@ -637,7 +637,7 @@ func (c *Client) WatchWorkflowExecution(namespace, uid string) (<-chan *Workflow
 			"Namespace": namespace,
 			"UID":       uid,
 			"Error":     err.Error(),
-		}).Error("Workflow template not found.")
+		}).Errorf("Workflow execution not found for namespace: %v, uid: %v).", namespace, uid)
 		return nil, util.NewUserError(codes.NotFound, "Workflow not found.")
 	}
 
@@ -655,76 +655,19 @@ func (c *Client) WatchWorkflowExecution(namespace, uid string) (<-chan *Workflow
 	}
 
 	workflowWatcher := make(chan *WorkflowExecution)
-	ticker := time.NewTicker(time.Second)
 	go func() {
-		var workflow *wfv1.Workflow
-		ok := true
 		var next watch.Event
+		done := false
 
-		for {
-			select {
-			case next = <-watcher.ResultChan():
-				// -------
-
-				if next.Type == "" {
-					workflow, err = c.ArgoprojV1alpha1().Workflows(namespace).Get(uid, metav1.GetOptions{})
-					if err != nil {
-						log.WithFields(log.Fields{
-							"Namespace": namespace,
-							"UID":       uid,
-							"Workflow":  workflow,
-							"Error":     err.Error(),
-						}).Error("Unable to get workflow.")
-					}
-
-					if workflow.Status.Phase == wfv1.NodeRunning {
-						watcher, err = c.ArgoprojV1alpha1().Workflows(namespace).Watch(metav1.ListOptions{
-							FieldSelector: fieldSelector.String(),
-						})
-						if err != nil {
-							log.WithFields(log.Fields{
-								"Namespace": namespace,
-								"UID":       uid,
-								"Error":     err.Error(),
-							}).Error("Watch Workflow error.")
-						} else {
-							continue
-						}
-					}
-
-				}
-
-				// -------
-				workflow, ok = next.Object.(*wfv1.Workflow)
-			case <-ticker.C:
-				time.Sleep(time.Millisecond * 200)
-				continue
-			}
-
-			if workflow == nil && ok {
-				continue
-			}
-
-			if workflow == nil && !ok {
-				workflow, err = c.ArgoprojV1alpha1().Workflows(namespace).Get(uid, metav1.GetOptions{})
-				if err != nil {
-					log.WithFields(log.Fields{
-						"Namespace": namespace,
-						"UID":       uid,
-						"Workflow":  workflow,
-						"Error":     err.Error(),
-					}).Error("Unable to get workflow.")
-
+		for !done {
+			for next = range watcher.ResultChan() {
+				workflow, ok := next.Object.(*wfv1.Workflow)
+				if !ok {
+					done = true
 					break
 				}
-
 				if workflow == nil {
-					break
-				}
-			}
-
-			if workflow == nil {
-				log.Printf("We hit a bad spot where the workflow is nil")
+					continue
 				}
 
 				manifest, err := json.Marshal(workflow)
@@ -735,6 +678,7 @@ func (c *Client) WatchWorkflowExecution(namespace, uid string) (<-chan *Workflow
 						"Workflow":  workflow,
 						"Error":     err.Error(),
 					}).Error("Error with trying to JSON Marshal workflow.Status.")
+					done = true
 					break
 				}
 
@@ -747,14 +691,50 @@ func (c *Client) WatchWorkflowExecution(namespace, uid string) (<-chan *Workflow
 					Manifest:   string(manifest),
 				}
 
-			if !workflow.Status.FinishedAt.IsZero() || !ok {
+				if !workflow.Status.FinishedAt.IsZero() {
+					done = true
 					break
 				}
 			}
 
-		close(workflowWatcher)
+			// We want to continue to watch the workflow until it is done, or an error occurred
+			// If it is not done, create a new watch and continue watching.
+			if !done {
+				workflow, err := c.ArgoprojV1alpha1().Workflows(namespace).Get(uid, metav1.GetOptions{})
+				if err != nil {
+					log.WithFields(log.Fields{
+						"Namespace": namespace,
+						"UID":       uid,
+						"Workflow":  workflow,
+						"Error":     err.Error(),
+					}).Error("Unable to get workflow.")
+
+					done = true
+					break
+				}
+
+				if workflow.Status.Phase == wfv1.NodeRunning {
+					watcher, err = c.ArgoprojV1alpha1().Workflows(namespace).Watch(metav1.ListOptions{
+						FieldSelector: fieldSelector.String(),
+					})
+					if err != nil {
+						log.WithFields(log.Fields{
+							"Namespace": namespace,
+							"UID":       uid,
+							"Error":     err.Error(),
+						}).Error("Watch Workflow error.")
+						done = true
+						break
+					}
+				} else {
+					done = true
+					break
+				}
+			}
+		}
+
 		watcher.Stop()
-		ticker.Stop()
+		close(workflowWatcher)
 	}()
 
 	return workflowWatcher, nil

server/workflow_server.go

@@ -180,14 +180,7 @@ func (s *WorkflowServer) WatchWorkflowExecution(req *api.WatchWorkflowExecutionR
 		return err
 	}
 
-	wf := &v1.WorkflowExecution{}
-	ticker := time.NewTicker(time.Second)
-	for {
-		select {
-		case wf = <-watcher:
-		case <-ticker.C:
-		}
-
+	for wf := range watcher {
 		if wf == nil {
 			break
 		}