fix(CLI): Re-establish workflow watch on disconnect. Fixes argoproj#2796

(argoproj#2830)
highb · Apr 27, 2020 · d7f8e0c · d7f8e0c
1 parent 31358d6
commit d7f8e0c
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 30 deletions.
diff --git a/server/workflow/workflow_server.go b/server/workflow/workflow_server.go
@@ -155,53 +155,59 @@ func (s *workflowServer) WatchWorkflows(req *workflowpkg.WatchWorkflowsRequest,
  if req.ListOptions != nil {
  opts = *req.ListOptions
  }
- watch, err := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace).Watch(s.withInstanceID(opts))
+ opts = s.withInstanceID(opts)
+ wfIf := wfClient.ArgoprojV1alpha1().Workflows(req.Namespace)
+ watch, err := wfIf.Watch(opts)
  if err != nil {
  return err
  }
  defer watch.Stop()
  ctx := ws.Context()
 
  log.Debug("Piping events to channel")
+ defer log.Debug("Result channel done")
 
- for next := range watch.ResultChan() {
+ for {
  select {
  case <-ctx.Done():
  return ctx.Err()
- default:
- }
- log.Debug("Received event")
- wf, ok := next.Object.(*v1alpha1.Workflow)
- if !ok {
- return fmt.Errorf("watch object was not a workflow %v", reflect.TypeOf(next.Object))
- }
- logCtx := log.WithFields(log.Fields{"workflow": wf.Name, "type": next.Type, "phase": wf.Status.Phase})
- err := packer.DecompressWorkflow(wf)
- if err != nil {
- return err
- }
- if wf.Status.IsOffloadNodeStatus() {
- if s.offloadNodeStatusRepo.IsEnabled() {
- offloadedNodes, err := s.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
+ case event, open := <-watch.ResultChan():
+ if !open {
+ log.Info("Re-establishing workflow watch")
+ watch, err = wfIf.Watch(opts)
  if err != nil {
  return err
  }
- wf.Status.Nodes = offloadedNodes
- } else {
- log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabled)
+ continue
+ }
+ log.Debug("Received event")
+ wf, ok := event.Object.(*v1alpha1.Workflow)
+ if !ok {
+ return fmt.Errorf("watch object was not a workflow %v", reflect.TypeOf(event.Object))
+ }
+ logCtx := log.WithFields(log.Fields{"workflow": wf.Name, "type": event.Type, "phase": wf.Status.Phase})
+ err := packer.DecompressWorkflow(wf)
+ if err != nil {
+ return err
+ }
+ if wf.Status.IsOffloadNodeStatus() {
+ if s.offloadNodeStatusRepo.IsEnabled() {
+ offloadedNodes, err := s.offloadNodeStatusRepo.Get(string(wf.UID), wf.GetOffloadNodeStatusVersion())
+ if err != nil {
+ return err
+ }
+ wf.Status.Nodes = offloadedNodes
+ } else {
+ log.WithFields(log.Fields{"namespace": wf.Namespace, "name": wf.Name}).Warn(sqldb.OffloadNodeStatusDisabled)
+ }
+ }
+ logCtx.Debug("Sending event")
+ err = ws.Send(&workflowpkg.WorkflowWatchEvent{Type: string(event.Type), Object: wf})
+ if err != nil {
+ return err
  }
  }
- logCtx.Debug("Sending event")
- err = ws.Send(&workflowpkg.WorkflowWatchEvent{Type: string(next.Type), Object: wf})
- if err != nil {
- return err
- }
-
  }
-
- log.Debug("Result channel done")
-
- return nil
 }
 
 func (s *workflowServer) DeleteWorkflow(ctx context.Context, req *workflowpkg.WorkflowDeleteRequest) (*workflowpkg.WorkflowDeleteResponse, error) {

diff --git a/util/logs/workflow-logger.go b/util/logs/workflow-logger.go
@@ -166,6 +166,7 @@ func WorkflowLogs(ctx context.Context, wfClient versioned.Interface, kubeClient
  logCtx.Error(err)
  return
  }
+ continue
  }
  wf, ok := event.Object.(*wfv1.Workflow)
  if !ok {
@@ -198,6 +199,7 @@ func WorkflowLogs(ctx context.Context, wfClient versioned.Interface, kubeClient
  logCtx.Error(err)
  return
  }
+ continue
  }
  pod, ok := event.Object.(*corev1.Pod)
  if !ok {