feat: change Errors to RecentErrors

api/v1alpha1/const.go

@@ -7,7 +7,7 @@ import (
 const (
 	// conditions
 	ConditionCompleted    = "Completed"    // the pipeline completed
-	ConditionErrors       = "Errors"       // added if any step encounters an error
+	ConditionRecentErrors = "RecentErrors" // added if any step encountered an error recently
 	ConditionRunning      = "Running"      // added if any step is currently running
 	ConditionSunkMessages = "SunkMessages" // added if any messages have been written to a sink for any step
 	ConditionTerminating  = "Terminating"  // added if any terminator step terminated

api/v1alpha1/source_status.go

@@ -1,5 +1,7 @@
 package v1alpha1
 
+import "time"
+
 type SourceStatus struct {
 	LastMessage *Message           `json:"lastMessage,omitempty" protobuf:"bytes,2,opt,name=lastMessage"`
 	LastError   *Error             `json:"lastError,omitempty" protobuf:"bytes,5,opt,name=lastError"`
@@ -34,6 +36,6 @@ func (in SourceStatus) AnySunk() bool {
 	return in.GetTotal() > 0
 }
 
-func (in SourceStatus) AnyErrors() bool {
-	return in.GetErrors() > 0
+func (in SourceStatus) RecentErrors() bool {
+	return in.LastError != nil && time.Since(in.LastError.Time.Time) < 15*time.Minute
 }

api/v1alpha1/source_status_test.go

@@ -3,6 +3,8 @@ package v1alpha1
 import (
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"github.com/stretchr/testify/assert"
 )
 
@@ -14,8 +16,24 @@ func TestSourceStatus_GetTotal(t *testing.T) {
 }
 
 func TestSourceStatus_GetErrors(t *testing.T) {
-	assert.Equal(t, uint64(0), (SourceStatus{}).GetErrors())
-	assert.Equal(t, uint64(1), (SourceStatus{
-		Metrics: map[string]Metrics{"": {Errors: 1}},
-	}).GetErrors())
+	t.Run("None", func(t *testing.T) {
+		x := SourceStatus{}
+		assert.Equal(t, uint64(0), x.GetErrors())
+	})
+	t.Run("NotRecent", func(t *testing.T) {
+		x := SourceStatus{
+			LastError: &Error{Time: metav1.Time{}},
+			Metrics:   map[string]Metrics{"": {Errors: 1}},
+		}
+		assert.Equal(t, uint64(1), x.GetErrors())
+		assert.False(t, x.RecentErrors())
+	})
+	t.Run("Recent", func(t *testing.T) {
+		x := SourceStatus{
+			LastError: &Error{Time: metav1.Now()},
+			Metrics:   map[string]Metrics{"": {Errors: 1}},
+		}
+		assert.Equal(t, uint64(1), x.GetErrors())
+		assert.True(t, x.RecentErrors())
+	})
 }

api/v1alpha1/source_statuses.go

@@ -59,9 +59,9 @@ func (in SourceStatuses) GetPending() uint64 {
 	return v
 }
 
-func (in SourceStatuses) AnyErrors() bool {
+func (in SourceStatuses) RecentErrors() bool {
 	for _, s := range in {
-		if s.AnyErrors() {
+		if s.RecentErrors() {
 			return true
 		}
 	}

api/v1alpha1/source_statuses_test.go

@@ -5,6 +5,8 @@ import (
 	"strings"
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
 	"k8s.io/apimachinery/pkg/api/resource"
 
 	"github.com/stretchr/testify/assert"
@@ -106,7 +108,7 @@ func TestSourceStatus_GetPending(t *testing.T) {
 	assert.Equal(t, uint64(1), SourceStatuses{"0": {Pending: &v}}.GetPending())
 }
 
-func TestSourceStatus_AnyErrors(t *testing.T) {
-	assert.False(t, SourceStatuses{}.AnyErrors())
-	assert.True(t, SourceStatuses{"0": {Metrics: map[string]Metrics{"0": {Errors: 1}}}}.AnyErrors())
+func TestSourceStatus_RecentErrors(t *testing.T) {
+	assert.False(t, SourceStatuses{}.RecentErrors())
+	assert.True(t, SourceStatuses{"0": {LastError: &Error{Time: metav1.Now()}}}.RecentErrors())
 }

api/v1alpha1/step_status.go

@@ -19,6 +19,6 @@ func (m StepStatus) GetReplicas() int {
 	return int(m.Replicas)
 }
 
-func (in StepStatus) AnyErrors() bool {
-	return in.SinkStatues.AnyErrors() || in.SourceStatuses.AnyErrors()
+func (in StepStatus) RecentErrors() bool {
+	return in.SinkStatues.RecentErrors() || in.SourceStatuses.RecentErrors()
 }

api/v1alpha1/step_status_test.go

@@ -6,6 +6,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestStepStatus_AnyErrors(t *testing.T) {
-	assert.False(t, StepStatus{}.AnyErrors())
+func TestStepStatus_RecentErrors(t *testing.T) {
+	assert.False(t, StepStatus{}.RecentErrors())
 }

examples/301-erroring-pipeline.py

@@ -12,7 +12,7 @@ def handler(msg):
     (pipeline("301-erroring")
      .owner('argoproj-labs')
      .describe("""This example showcases retry policies.""")
-     .annotate('dataflow.argoproj.io/wait-for', 'Errors')
+     .annotate('dataflow.argoproj.io/wait-for', 'RecentErrors')
      .step(
         (cron('*/3 * * * * *', retryPolicy='Always')
          .handler('always', handler=handler)

examples/301-erroring-pipeline.yaml

@@ -4,7 +4,7 @@ metadata:
   annotations:
     dataflow.argoproj.io/description: This example showcases retry policies.
     dataflow.argoproj.io/owner: argoproj-labs
-    dataflow.argoproj.io/wait-for: Errors
+    dataflow.argoproj.io/wait-for: RecentErrors
   name: 301-erroring
 spec:
   steps:

manager/controllers/pipeline_controller.go

@@ -111,7 +111,7 @@ func (r *PipelineReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	pending, running, succeeded, failed := 0, 0, 0, 0
 	newStatus := *pipeline.Status.DeepCopy()
 	newStatus.Phase = dfv1.PipelineUnknown
-	terminate, sunkMessages, errors := false, false, false
+	terminate, sunkMessages, recentErrors := false, false, false
 	for _, step := range steps.Items {
 		stepName := step.Spec.Name
 		if !pipeline.Spec.HasStep(stepName) { // this happens when a pipeline changes and a step is removed
@@ -139,7 +139,7 @@ func (r *PipelineReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		}
 		terminate = terminate || step.Status.Phase.Completed() && step.Spec.Terminator
 		sunkMessages = sunkMessages || step.Status.SinkStatues.AnySunk()
-		errors = errors || step.Status.AnyErrors()
+		recentErrors = recentErrors || step.Status.RecentErrors()
 	}
 
 	if newStatus.Phase.Completed() {
@@ -167,7 +167,7 @@ func (r *PipelineReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		dfv1.ConditionRunning:      newStatus.Phase == dfv1.PipelineRunning,
 		dfv1.ConditionCompleted:    newStatus.Phase.Completed(),
 		dfv1.ConditionSunkMessages: sunkMessages,
-		dfv1.ConditionErrors:       errors,
+		dfv1.ConditionRecentErrors: recentErrors,
 		dfv1.ConditionTerminating:  terminate,
 	} {
 		if ok {