Merge pull request #155 from josefkarasek/master_detect-metadata-source

deprecate use_journal - add lookup_from_k8s_field - detect input

Author: jcantrill

README.md

@@ -44,21 +44,33 @@ This must used named capture groups for `container_name`, `pod_name` & `namespac
 * `watch` - set up a watch on pods on the API server for updates to metadata (default: `true`)
 * `de_dot` - replace dots in labels and annotations with configured `de_dot_separator`, required for ElasticSearch 2.x compatibility (default: `true`)
 * `de_dot_separator` - separator to use if `de_dot` is enabled (default: `_`)
-* `use_journal` - If false (default), messages are expected to be formatted and tagged as if read by the fluentd in\_tail plugin with wildcard filename.  If true, messages are expected to be formatted as if read from the systemd journal.  The `MESSAGE` field has the full message.  The `CONTAINER_NAME` field has the encoded k8s metadata (see below).  The `CONTAINER_ID_FULL` field has the full container uuid.  This requires docker to use the `--log-driver=journald` log driver.
+* *DEPRECATED* `use_journal` - If false, messages are expected to be formatted and tagged as if read by the fluentd in\_tail plugin with wildcard filename.  If true, messages are expected to be formatted as if read from the systemd journal.  The `MESSAGE` field has the full message.  The `CONTAINER_NAME` field has the encoded k8s metadata (see below).  The `CONTAINER_ID_FULL` field has the full container uuid.  This requires docker to use the `--log-driver=journald` log driver.  If unset (the default), the plugin will use the `CONTAINER_NAME` and `CONTAINER_ID_FULL` fields
+if available, otherwise, will use the tag in the `tag_to_kubernetes_name_regexp` format.
 * `container_name_to_kubernetes_regexp` - The regular expression used to extract the k8s metadata encoded in the journal `CONTAINER_NAME` field (default: `'^(?<name_prefix>[^_]+)_(?<container_name>[^\._]+)(\.(?<container_hash>[^_]+))?_(?<pod_name>[^_]+)_(?<namespace>[^_]+)_[^_]+_[^_]+$'`
   * This corresponds to the definition [in the source](https://github.com/kubernetes/kubernetes/blob/master/pkg/kubelet/dockertools/docker.go#L317)
 * `annotation_match` - Array of regular expressions matching annotation field names. Matched annotations are added to a log record.
 * `allow_orphans` - Modify the namespace and namespace id to the values of `orphaned_namespace_name` and `orphaned_namespace_id`
 when true (default: `true`)
 * `orphaned_namespace_name` - The namespace to associate with records where the namespace can not be determined (default: `.orphaned`)
 * `orphaned_namespace_id` - The namespace id to associate with records where the namespace can not be determined (default: `orphaned`)
+* `lookup_from_k8s_field` - If the field `kubernetes` is present, lookup the metadata from the given subfields such as `kubernetes.namespace_name`, `kubernetes.pod_name`, etc.  This allows you to avoid having to pass in metadata to lookup in an explicitly formatted tag name or in an explicitly formatted `CONTAINER_NAME` value.  For example, set `kubernetes.namespace_name`, `kubernetes.pod_name`, `kubernetes.container_name`, and `docker.id` in the record, and the filter will fill in the rest. (default: `true`)
 
 **NOTE:** As of the release 2.1.x of this plugin, it no longer supports parsing the source message into JSON and attaching it to the
 payload.  The following configuration options are removed:
 
 * `merge_json_log`
 * `preserve_json_log`
 
+**NOTE** As of this release, the use of `use_journal` is **DEPRECATED**.  If this setting is not present, the plugin will
+attempt to figure out the source of the metadata fields from the following:
+- If `lookup_from_k8s_field true` (the default) and the following fields are present in the record:
+`docker.container_id`, `kubernetes.namespace_name`, `kubernetes.pod_name`, `kubernetes.container_name`,
+then the plugin will use those values as the source to use to lookup the metadata
+- If `use_journal true`, or `use_journal` is unset, and the fields `CONTAINER_NAME` and `CONTAINER_ID_FULL` are present in the record,
+then the plugin will parse those values using `container_name_to_kubernetes_regexp` and use those as the source to lookup the metadata
+- Otherwise, if the tag matches `tag_to_kubernetes_name_regexp`, the plugin will parse the tag and use those values to
+lookup the metdata
+
 Reading from the JSON formatted log files with `in_tail` and wildcard filenames:
 ```
 <source>

lib/fluent/plugin/filter_kubernetes_metadata.rb

