feat: automatic GC of pipelines 30m after completion

argoproj-labs · Jul 2, 2021 · 810aef7 · 810aef7
1 parent be5e5ab
commit 810aef7
Show file tree

Hide file tree

Showing 14 changed files with 328 additions and 209 deletions.
diff --git a/api/v1alpha1/const.go b/api/v1alpha1/const.go
@@ -16,7 +16,8 @@ const (
 	CtrMain    = "main"
 	CtrSidecar = "sidecar"
 	// env vars
-	EnvImageFormat                 = "ARGO_DATAFLOW_IMAGE_FORMAT" // default "quay.io/argoproj/%s:latest"
+	EnvImageFormat                 = "ARGO_DATAFLOW_IMAGE_FORMAT"   // default "quay.io/argoproj/%s:latest"
+	EnvDeletionDelay               = "ARGO_DATAFLOW_DELETION_DELAY" // default "30m"
 	EnvNamespace                   = "ARGO_DATAFLOW_NAMESPACE"
 	EnvPipelineName                = "ARGO_DATAFLOW_PIPELINE_NAME"
 	EnvReplica                     = "ARGO_DATAFLOW_REPLICA"

diff --git a/api/v1alpha1/generated.pb.go b/api/v1alpha1/generated.pb.go
diff --git a/api/v1alpha1/generated.proto b/api/v1alpha1/generated.proto
diff --git a/api/v1alpha1/pipeline_status.go b/api/v1alpha1/pipeline_status.go
@@ -6,5 +6,5 @@ type PipelineStatus struct {
 	Phase       PipelinePhase      `json:"phase,omitempty" protobuf:"bytes,1,opt,name=phase,casttype=PipelinePhase"`
 	Message     string             `json:"message,omitempty" protobuf:"bytes,2,opt,name=message"`
 	Conditions  []metav1.Condition `json:"conditions,omitempty" protobuf:"bytes,3,rep,name=conditions"`
-	LastUpdated metav1.Time        `json:"lastUpdated"`
+	LastUpdated metav1.Time        `json:"lastUpdated" protobuf:"bytes,4,opt,name=lastUpdated"`
 }
diff --git a/config/ci.yaml b/config/ci.yaml
@@ -1881,6 +1881,9 @@ spec:
                   - type
                   type: object
                 type: array
+              lastUpdated:
+                format: date-time
+                type: string
               message:
                 type: string
               phase:
@@ -1891,6 +1894,8 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+            required:
+            - lastUpdated
             type: object
         required:
         - spec

diff --git a/config/crd/bases/dataflow.argoproj.io_pipelines.yaml b/config/crd/bases/dataflow.argoproj.io_pipelines.yaml
@@ -2929,6 +2929,9 @@ spec:
                   - type
                   type: object
                 type: array
+              lastUpdated:
+                format: date-time
+                type: string
               message:
                 type: string
               phase:
@@ -2939,6 +2942,8 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+            required:
+            - lastUpdated
             type: object
         required:
         - spec

diff --git a/config/default.yaml b/config/default.yaml
@@ -1881,6 +1881,9 @@ spec:
                   - type
                   type: object
                 type: array
+              lastUpdated:
+                format: date-time
+                type: string
               message:
                 type: string
               phase:
@@ -1891,6 +1894,8 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+            required:
+            - lastUpdated
             type: object
         required:
         - spec

diff --git a/config/dev.yaml b/config/dev.yaml
@@ -1881,6 +1881,9 @@ spec:
                   - type
                   type: object
                 type: array
+              lastUpdated:
+                format: date-time
+                type: string
               message:
                 type: string
               phase:
@@ -1891,6 +1894,8 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+            required:
+            - lastUpdated
             type: object
         required:
         - spec
@@ -4489,7 +4494,6 @@ metadata:
   name: kafka-broker
   namespace: argo-dataflow-system
 spec:
-  replicas: 2
   selector:
     matchLabels:
       app: kafka-broker
@@ -4504,13 +4508,11 @@ spec:
         - name: KAFKA_ADVERTISED_PORT
           value: "9092"
         - name: KAFKA_ADVERTISED_HOST_NAME
-          valueFrom:
-            fieldRef:
-              fieldPath: status.podIP
+          value: kafka-broker
         - name: KAFKA_ZOOKEEPER_CONNECT
           value: zookeeper:2181
-        - name: KAFKA_CREATE_TOPICS
-          value: __consumer_offsets:1:1
+        - name: KAFKA_BROKER_ID
+          value: "0"
         image: wurstmeister/kafka
         imagePullPolicy: IfNotPresent
         name: main

diff --git a/config/kafka-dev.yaml b/config/kafka-dev.yaml
@@ -31,7 +31,6 @@ metadata:
   name: kafka-broker
   namespace: argo-dataflow-system
 spec:
-  replicas: 2
   selector:
     matchLabels:
       app: kafka-broker
@@ -46,13 +45,11 @@ spec:
         - name: KAFKA_ADVERTISED_PORT
           value: "9092"
         - name: KAFKA_ADVERTISED_HOST_NAME
-          valueFrom:
-            fieldRef:
-              fieldPath: status.podIP
+          value: kafka-broker
         - name: KAFKA_ZOOKEEPER_CONNECT
           value: zookeeper:2181
-        - name: KAFKA_CREATE_TOPICS
-          value: __consumer_offsets:1:1
+        - name: KAFKA_BROKER_ID
+          value: "0"
         image: wurstmeister/kafka
         imagePullPolicy: IfNotPresent
         name: main

diff --git a/config/quick-start.yaml b/config/quick-start.yaml
@@ -1881,6 +1881,9 @@ spec:
                   - type
                   type: object
                 type: array
+              lastUpdated:
+                format: date-time
+                type: string
               message:
                 type: string
               phase:
@@ -1891,6 +1894,8 @@ spec:
                 - Succeeded
                 - Failed
                 type: string
+            required:
+            - lastUpdated
             type: object
         required:
         - spec

diff --git a/manager/controllers/config.go b/manager/controllers/config.go
@@ -29,7 +29,8 @@ var (
 			"memory": resource.MustParse("64Mi"),
 		},
 	}
-	logger = util.NewLogger()
+	deletionDelay = util.GetEnvDuration(dfv1.EnvDeletionDelay, 30*time.Minute)
+	logger        = util.NewLogger()
 )
 
 func init() {
@@ -52,8 +53,9 @@ func init() {
 		"runnerImage", runnerImage,
 		"pullPolicy", pullPolicy,
 		"updateInterval", updateInterval.String(),
-		"scalingDelay", scalingDelay,
-		"peekDelay", peekDelay,
+		"scalingDelay", scalingDelay.String(),
+		"peekDelay", peekDelay.String(),
 		"defaultResourceRequirements", defaultResourceRequirements,
+		"deletionDelay", deletionDelay.String(),
 	)
 }
diff --git a/runner/main.go b/runner/main.go
@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/argoproj-labs/argo-dataflow/runner/dedupe"
 	"io/ioutil"
@@ -60,7 +61,14 @@ func main() {
 			return fmt.Errorf("unknown comand")
 		}
 	}()
-	if err != nil && err != context.Canceled {
+	is := errors.Is(err, context.Canceled)
+	logger.Info("is", "is", is)
+
+
+
+	if is {
+		println(fmt.Errorf("ignoring context cancelled error, expected"))
+	} else if err != nil {
 		if err := ioutil.WriteFile("/dev/termination-log", []byte(err.Error()), 0o600); err != nil {
 			println(fmt.Sprintf("failed to write termination-log: %v", err))
 		}

diff --git a/test/e2e/completion_test.go b/test/e2e/completion_test.go
@@ -0,0 +1,35 @@
+// +build test
+
+package e2e
+
+import (
+	"testing"
+
+	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
+	. "github.com/argoproj-labs/argo-dataflow/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestCompletion(t *testing.T) {
+
+	Setup(t)
+	defer Teardown(t)
+
+	CreatePipeline(Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "completion"},
+		Spec: PipelineSpec{
+			Steps: []StepSpec{{
+				Name: "main",
+				Container: &Container{
+					Image:   "alpine",
+					Command: []string{"sh"},
+					Args:    []string{"-c", "exit 0"},
+				},
+			}},
+		},
+	})
+
+	WaitForPipeline(UntilSucceeded)
+	DeletePipelines()
+	WaitForPodsToBeDeleted()
+}
diff --git a/test/pipeline.go b/test/pipeline.go
@@ -27,6 +27,10 @@ func UntilMessagesSunk(pl Pipeline) bool {
 	return meta.FindStatusCondition(pl.Status.Conditions, ConditionSunkMessages) != nil
 }
 
+func UntilSucceeded(pl Pipeline) bool {
+	return pl.Status.Phase == PipelineSucceeded
+}
+
 func DeletePipelines() {
 	log.Printf("deleting pipelines\n")
 	ctx := context.Background()