chore: Introduce WorkflowPhase (argoproj#4856)

Signed-off-by: Alex Collins <[email protected]>

cmd/argo/commands/get.go

@@ -142,7 +142,7 @@ func printWorkflowHelper(wf *wfv1.Workflow, getArgs getFlags) string {
  if !wf.Status.StartedAt.IsZero() {
  out += fmt.Sprintf(fmtStr, "Duration:", humanize.RelativeDuration(wf.Status.StartedAt.Time, wf.Status.FinishedAt.Time))
  }
- if wf.Status.Phase == wfv1.NodeRunning {
+ if wf.Status.Phase == wfv1.WorkflowRunning {
  if wf.Status.EstimatedDuration > 0 {
  out += fmt.Sprintf(fmtStr, "EstimatedDuration:", humanize.Duration(wf.Status.EstimatedDuration.ToDuration()))
  }

cmd/argo/commands/wait.go

@@ -103,7 +103,7 @@ func waitOnOne(serviceClient workflowpkg.WorkflowServiceClient, ctx context.Cont
  if !quiet {
  fmt.Printf("%s %s at %v\n", wfName, wf.Status.Phase, wf.Status.FinishedAt)
  }
- if wf.Status.Phase == wfv1.NodeFailed || wf.Status.Phase == wfv1.NodeError {
+ if wf.Status.Phase == wfv1.WorkflowFailed || wf.Status.Phase == wfv1.WorkflowError {
  return false
  }
  return true

persist/sqldb/workflow_archive.go

@@ -21,14 +21,14 @@ const archiveTableName = "argo_archived_workflows"
 const archiveLabelsTableName = archiveTableName + "_labels"
 
 type archivedWorkflowMetadata struct {
- ClusterName string `db:"clustername"`
- InstanceID string `db:"instanceid"`
- UID string `db:"uid"`
- Name string `db:"name"`
- Namespace string `db:"namespace"`
- Phase wfv1.NodePhase `db:"phase"`
- StartedAt time.Time `db:"startedat"`
- FinishedAt time.Time `db:"finishedat"`
+ ClusterName string  `db:"clustername"`
+ InstanceID string  `db:"instanceid"`
+ UID string  `db:"uid"`
+ Name string  `db:"name"`
+ Namespace string  `db:"namespace"`
+ Phase wfv1.WorkflowPhase `db:"phase"`
+ StartedAt time.Time  `db:"startedat"`
+ FinishedAt time.Time  `db:"finishedat"`
 }
 
 type archivedWorkflowRecord struct {

pkg/apis/workflow/v1alpha1/workflow_phase.go

@@ -0,0 +1,22 @@
+package v1alpha1
+
+// the workflow's phase
+type WorkflowPhase string
+
+const (
+ WorkflowUnknown WorkflowPhase = ""
+ WorkflowPending WorkflowPhase = "Pending" // pending some set-up - rarely used
+ WorkflowRunning WorkflowPhase = "Running" // any node has started; pods might not be running yet, the workflow maybe suspended too
+ WorkflowSucceeded WorkflowPhase = "Succeeded"
+ WorkflowFailed WorkflowPhase = "Failed" // it maybe that the the workflow was terminated
+ WorkflowError WorkflowPhase = "Error"
+)
+
+func (p WorkflowPhase) Completed() bool {
+ switch p {
+ case WorkflowSucceeded, WorkflowFailed, WorkflowError:
+ return true
+ default:
+ return false
+ }
+}

pkg/apis/workflow/v1alpha1/workflow_phase_test.go

@@ -0,0 +1,16 @@
+package v1alpha1
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestWorkflowPhase_Completed(t *testing.T) {
+ assert.False(t, WorkflowUnknown.Completed())
+ assert.False(t, WorkflowPending.Completed())
+ assert.False(t, WorkflowRunning.Completed())
+ assert.True(t, WorkflowSucceeded.Completed())
+ assert.True(t, WorkflowFailed.Completed())
+ assert.True(t, WorkflowError.Completed())
+}

pkg/apis/workflow/v1alpha1/workflow_types.go

@@ -11,13 +11,13 @@ import (
  "strings"
  "time"
 
- "github.com/argoproj/argo/util/slice"
-
  apiv1 "k8s.io/api/core/v1"
  policyv1beta "k8s.io/api/policy/v1beta1"
  "k8s.io/apimachinery/pkg/api/resource"
  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  "k8s.io/apimachinery/pkg/util/intstr"
+
+ "github.com/argoproj/argo/util/slice"
 )
 
 // TemplateType is the type of a template
@@ -1291,7 +1291,7 @@ type UserContainer struct {
 // WorkflowStatus contains overall status information about a workflow
 type WorkflowStatus struct {
  // Phase a simple, high-level summary of where the workflow is in its lifecycle.
- Phase NodePhase `json:"phase,omitempty" protobuf:"bytes,1,opt,name=phase,casttype=NodePhase"`
+ Phase WorkflowPhase `json:"phase,omitempty" protobuf:"bytes,1,opt,name=phase,casttype=WorkflowPhase"`
 
  // Time at which this workflow started
  StartedAt metav1.Time `json:"startedAt,omitempty" protobuf:"bytes,2,opt,name=startedAt"`
@@ -1646,19 +1646,19 @@ func (phase NodePhase) FailedOrError() bool {
  return phase == NodeFailed || phase == NodeError
 }
 
-// Fulfilled returns whether or not the workflow has fulfilled its execution, i.e. it completed execution or was skipped
+// Fulfilled returns whether or not the workflow has fulfilled its execution
 func (ws WorkflowStatus) Fulfilled() bool {
- return ws.Phase.Fulfilled()
+ return ws.Phase.Completed()
 }
 
 // Successful return whether or not the workflow has succeeded
 func (ws WorkflowStatus) Successful() bool {
- return ws.Phase == NodeSucceeded
+ return ws.Phase == WorkflowSucceeded
 }
 
 // Failed return whether or not the workflow has failed
 func (ws WorkflowStatus) Failed() bool {
- return ws.Phase == NodeFailed
+ return ws.Phase == WorkflowFailed
 }
 
 func (ws WorkflowStatus) StartTime() *metav1.Time {

server/workflow/workflow_server_test.go

@@ -631,7 +631,7 @@ func (t testWatchWorkflowServer) Send(*workflowpkg.WorkflowWatchEvent) error {
 func TestWatchWorkflows(t *testing.T) {
  server, ctx := getWorkflowServer()
  wf := &v1alpha1.Workflow{
- Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.NodeSucceeded},
+ Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.WorkflowSucceeded},
  }
  assert.NoError(t, json.Unmarshal([]byte(wf1), &wf))
  ctx, cancel := context.WithCancel(ctx)
@@ -645,7 +645,7 @@ func TestWatchWorkflows(t *testing.T) {
 func TestWatchLatestWorkflow(t *testing.T) {
  server, ctx := getWorkflowServer()
  wf := &v1alpha1.Workflow{
- Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.NodeSucceeded},
+ Status: v1alpha1.WorkflowStatus{Phase: v1alpha1.WorkflowSucceeded},
  }
  assert.NoError(t, json.Unmarshal([]byte(wf1), &wf))
  ctx, cancel := context.WithCancel(ctx)
@@ -818,7 +818,7 @@ func TestStopWorkflow(t *testing.T) {
  wf, err = server.StopWorkflow(ctx, &rsmWfReq)
  if assert.NoError(t, err) {
  assert.NotNil(t, wf)
- assert.Equal(t, v1alpha1.NodeRunning, wf.Status.Phase)
+ assert.Equal(t, v1alpha1.WorkflowRunning, wf.Status.Phase)
  }
 }
 

test/e2e/cli_test.go

@@ -424,7 +424,7 @@ func (s *CLISuite) TestRoot() {
  WaitForWorkflow(createdWorkflowName).
  Then().
  ExpectWorkflowName(createdWorkflowName, func(t *testing.T, metadata *corev1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  })
  })
 }
@@ -449,7 +449,7 @@ func (s *CLISuite) TestWorkflowSuspendResume() {
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, _ *corev1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  })
 }
 
@@ -476,11 +476,11 @@ func (s *CLISuite) TestNodeSuspendResume() {
  }
  }).
  WaitForWorkflow(fixtures.Condition(func(wf *wfv1.Workflow) bool {
- return wf.Status.Phase == wfv1.NodeFailed
+ return wf.Status.Phase == wfv1.WorkflowFailed
  }), "suspended node").
  Then().
  ExpectWorkflow(func(t *testing.T, _ *corev1.ObjectMeta, status *wfv1.WorkflowStatus) {
- if assert.Equal(t, wfv1.NodeFailed, status.Phase) {
+ if assert.Equal(t, wfv1.WorkflowFailed, status.Phase) {
  r := regexp.MustCompile(`child '(node-suspend-[0-9]+)' failed`)
  res := r.FindStringSubmatch(status.Message)
  assert.Equal(t, len(res), 2)
@@ -762,7 +762,7 @@ func (s *CLISuite) TestWorkflowRetry() {
  }).
  WaitForWorkflow(fixtures.Condition(func(wf *wfv1.Workflow) bool {
  retryTime = wf.Status.FinishedAt
- return wf.Status.Phase == wfv1.NodeFailed
+ return wf.Status.Phase == wfv1.WorkflowFailed
  }), "is terminated", 20*time.Second).
  Wait(3*time.Second).
  RunCli([]string{"retry", "retry-test", "--restart-successful", "--node-field-selector", "templateName==steps-inner"}, func(t *testing.T, output string, err error) {
@@ -1184,13 +1184,13 @@ func (s *CLISuite) TestWorkflowLevelSemaphore() {
  }).
  SubmitWorkflow().
  WaitForWorkflow(fixtures.Condition(func(wf *wfv1.Workflow) bool {
- return wf.Status.Phase == ""
+ return wf.Status.Phase == wfv1.WorkflowUnknown
  }), "Workflow is waiting for lock").
  WaitForWorkflow().
  DeleteConfigMap("my-config").
  Then().
  ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  })
 }
 
@@ -1206,7 +1206,7 @@ func (s *CLISuite) TestTemplateLevelSemaphore() {
  CreateConfigMap("my-config", semaphoreData).
  SubmitWorkflow().
  WaitForWorkflow(fixtures.Condition(func(wf *wfv1.Workflow) bool {
- return wf.Status.Phase == wfv1.NodeRunning
+ return wf.Status.Phase == wfv1.WorkflowRunning
  }), "waiting for Workflow to run", 10*time.Second).
  RunCli([]string{"get", "semaphore-tmpl-level"}, func(t *testing.T, output string, err error) {
  assert.Contains(t, output, "Waiting for")
@@ -1446,7 +1446,7 @@ spec:
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, _ *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  nodeStatus := status.Nodes.FindByDisplayName("release")
  if assert.NotNil(t, nodeStatus) {
  assert.Equal(t, "Hello, World!", nodeStatus.Inputs.Parameters[0].Value.String())

test/e2e/cluster_workflow_template_test.go

@@ -29,7 +29,7 @@ func (s *ClusterWorkflowTemplateSuite) TestSubmitClusterWorkflowTemplate() {
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, metadata *v1.ObjectMeta, status *v1alpha1.WorkflowStatus) {
- assert.Equal(t, status.Phase, v1alpha1.NodeSucceeded)
+ assert.Equal(t, status.Phase, v1alpha1.WorkflowSucceeded)
  })
 }
 
@@ -68,7 +68,7 @@ spec:
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, metadata *v1.ObjectMeta, status *v1alpha1.WorkflowStatus) {
- assert.Equal(t, v1alpha1.NodeSucceeded, status.Phase)
+ assert.Equal(t, v1alpha1.WorkflowSucceeded, status.Phase)
  })
 
 }

test/e2e/estimated_duration_test.go

@@ -28,7 +28,7 @@ func (s *EstimatedDurationSuite) TestWorkflowTemplate() {
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  assert.NotEmpty(t, status.EstimatedDuration)
  assert.NotEmpty(t, status.Nodes[metadata.Name].EstimatedDuration)
  })
@@ -45,7 +45,7 @@ func (s *EstimatedDurationSuite) TestClusterWorkflowTemplate() {
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  assert.NotEmpty(t, status.EstimatedDuration)
  assert.NotEmpty(t, status.Nodes[metadata.Name].EstimatedDuration)
  })
@@ -62,7 +62,7 @@ func (s *EstimatedDurationSuite) TestCronWorkflow() {
  WaitForWorkflow().
  Then().
  ExpectWorkflow(func(t *testing.T, metadata *metav1.ObjectMeta, status *wfv1.WorkflowStatus) {
- assert.Equal(t, wfv1.NodeSucceeded, status.Phase)
+ assert.Equal(t, wfv1.WorkflowSucceeded, status.Phase)
  assert.NotEmpty(t, status.EstimatedDuration)
  assert.NotEmpty(t, status.Nodes[metadata.Name].EstimatedDuration)
  })