@@ -57,7 +57,7 @@ class KubernetesMetadataFilter < Fluent::Plugin::Filter
     # format:
     # CONTAINER_NAME=k8s_$containername.$containerhash_$podname_$namespacename_$poduuid_$rand32bitashex
     # CONTAINER_FULL_ID=dockeridassha256hexvalue
-    config_param :use_journal, :bool, default: false
+    config_param :use_journal, :bool, default: nil
     # Field 2 is the container_hash, field 5 is the pod_id, and field 6 is the pod_randhex
     # I would have included them as named groups, but you can't have named groups that are
     # non-capturing :P
@@ -71,6 +71,7 @@ class KubernetesMetadataFilter < Fluent::Plugin::Filter
     config_param :allow_orphans, :bool, default: true
     config_param :orphaned_namespace_name, :string, default: '.orphaned'
     config_param :orphaned_namespace_id, :string, default: 'orphaned'
+    config_param :lookup_from_k8s_field, :bool, default: true
 
     def fetch_pod_metadata(namespace_name, pod_name)
       log.trace("fetching pod metadata: #{namespace_name}/#{pod_name}") if log.trace?
@@ -243,13 +244,10 @@ def log.trace?
           namespace_thread.abort_on_exception = true
         end
       end
-      if @use_journal
-        log.debug "Will stream from the journal"
-        self.class.class_eval { alias_method :filter_stream, :filter_stream_from_journal }
-      else
-        log.debug "Will stream from the files"
-        self.class.class_eval { alias_method :filter_stream, :filter_stream_from_files }
-      end
+      @time_fields = []
+      @time_fields.push('_SOURCE_REALTIME_TIMESTAMP', '__REALTIME_TIMESTAMP') if @use_journal || @use_journal.nil?
+      @time_fields.push('time') unless @use_journal
+      @time_fields.push('@timestamp') if @lookup_from_k8s_field
 
       @annotations_regexps = []
       @annotation_match.each do |regexp|
@@ -262,102 +260,93 @@ def log.trace?
 
     end
 
-    def get_metadata_for_record(match_data, container_id, create_time, batch_miss_cache)
-      namespace_name = match_data['namespace']
-      pod_name = match_data['pod_name']
-      container_name = match_data['container_name']
+    def get_metadata_for_record(namespace_name, pod_name, container_name, container_id, create_time, batch_miss_cache)
       metadata = {
-        'container_name'  => container_name,
-        'namespace_name'  => namespace_name,
-        'pod_name'        => pod_name
+        'docker' => {'container_id' => container_id},
+        'kubernetes' => {
+          'container_name'  => container_name,
+          'namespace_name'  => namespace_name,
+          'pod_name'        => pod_name
+        }
       }
       if @kubernetes_url.present?
         pod_metadata = get_pod_metadata(container_id, namespace_name, pod_name, create_time, batch_miss_cache)
 
         if (pod_metadata.include? 'containers') && (pod_metadata['containers'].include? container_id)
-          metadata['container_image'] = pod_metadata['containers'][container_id]['image']
-          metadata['container_image_id'] = pod_metadata['containers'][container_id]['image_id']
+          metadata['kubernetes']['container_image'] = pod_metadata['containers'][container_id]['image']
+          metadata['kubernetes']['container_image_id'] = pod_metadata['containers'][container_id]['image_id']
         end
 
-        metadata.merge!(pod_metadata) if pod_metadata
-        metadata.delete('containers')
+        metadata['kubernetes'].merge!(pod_metadata) if pod_metadata
+        metadata['kubernetes'].delete('containers')
       end
       metadata
     end
 
-    def create_time_from_record(record)
-        time = if @use_journal
-           record['_SOURCE_REALTIME_TIMESTAMP'].nil? ? record['_SOURCE_REALTIME_TIMESTAMP'] : record['__REALTIME_TIMESTAMP']
-        else
-          record['time']
-        end
-        (time.nil? || time.chop.empty?) ? Time.now : Time.parse(time)
+    def create_time_from_record(record, internal_time)
+      time_key = @time_fields.detect{ |ii| record.has_key?(ii) }
+      time = record[time_key]
+      if time.nil? || time.chop.empty?
+        return internal_time
+      end
+      if ['_SOURCE_REALTIME_TIMESTAMP', '__REALTIME_TIMESTAMP'].include?(time_key)
+        timei= time.to_i
+        return Time.at(timei / 1000000, timei % 1000000)
+      end
+      return Time.parse(time)
     end
 
     def filter_stream(tag, es)
-      es
-    end
-
-    def filter_stream_from_files(tag, es)
       return es if (es.respond_to?(:empty?) && es.empty?) || !es.is_a?(Fluent::EventStream)
       new_es = Fluent::MultiEventStream.new
