feat: add action to delete recyclable Numaflow pipelines

resource_customizations/numaflow.numaproj.io/Pipeline/actions/action_test.yaml

@@ -11,6 +11,9 @@ actionTests:
 - action: force-promote
   inputPath: testdata/pipeline.yaml
   expectedOutputPath: testdata/pipeline-force-promote.yaml
+- action: delete
+  inputPath: testdata/pipeline-recyclable.yaml
+  expectedOutputPath: testdata/pipeline-delete.yaml
 
 discoveryTests:
   - inputPath: testdata/pipeline.yaml
@@ -30,6 +33,10 @@ discoveryTests:
         disabled: false
         iconClass: 'fa-solid fa-fw fa-forward'
         displayName: 'Force Promote'
+      - name: delete
+        disabled: true
+        iconClass: 'fa-solid fa-fw fa-trash'
+        displayName: 'Delete'
   - inputPath: testdata/pipeline-paused.yaml
     result:
       - name: pause
@@ -47,3 +54,28 @@ discoveryTests:
         disabled: false
         iconClass: 'fa-solid fa-fw fa-forward'
         displayName: 'Force Promote'
+      - name: delete
+        disabled: true
+        iconClass: 'fa-solid fa-fw fa-trash'
+        displayName: 'Delete'
+  - inputPath: testdata/pipeline-recyclable.yaml
+    result:
+      - name: pause
+        disabled: false
+        iconClass: 'fa-solid fa-fw fa-pause'
+      - name: unpause-gradual
+        disabled: true
+        iconClass: 'fa-solid fa-fw fa-play'
+        displayName: 'Unpause (gradual)'
+      - name: unpause-fast
+        disabled: true
+        iconClass: 'fa-solid fa-fw fa-play'
+        displayName: 'Unpause (fast)'
+      - name: force-promote
+        disabled: true
+        iconClass: 'fa-solid fa-fw fa-forward'
+        displayName: 'Force Promote'
+      - name: delete
+        disabled: false
+        iconClass: 'fa-solid fa-fw fa-trash'
+        displayName: 'Delete'

resource_customizations/numaflow.numaproj.io/Pipeline/actions/delete/action.lua

@@ -0,0 +1,5 @@
+if obj.metadata.annotations == nil then
+    obj.metadata.annotations = {}
+end
+obj.metadata.annotations["numaplane.numaproj.io/marked-for-deletion"] = "true"
+return obj

resource_customizations/numaflow.numaproj.io/Pipeline/actions/discovery.lua

@@ -18,6 +18,11 @@ actions["force-promote"] = {
   ["displayName"] = "Force Promote",
   ["iconClass"] = "fa-solid fa-fw fa-forward"
 }
+actions["delete"] = {
+  ["disabled"] = true,
+  ["displayName"] = "Delete",
+  ["iconClass"] = "fa-solid fa-fw fa-trash"
+}
 
 -- pause/unpause
 local paused = false
@@ -57,4 +62,15 @@ else
   actions["force-promote"]["disabled"] = true
 end
 
+-- delete
+local recyclable = false
+if obj.metadata.labels ~= nil and (obj.metadata.labels["numaplane.numaproj.io/upgrade-state"] == "recyclable") then
+  recyclable = true
+end
+if recyclable then
+  actions["delete"]["disabled"] = false
+else
+  actions["delete"]["disabled"] = true
+end
+
 return actions

resource_customizations/numaflow.numaproj.io/Pipeline/actions/testdata/pipeline-delete.yaml

