feat: adds support for multiple watches, namespace filters, updated metadata fields, controller log forwarding

Author: cludden

README.md

@@ -1,5 +1,5 @@
 # benthos-kubernetes-input
-a [benthos](https://github.com/Jeffail/benthos) kubernetes input plugin
+a [benthos](https://github.com/Jeffail/benthos) kubernetes input plugin that streams resource messages based on one or more configured watches.
 
 ## Installing
 - with Docker
@@ -15,15 +15,44 @@ Sample benthos stream config:
 input:
   type: kubernetes
   plugin:
-    group: ""
-    version: v1
-    kind: Pod
+    watches:
+      # watch pods in all namespaces
+      - group: ""
+        version: v1
+        kind: Pod
+      # watch custom resources in specified namespaces
+      - group: example.com
+        version: v1alpha1
+        kind: Foo
+        namespaces: [default, kube-system]
+      # watch custom resources that match a label selector
+      - group: example.com
+        version: v1apha1
+        kind: Bar
+        selector:
+          matchLabels:
+            color: blue
+          matchExpressions:
+            - key: color
+              operator: NotIn
+              values: [green, yellow]
 
 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
+- kind
+- name
+- namespace
+- version
+```
+
 ## License
 Licensed under the [MIT License](LICENSE.md)  
 Copyright (c) 2020 Chris Ludden

example/benthos.yml

@@ -1,19 +1,49 @@
 input:
   type: kubernetes
   plugin:
-    group: example.com
-    version: v1
-    kind: Foo
+    watches:
+      - group: example.com
+        version: v1
+        kind: Foo
+      - group: example.com
+        version: v1
+        kind: Bar
 
 pipeline:
   processors:
     - bloblang: |
         object = this
+        metadata = meta()
         action = match {
-          metadata.exists("deletionTimestamp") && metadata.get("finalizers").or([]).join(",") == "finalizer.foos.example.com" => "deprovision"
+          meta().exists("deleted") => "deleted"
+          metadata.exists("deletionTimestamp") && metadata.get("finalizers").or([]).join(",") == "finalizer.%ss.example.com".format(meta("kind").lowercase()) => "deprovision"
           metadata.exists("deletionTimestamp") => "deleting"
           _ => "provision"
         }
 
+    - conditional:
+        condition:
+          bloblang: action != "deleted"
+        processors:
+          - cache:
+              cache: objects
+              operator: set
+              key: ${!meta("kind")}/${!meta("namespace")}/${!meta("name")}
+              value: ${!json("object").string()}
+        else_processors:
+          - branch:
+              request_map: 'root = ""'
+              processors:
+                - cache:
+                    cache: objects
+                    operator: get
+                    key: ${!meta("kind")}/${!meta("namespace")}/${!meta("name")}
+              result_map: root.object = this
+
 output:
   stdout: {}
+
+resources:
+  caches:
+    objects:
+      memory: {}

example/cr.yml

@@ -0,0 +1,77 @@
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: foos.example.com
+spec:
+  group: example.com
+  names:
+    kind: Foo
+    listKind: FooList
+    plural: foos
+    singular: foo
+  scope: Namespaced
+  versions:
+    - name: v1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                size:
+                  type: number
+      subresources:
+        status: {}
+      additionalPrinterColumns:
+        - name: Status
+          type: string
+          description: resource status
+          jsonPath: .status.status
+        - name: Age
+          type: date
+          description: resource creation timestamp
+          format: date-time
+          jsonPath: .metadata.creationTimestamp
+
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: bars.example.com
+spec:
+  group: example.com
+  names:
+    kind: Bar
+    listKind: BarList
+    plural: bars
+    singular: bar
+  scope: Namespaced
+  versions:
+    - name: v1
+      served: true
+      storage: true
+      schema:
+        openAPIV3Schema:
+          type: object
+          properties:
+            spec:
+              type: object
+              properties:
+                size:
+                  type: number
+      subresources:
+        status: {}
+      additionalPrinterColumns:
+        - name: Status
+          type: string
+          description: resource status
+          jsonPath: .status.status
+        - name: Age
+          type: date
+          description: resource creation timestamp
+          format: date-time
+          jsonPath: .metadata.creationTimestamp

example/crd.yml

@@ -0,0 +1,49 @@
+---
+apiVersion: example.com/v1
+kind: Foo
+metadata:
+  name: one
+  labels:
+    color: green
+  finalizers:
+    - finalizer.foos.example.com
+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