-
-      match_data = tag.match(@tag_to_kubernetes_name_regexp_compiled)
+      tag_match_data = tag.match(@tag_to_kubernetes_name_regexp_compiled) unless @use_journal
+      tag_metadata = nil
       batch_miss_cache = {}
-      metadata = nil
-
       es.each do |time, record|
-        if match_data && metadata.nil?
-          container_id = match_data['docker_id']
-          metadata = {
-              'docker' => {
-                  'container_id' => container_id
-              },
-              'kubernetes' => get_metadata_for_record(match_data, container_id, create_time_from_record(record), batch_miss_cache)
-          }
+        if tag_match_data && tag_metadata.nil?
+          tag_metadata = get_metadata_for_record(tag_match_data['namespace'], tag_match_data['pod_name'], tag_match_data['container_name'],
+            tag_match_data['docker_id'], create_time_from_record(record, time), batch_miss_cache)
+        end
+        metadata = Marshal.load(Marshal.dump(tag_metadata)) if tag_metadata
+        if (@use_journal || @use_journal.nil?) &&
+          (j_metadata = get_metadata_for_journal_record(record, time, batch_miss_cache))
+          metadata = j_metadata
+        end
+        if @lookup_from_k8s_field && record.has_key?('kubernetes') && record.has_key?('docker') &&
+          record['kubernetes'].respond_to?(:has_key?) && record['docker'].respond_to?(:has_key?) &&
+          record['kubernetes'].has_key?('namespace_name') &&
+          record['kubernetes'].has_key?('pod_name') &&
+          record['kubernetes'].has_key?('container_name') &&
+          record['docker'].has_key?('container_id') &&
+          (k_metadata = get_metadata_for_record(record['kubernetes']['namespace_name'], record['kubernetes']['pod_name'],
+            record['kubernetes']['container_name'], record['docker']['container_id'],
+            create_time_from_record(record, time), batch_miss_cache))
+            metadata = k_metadata
         end
 
-        record = record.merge(Marshal.load(Marshal.dump(metadata))) if metadata
+        record = record.merge(metadata) if metadata
         new_es.add(time, record)
       end
       dump_stats
       new_es
     end
 
-    def filter_stream_from_journal(tag, es)
-      return es if (es.respond_to?(:empty?) && es.empty?) || !es.is_a?(Fluent::EventStream)
-      new_es = Fluent::MultiEventStream.new
-      batch_miss_cache = {}
-      es.each do |time, record|
-        metadata = nil
-        if record.has_key?('CONTAINER_NAME') && record.has_key?('CONTAINER_ID_FULL')
-          metadata = record['CONTAINER_NAME'].match(@container_name_to_kubernetes_regexp_compiled) do |match_data|
-           container_id = record['CONTAINER_ID_FULL']
-            metadata = {
-              'docker' => {
-                'container_id' => container_id
-              },
-              'kubernetes' => get_metadata_for_record(match_data, container_id, create_time_from_record(record), batch_miss_cache)
-            }
-
-            metadata
-          end
-          unless metadata
-            log.debug "Error: could not match CONTAINER_NAME from record #{record}"
-            @stats.bump(:container_name_match_failed)
-          end
-        elsif record.has_key?('CONTAINER_NAME') && record['CONTAINER_NAME'].start_with?('k8s_')
-          log.debug "Error: no container name and id in record #{record}"
-          @stats.bump(:container_name_id_missing)
+    def get_metadata_for_journal_record(record, time, batch_miss_cache)
+      metadata = nil
+      if record.has_key?('CONTAINER_NAME') && record.has_key?('CONTAINER_ID_FULL')
+        metadata = record['CONTAINER_NAME'].match(@container_name_to_kubernetes_regexp_compiled) do |match_data|
+          get_metadata_for_record(match_data['namespace'], match_data['pod_name'], match_data['container_name'],
+            record['CONTAINER_ID_FULL'], create_time_from_record(record, time), batch_miss_cache)
         end
-
-        record = record.merge(metadata) if metadata
-
-        new_es.add(time, record)
+        unless metadata
+          log.debug "Error: could not match CONTAINER_NAME from record #{record}"
+          @stats.bump(:container_name_match_failed)
+        end
+      elsif record.has_key?('CONTAINER_NAME') && record['CONTAINER_NAME'].start_with?('k8s_')
+        log.debug "Error: no container name and id in record #{record}"
+        @stats.bump(:container_name_id_missing)
       end
-
-      dump_stats
-      new_es
+      metadata
     end
 
     def de_dot!(h)