feat: add errors to step/status

Makefile

@@ -86,6 +86,7 @@ proto: api/v1alpha1/generated.pb.go
 $(GOBIN)/go-to-protobuf:
 	go install k8s.io/code-generator/cmd/[email protected]
 
+api/v1alpha1/generated.pb.go:
 api/v1alpha1/generated.%: $(shell find api/v1alpha1 -type f -name '*.go' -not -name '*generated*' -not -name groupversion_info.go) $(GOBIN)/go-to-protobuf
 	[ ! -e api/v1alpha1/groupversion_info.go ] || mv api/v1alpha1/groupversion_info.go api/v1alpha1/groupversion_info.go.0
 	go-to-protobuf \

Procfile

@@ -1,6 +1,6 @@
 runner: make runner
 runtimes: make runtimes
-controller: ARGO_DATAFLOW_PULL_POLICY=IfNotPresent ARGO_DATAFLOW_INSTALLER=true ARGO_DATAFLOW_UPDATE_INTERVAL=5s ARGO_DATAFLOW_NAMESPACE=argo-dataflow-system go run ./main.go -metrics-addr :7070
+controller: ARGO_DATAFLOW_PULL_POLICY=IfNotPresent ARGO_DATAFLOW_INSTALLER=true ARGO_DATAFLOW_UPDATE_INTERVAL=5s ARGO_DATAFLOW_NAMESPACE=argo-dataflow-system go run ./manager -metrics-addr :7070
 logs: make logs
 argocli: make argocli
 ui: make ui

api/util/containerkiller/container_killer.go

@@ -60,7 +60,7 @@ func (k *containerKiller) killContainer(namespace, pod, container string, comman
 		Stdout: os.Stdout,
 		Stderr: os.Stderr,
 		Tty:    true,
-	}); dfv1.IgnoreNotFound(dfv1.IgnoreContainerNotFound(err)) != nil {
+	}); util.IgnoreNotFound(util.IgnoreContainerNotFound(err)) != nil {
 		return fmt.Errorf("failed to stream: %w", err)
 	}
 	return nil

api/util/env.go

@@ -0,0 +1,18 @@
+package util
+
+import (
+	"fmt"
+	"os"
+	"time"
+)
+
+func GetEnvDuration(key string, def time.Duration) time.Duration {
+	if v, ok := os.LookupEnv(key); ok {
+		if v, err := time.ParseDuration(v); err != nil {
+			panic(fmt.Errorf("%s=%s; value must be duration: %w", key, v, err))
+		} else {
+			return v
+		}
+	}
+	return def
+}

api/util/env_test.go

@@ -0,0 +1,21 @@
+package util
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_GetEnvDuration(t *testing.T) {
+	defer os.Unsetenv("FOO")
+	assert.Equal(t, time.Minute, GetEnvDuration("FOO", time.Minute))
+	_ = os.Setenv("FOO", "2m")
+	assert.Equal(t, 2*time.Minute, GetEnvDuration("FOO", 0))
+	_ = os.Setenv("FOO", "xx")
+
+	assert.Panics(t, func() {
+		_ = GetEnvDuration("FOO", 0)
+	})
+}

api/util/error.go

@@ -0,0 +1,51 @@
+package util
+
+import (
+	"errors"
+	"os"
+	"strings"
+
+	apierr "k8s.io/apimachinery/pkg/api/errors"
+)
+
+func IgnorePermission(err error) error {
+	if errors.Is(err, os.ErrPermission) {
+		return nil
+	}
+	return err
+}
+
+func IgnoreExist(err error) error {
+	if errors.Is(err, os.ErrExist) {
+		return nil
+	}
+	return err
+}
+
+func IgnoreAlreadyExists(err error) error {
+	if apierr.IsAlreadyExists(err) {
+		return nil
+	}
+	return err
+}
+
+func IgnoreNotFound(err error) error {
+	if apierr.IsNotFound(err) {
+		return nil
+	}
+	return err
+}
+
+func IgnoreContainerNotFound(err error) error {
+	if err != nil && strings.Contains(err.Error(), "container not found") {
+		return nil
+	}
+	return err
+}
+
+func IgnoreConflict(err error) error {
+	if apierr.IsConflict(err) {
+		return nil
+	}
+	return err
+}

api/v1alpha1/process.go → api/util/process.go

@@ -1,4 +1,4 @@
-package v1alpha1
+package util
 
 import "os"
 

api/v1alpha1/const.go

@@ -31,7 +31,9 @@ const (
 	EnvPipelineName    = "ARGO_DATAFLOW_PIPELINE_NAME"
 	EnvReplica         = "ARGO_DATAFLOW_REPLICA"
 	EnvStepSpec        = "ARGO_DATAFLOW_STEP_SPEC"
+	EnvPeekDelay       = "ARGO_DATAFLOW_PEEK_DELAY"      // how long between peeking (default 4m)
 	EnvPullPolicy      = "ARGO_DATAFLOW_PULL_POLICY"     // default ""
+	EnvScalingDelay    = "ARGO_DATAFLOW_SCALING_DELAY"   // // how long to wait between any scaling events (including peeking) default "4m"
 	EnvUpdateInterval  = "ARGO_DATAFLOW_UPDATE_INTERVAL" // default "15s"
 	// label/annotation keys
 	KeyDefaultContainer = "kubectl.kubernetes.io/default-container"

api/v1alpha1/container_supplier.go

@@ -2,13 +2,15 @@ package v1alpha1
 
 import corev1 "k8s.io/api/core/v1"
 
+// +kubebuilder:skipversion
 type getContainerReq struct {
 	imageFormat     string
 	runnerImage     string
 	imagePullPolicy corev1.PullPolicy
 	volumeMount     corev1.VolumeMount
 }
 
+// +kubebuilder:skipversion
 type containerSupplier interface {
 	getContainer(req getContainerReq) corev1.Container
 }

api/v1alpha1/error.go

@@ -1,51 +1,9 @@
 package v1alpha1
 
-import (
-	"errors"
-	"os"
-	"strings"
+import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
-	apierr "k8s.io/apimachinery/pkg/api/errors"
-)
-
-func IgnorePermission(err error) error {
-	if errors.Is(err, os.ErrPermission) {
-		return nil
-	}
-	return err
-}
-
-func IgnoreExist(err error) error {
-	if errors.Is(err, os.ErrExist) {
-		return nil
-	}
-	return err
-}
-
-func IgnoreAlreadyExists(err error) error {
-	if apierr.IsAlreadyExists(err) {
-		return nil
-	}
-	return err
-}
-
-func IgnoreNotFound(err error) error {
-	if apierr.IsNotFound(err) {
-		return nil
-	}
-	return err
-}
-
-func IgnoreContainerNotFound(err error) error {
-	if err != nil && strings.Contains(err.Error(), "container not found") {
-		return nil
-	}
-	return err
-}
-
-func IgnoreConflict(err error) error {
-	if apierr.IsConflict(err) {
-		return nil
-	}
-	return err
+type Error struct {
+	// +kubebuilder:validation:MaxLength=32
+	Message string      `json:"message" protobuf:"bytes,1,opt,name=message"`
+	Time    metav1.Time `json:"time" protobuf:"bytes,2,opt,name=time"`
 }