@@ -0,0 +1,103 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  creationTimestamp: "2024-10-08T18:22:18Z"
+  finalizers:
+  - pipeline-controller
+  generation: 1
+  name: simple-pipeline
+  namespace: numaflow-system
+  resourceVersion: "382381"
+  uid: bb6cc91c-eb05-4fe7-9380-63b9532a85db
+  labels:
+    numaplane.numaproj.io/upgrade-state: "recyclable"
+  annotations:
+    numaflow.numaproj.io/allowed-resume-strategies: "slow, fast"
+    numaplane.numaproj.io/marked-for-deletion: "true"
+spec:
+  edges:
+  - from: in
+    to: cat
+  - from: cat
+    to: out
+  lifecycle:
+    deleteGracePeriodSeconds: 30
+    desiredPhase: Running
+    pauseGracePeriodSeconds: 30
+  limits:
+    bufferMaxLength: 30000
+    bufferUsageLimit: 80
+    readBatchSize: 500
+    readTimeout: 1s
+  vertices:
+  - name: in
+    scale:
+      min: 1
+    source:
+      generator:
+        duration: 1s
+        jitter: 0s
+        msgSize: 8
+        rpu: 5
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  - name: cat
+    scale:
+      min: 1
+    udf:
+      builtin:
+        name: cat
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  - name: out
+    scale:
+      min: 1
+    sink:
+      log: {}
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  watermark:
+    disabled: false
+    maxDelay: 0s
+status:
+  conditions:
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: Configured
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: DaemonServiceHealthy
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: Deployed
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: No Side Inputs attached to the pipeline
+    reason: NoSideInputs
+    status: "True"
+    type: SideInputsManagersHealthy
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: All vertices are healthy
+    reason: Successful
+    status: "True"
+    type: VerticesHealthy
+  lastUpdated: "2024-10-09T20:26:54Z"
+  mapUDFCount: 1
+  observedGeneration: 1
+  phase: Running
+  reduceUDFCount: 0
+  sinkCount: 1
+  sourceCount: 1
+  udfCount: 1
+  vertexCount: 3

resource_customizations/numaflow.numaproj.io/Pipeline/actions/testdata/pipeline-recyclable.yaml

@@ -0,0 +1,102 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  creationTimestamp: "2024-10-08T18:22:18Z"
+  finalizers:
+  - pipeline-controller
+  generation: 1
+  name: simple-pipeline
+  namespace: numaflow-system
+  resourceVersion: "382381"
+  uid: bb6cc91c-eb05-4fe7-9380-63b9532a85db
+  labels:
+    numaplane.numaproj.io/upgrade-state: "recyclable"
+  annotations:
+    numaflow.numaproj.io/allowed-resume-strategies: "slow, fast"
+spec:
+  edges:
+  - from: in
+    to: cat
+  - from: cat
+    to: out
+  lifecycle:
+    deleteGracePeriodSeconds: 30
+    desiredPhase: Running
+    pauseGracePeriodSeconds: 30
+  limits:
+    bufferMaxLength: 30000
+    bufferUsageLimit: 80
+    readBatchSize: 500
+    readTimeout: 1s
+  vertices:
+  - name: in
+    scale:
+      min: 1
+    source:
+      generator:
+        duration: 1s
+        jitter: 0s
+        msgSize: 8
+        rpu: 5
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  - name: cat
+    scale:
+      min: 1
+    udf:
+      builtin:
+        name: cat
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  - name: out
+    scale:
+      min: 1
+    sink:
+      log: {}
+    updateStrategy:
+      rollingUpdate:
+        maxUnavailable: 25%
+      type: RollingUpdate
+  watermark:
+    disabled: false
+    maxDelay: 0s
+status:
+  conditions:
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: Configured
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: DaemonServiceHealthy
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: Successful
+    reason: Successful
+    status: "True"
+    type: Deployed
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: No Side Inputs attached to the pipeline
+    reason: NoSideInputs
+    status: "True"
+    type: SideInputsManagersHealthy
+  - lastTransitionTime: "2024-10-09T20:26:54Z"
+    message: All vertices are healthy
+    reason: Successful
+    status: "True"
+    type: VerticesHealthy
+  lastUpdated: "2024-10-09T20:26:54Z"
+  mapUDFCount: 1
+  observedGeneration: 1
+  phase: Running
+  reduceUDFCount: 0
+  sinkCount: 1
+  sourceCount: 1
+  udfCount: 1
+  vertexCount: 3