Update Kubernetes plugin

plugins/kubernetes.yaml

@@ -1,4 +1,4 @@
-version: 0.0.5
+version: 0.0.6
 title: Kubernetes
 description: Log parser for Kubernetes
 parameters:
@@ -25,31 +25,71 @@ pipeline:
     type: file_input
     include:
       - {{ .container_log_path }}
-    # {{ if .start_at }}
-    start_at: {{ .start_at }}
-    # {{ end }}
+    start_at: {{ or .start_at "end" }}
     file_path_field: log_name
+    file_name_field: file_name
     write_to: log
-    output: container_json_parser
 
+  # Filter carbon logs. Check if file_name field starts with carbon. if it does drop the log entry otherwise continue down pipeline
+  - id: filename_router
+    type: router
+    routes:
+      - expr: '$record.file_name != nil and $record.file_name matches "^carbon"'
+        output: drop_output
+      - expr: true
+        output: remove_file_name
+
+  # Drop unwanted logs
+  - type: "drop_output"
+
+  # Remove file_name field. We have filtered carbon logs and no longer need file_name field
+  - id: remove_file_name
+    type: restructure
+    ops:
+      - remove: file_name
+
+  # Initial log entry should be safe to parse as JSON
   - id: container_json_parser
     type: json_parser
     parse_from: log
-    output: nested_json_router
 
-  - id: nested_json_router
+  # Attempt to parse nested JSON in log field if it exists and if JSON is detected
+  - id: log_json_router
     type: router
     routes:
+      # If there is no log field under record bypass json parser
+      - output: container_regex_parser
+        expr: '$record.log == nil'
+      # If log field is JSON with trailing newline route to regex_parser to remove newline before parsing.
+      - output: remove_new_line
+        expr: $record.log matches '^{.*}\\n$'
+      # It appears to be JSON so send it to be parsed as JSON.
       - output: nested_json_parser
         expr: $record.log matches '^{.*}$'
-      - output: container_regex_parser
+      # If log field doesn't appear to be JSON then, skip nested JSON parsers
+      - output: container_regex_parser 
         expr: true
 
+  # Remove new line from end of jsonUnable to parse json if it has a newline. 
+  - id: remove_new_line
+    type: regex_parser
+    parse_from: $record.log
+    regex: '(?P<log_tmp>{.*})\s+'
+    output: nested_tmp_json_parser
+
+  # We now have removed newline and saved log field to log_tmp
+  - id: nested_tmp_json_parser
+    type: json_parser
+    parse_from: $record.log_tmp
+    output: container_regex_parser
+
+  # Use this JSON parser since there was no trailing newline in log field.
   - id: nested_json_parser
     type: json_parser
-    parse_from: $.log
+    parse_from: $record.log
     output: container_regex_parser
 
+  # Log field has been parsed if possible and now we can parse log_name field for container information.
   - id: container_regex_parser
     type: regex_parser
     parse_from: log_name
@@ -58,23 +98,37 @@ pipeline:
       parse_from: stream
       preserve: true
       mapping:
-        error:
-          - stderr
-        info:
-          - stdout
+        error: stderr
+        info: stdout
     timestamp:
       parse_from: time
       layout: '%Y-%m-%dT%H:%M:%S.%sZ'
-    output: log_parse_router
 
-  - id: log_parse_router
+  # Check if there is a timestamp field in record and if it is in expected format. If so parse it otherwise continue down pipeline.
+  - id: timestamp_parser_router
+    type: router
+    routes:
+      - output: timestamp_parser
+        expr: '$record.timestamp != nil and $record.timestamp matches "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d+Z"'
+      - output: standard_regex_parser_router
+        expr: true
+
+  # Parse timestamp field in record.
+  - id: timestamp_parser
+    type: time_parser
+    parse_from: timestamp
+    layout: '%Y-%m-%dT%H:%M:%S.%sZ'
+
+  # Check if log field exists and if it matches format. If so parse it otherwise continue down the pipeline
+  - id: standard_regex_parser_router
     type: router
     routes:
       - output: standard_regex_parser
-        expr: '$record.log matches "^\\w\\d{4}"'
+        expr: '$record.log != nil and $record.log matches "^\\w\\d{4}"'
       - output: add_kubernetes_metadata
         expr: true
 
+  # Log field exists and matches expected format.
   - id: standard_regex_parser
     type: regex_parser
     parse_from: log
@@ -90,12 +144,13 @@ pipeline:
     timestamp:
       parse_from: timestamp
       layout: '%m%d %H:%M:%S.%s'
-    output: add_kubernetes_metadata
 
+  # Add kubernetes metadata
   - id: add_kubernetes_metadata
     type: k8s_metadata_decorator
     output: add_labels_router
 
+  # Add label log_name
   - id: add_labels_router
     type: router
     routes:
@@ -110,50 +165,58 @@ pipeline:
       - output: add_container_metadata
         expr: true
 
+  # Set log_name label to kubernetes.controller
   - id: add_kube_controller_metadata
     type: metadata
     labels:
       log_name: 'kubernetes.controller'
     output: {{ .output }}
 
+  # Set log_name label to kubernetes.scheduler
   - id: add_kube_scheduler_metadata
     type: metadata
     labels:
       log_name: 'kubernetes.scheduler'
     output: {{ .output }}
 
+  # Set log_name label to kubernetes.apiserver
   - id: add_kube_apiserver_metadata
     type: metadata
     labels:
       log_name: 'kubernetes.apiserver'
     output: {{ .output }}
 
+  # Set log_name label to kubernetes.proxy
   - id: add_kube_proxy_metadata
     type: metadata
     labels:
       log_name: 'kubernetes.proxy'
     output: {{ .output }}
 
+  # Set log_name label to kubernetes.container
   - id: add_container_metadata
     type: metadata
     labels:
       log_name: 'kubernetes.container'
     output: {{ .output }}
   # {{ end }}
 
+  # Use journald to gather kubelet logs. Use provider path for journald if available otherwise use default locations.
   - id: kubelet_reader
     type: journald_input
     # {{ if .kubelet_journald_log_path }}
     directory: {{ .kubelet_journald_log_path }}
     # {{ end }}
     output: kubelet_filter_router
 
+  # Only grab entry if it is the kubelet.server
   - id: kubelet_filter_router
     type: router
     routes:
       - output: kubelet_message_parser_router
         expr: '$record._SYSTEMD_UNIT == "kubelet.service"'
 
+  # If MESSAGE field matches format then, parse it otherwise send down the pipeline.
   - id: kubelet_message_parser_router
     type: router
     routes:
@@ -162,6 +225,7 @@ pipeline:
       - output: add_kublet_metadata
         expr: true
 
+  # MESSAGE field seems to match expected format.
   - id: message_regex_parser
     type: regex_parser
     parse_from: MESSAGE
@@ -179,6 +243,7 @@ pipeline:
       layout: '%m%d %H:%M:%S.%s'
     output: add_kublet_metadata
 
+  # Set log_name label to kubernetes.kubelet
   - id: add_kublet_metadata
     type: metadata
     labels: