feat: adds kubernetes and kubernetes_status outputs

README.md

@@ -12,15 +12,18 @@ a kubernetes plugin [benthos](https://github.com/Jeffail/benthos) which includes
 - `kubernetes_status` writes object status to kubernetes
 
 ## Installing
+
 - with Docker
   ```shell
-  $ docker run cludden/benthos-kubernetes-input -h
+  $ docker run cludden/benthos-kubernetes -h
   ```
-- download a [release](https://github.com/cludden/benthos-kubernetes-input/releases)
-- as benthos [plugin](./cmd/benthos/main.go)
+- download a [release](https://github.com/cludden/benthos-kubernetes/releases)
+- as a benthos [plugin](./cmd/benthos/main.go)
 
 ## Getting Started
+
 Sample benthos stream config:
+
 ```yaml
 input:
   type: kubernetes
@@ -50,10 +53,13 @@ input:
 output:
   stdout: {}
 ```
+
 Or see [examples](./example)
 
 ## Metadata
+
 This input adds the following metadata fields to each message:
+
 ```
 - deleted (present only if object has been deleted)
 - group
@@ -64,5 +70,6 @@ This input adds the following metadata fields to each message:
 ```
 
 ## License
+
 Licensed under the [MIT License](LICENSE.md)  
 Copyright (c) 2020 Chris Ludden

cmd/benthos/main.go

@@ -3,7 +3,8 @@ package main
 import (
 	"github.com/Jeffail/benthos/v3/lib/service"
 
-	_ "github.com/cludden/benthos-kubernetes-input/input"
+	_ "github.com/cludden/benthos-kubernetes/input"
+	_ "github.com/cludden/benthos-kubernetes/output"
 )
 
 //------------------------------------------------------------------------------

example/crds.yml

@@ -24,6 +24,9 @@ spec:
               properties:
                 size:
                   type: number
+            status:
+              type: object
+              x-kubernetes-preserve-unknown-fields: true
       subresources:
         status: {}
       additionalPrinterColumns:

example/finalizer.yml

@@ -0,0 +1,51 @@
+input:
+  type: kubernetes
+  plugin:
+    watches:
+      - group: example.com
+        version: v1
+        kind: Bar
+
+pipeline:
+  processors:
+    - conditional:
+        condition:
+          bloblang: meta().exists("deleted")
+        processors:
+          - log:
+              message: deletion detected, substituting last cached object image...
+          - cache:
+              cache: objects
+              operator: get
+              key: ${!meta("kind")}/${INVISION_CONTEXT}/${!meta("namespace")}/${!meta("name")}
+        else_processors:
+          - cache:
+              cache: objects
+              operator: set
+              key: ${!meta("kind")}/${INVISION_CONTEXT}/${!meta("namespace")}/${!meta("name")}
+              value: ${!content()}
+
+    - bloblang: |
+        map finalizer {
+          root = this
+          metadata.finalizers = metadata.finalizers.or([]).append("finalizer.bars.example.com")
+        }
+        root = match {
+          meta().exists("deleted") || metadata.finalizers.or([]).contains("finalizer.bars.example.com") => deleted()
+          _ => this.apply("finalizer")
+        }
+
+output:
+  broker:
+    outputs:
+      - type: kubernetes
+        plugin: {}
+      - stdout: {}
+
+logger:
+  level: info
+
+resources:
+  caches:
+    objects:
+      memory: {}

example/manifest.yml

@@ -7,43 +7,16 @@ metadata:
     color: green
   finalizers:
     - finalizer.foos.example.com
+    - random
 spec:
   size: 1
 
----
-apiVersion: example.com/v1
-kind: Foo
-metadata:
-  name: two
-  namespace: kube-system
-  labels:
-    color: yellow
-  finalizers:
-    - finalizer.foos.example.com
-spec:
-  size: 2
-
 ---
 apiVersion: example.com/v1
 kind: Bar
 metadata:
   name: three
   labels:
     color: blue
-  finalizers:
-    - finalizer.bars.example.com
 spec:
   size: 3
-
----
-apiVersion: example.com/v1
-kind: Bar
-metadata:
-  name: four
-  namespace: kube-system
-  labels:
-    color: purple
-  finalizers:
-    - finalizer.bars.example.com
-spec:
-  size: 4

example/status.yml

@@ -0,0 +1,76 @@
+input:
+  type: kubernetes
+  plugin:
+    watches:
+      - group: example.com
+        version: v1
+        kind: Foo
+      # - group: example.com
+      #   version: v1
+      #   kind: Bar
+
+pipeline:
+  processors:
+    - conditional:
+        condition:
+          bloblang: meta().exists("deleted")
+        processors:
+          - log:
+              message: deletion detected, substituting last cached object image...
+          - cache:
+              cache: objects
+              operator: get
+              key: ${!meta("kind")}/${INVISION_CONTEXT}/${!meta("namespace")}/${!meta("name")}
+        else_processors:
+          - cache:
+              cache: objects
+              operator: set
+              key: ${!meta("kind")}/${INVISION_CONTEXT}/${!meta("namespace")}/${!meta("name")}
+              value: ${!content()}
+
+    - bloblang: |
+        let metadataKeys = ["deletionTimestamp","finalizers"]
+        let state = {
+          "metadata": metadata.filter($metadataKeys.contains(this.key)),
+          "spec": spec
+        }
+        let hash = $state.string().hash("sha1").encode("hex")
+        let finalizer = "finalizer.%ss.example.com".format(kind.lowercase())
+        let finalizers = metadata.finalizers.or([])
+
+        map reconciling {
+          root = obj
+          status.hash = hash
+          status.lastReconciledAt = timestamp_utc("2006-01-02T15:04:05.999999999Z")
+          status.status = "Reconciling"
+        }
+
+        meta hash = $hash
+        root = match {
+          meta().exists("deleted") => this
+          metadata.exists("deletionTimestamp") && $finalizers.join(",") != $finalizer => deleted()
+          status.hash.or("") == $hash => deleted()
+          _ => {"obj":this,"hash":$hash}.apply("reconciling")
+        }
+
+output:
+  broker:
+    outputs:
+      - type: kubernetes_status
+        plugin: {}
+        processors:
+          - bloblang: |
+              root = match {
+                meta().exists("deleted") => deleted()
+              }
+          - log:
+              message: updating status
+      - stdout: {}
+
+logger:
+  level: info
+
+resources:
+  caches:
+    objects:
+      memory: {}