<commit_before>#
# Fluentd Kubernetes Metadata Filter Plugin - Enrich Fluentd events with
# Kubernetes metadata
#
# Copyright 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

require_relative 'kubernetes_metadata_cache_strategy'
require_relative 'kubernetes_metadata_common'
require_relative 'kubernetes_metadata_stats'
require_relative 'kubernetes_metadata_watch_namespaces'
require_relative 'kubernetes_metadata_watch_pods'

require 'fluent/plugin/filter'
require 'resolv'

module Fluent::Plugin
  class KubernetesMetadataFilter < Fluent::Plugin::Filter
    K8_POD_CA_CERT = 'ca.crt'
    K8_POD_TOKEN = 'token'

    include KubernetesMetadata::CacheStrategy
    include KubernetesMetadata::Common
    include KubernetesMetadata::WatchNamespaces
    include KubernetesMetadata::WatchPods

    Fluent::Plugin.register_filter('kubernetes_metadata', self)

    config_param :kubernetes_url, :string, default: nil
    config_param :cache_size, :integer, default: 1000
    config_param :cache_ttl, :integer, default: 60 * 60
    config_param :watch, :bool, default: true
    config_param :apiVersion, :string, default: 'v1'
    config_param :client_cert, :string, default: nil
    config_param :client_key, :string, default: nil
    config_param :ca_file, :string, default: nil
    config_param :verify_ssl, :bool, default: true
    config_param :tag_to_kubernetes_name_regexp,
                 :string,
                 :default => 'var\.log\.containers\.(?<pod_name>[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\.log$'
    config_param :bearer_token_file, :string, default: nil
    config_param :secret_dir, :string, default: '/var/run/secrets/kubernetes.io/serviceaccount'
    config_param :de_dot, :bool, default: true
    config_param :de_dot_separator, :string, default: '_'
    # if reading from the journal, the record will contain the following fields in the following
    # format:
    # CONTAINER_NAME=k8s_$containername.$containerhash_$podname_$namespacename_$poduuid_$rand32bitashex
    # CONTAINER_FULL_ID=dockeridassha256hexvalue
    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
    # parse format is defined here: https://github.com/kubernetes/kubernetes/blob/release-1.6/pkg/kubelet/dockertools/docker.go#L317
    config_param :container_name_to_kubernetes_regexp,
                 :string,
                 :default => '^(?<name_prefix>[^_]+)_(?<container_name>[^\._]+)(\.(?<container_hash>[^_]+))?_(?<pod_name>[^_]+)_(?<namespace>[^_]+)_[^_]+_[^_]+$'

    config_param :annotation_match, :array, default: []
    config_param :stats_interval, :integer, default: 30
    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
    # if `ca_file` is for an intermediate CA, or otherwise we do not have the root CA and want
    # to trust the intermediate CA certs we do have, set this to `true` - this corresponds to
    # the openssl s_client -partial_chain flag and X509_V_FLAG_PARTIAL_CHAIN
    config_param :ssl_partial_chain, :bool, default: false
    config_param :skip_labels, :bool, default: false
    config_param :skip_container_metadata, :bool, default: false
    config_param :skip_master_url, :bool, default: false
    config_param :skip_namespace_metadata, :bool, default: false
    # The time interval in seconds for retry backoffs when watch connections fail.
    config_param :watch_retry_interval, :integer, default: 1
    # The base number of exponential backoff for retries.
    config_param :watch_retry_exponential_backoff_base, :integer, default: 2
    # The maximum number of times to retry pod and namespace watches.
    config_param :watch_retry_max_times, :integer, default: 10

    def fetch_pod_metadata(namespace_name, pod_name)
      log.trace("fetching pod metadata: #{namespace_name}/#{pod_name}") if log.trace?
      begin
        metadata = @client.get_pod(pod_name, namespace_name)
        unless metadata
          log.trace("no metadata returned for: #{namespace_name}/#{pod_name}") if log.trace?
          @stats.bump(:pod_cache_api_nil_not_found)
        else
          begin
            log.trace("raw metadata for #{namespace_name}/#{pod_name}: #{metadata}") if log.trace?
            metadata = parse_pod_metadata(metadata)
            @stats.bump(:pod_cache_api_updates)
            log.trace("parsed metadata for #{namespace_name}/#{pod_name}: #{metadata}") if log.trace?
            @cache[metadata['pod_id']] = metadata
            return metadata
          rescue Exception=>e
            log.debug(e)
            @stats.bump(:pod_cache_api_nil_bad_resp_payload)
            log.trace("returning empty metadata for #{namespace_name}/#{pod_name} due to error '#{e}'") if log.trace?
          end
        end
      rescue Exception=>e
        @stats.bump(:pod_cache_api_nil_error)
        log.debug "Exception '#{e}' encountered fetching pod metadata from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}"
      end
      {}
    end

    def dump_stats
      @curr_time = Time.now
      return if @curr_time.to_i - @prev_time.to_i < @stats_interval
      @prev_time = @curr_time
      @stats.set(:pod_cache_size, @cache.count)
      @stats.set(:namespace_cache_size, @namespace_cache.count) if @namespace_cache
      log.info(@stats)
      if log.level == Fluent::Log::LEVEL_TRACE
        log.trace("       id cache: #{@id_cache.to_a}")
        log.trace("      pod cache: #{@cache.to_a}")
        log.trace("namespace cache: #{@namespace_cache.to_a}")
      end
    end

    def fetch_namespace_metadata(namespace_name)
      log.trace("fetching namespace metadata: #{namespace_name}") if log.trace?
      begin
        metadata = @client.get_namespace(namespace_name)
        unless metadata
            log.trace("no metadata returned for: #{namespace_name}") if log.trace?
            @stats.bump(:namespace_cache_api_nil_not_found)
        else
          begin
            log.trace("raw metadata for #{namespace_name}: #{metadata}") if log.trace?
            metadata = parse_namespace_metadata(metadata)
            @stats.bump(:namespace_cache_api_updates)
            log.trace("parsed metadata for #{namespace_name}: #{metadata}") if log.trace?
            @namespace_cache[metadata['namespace_id']] = metadata
            return metadata
          rescue Exception => e
            log.debug(e)
            @stats.bump(:namespace_cache_api_nil_bad_resp_payload)
            log.trace("returning empty metadata for #{namespace_name} due to error '#{e}'") if log.trace?
          end
        end
      rescue Exception => kube_error
        @stats.bump(:namespace_cache_api_nil_error)
        log.debug "Exception '#{kube_error}' encountered fetching namespace metadata from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}"
      end<commit_msg>metadata cannot be nil (#253)<commit_after>#
# Fluentd Kubernetes Metadata Filter Plugin - Enrich Fluentd events with
# Kubernetes metadata
#
# Copyright 2017 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

require_relative 'kubernetes_metadata_cache_strategy'
require_relative 'kubernetes_metadata_common'
require_relative 'kubernetes_metadata_stats'
require_relative 'kubernetes_metadata_watch_namespaces'
require_relative 'kubernetes_metadata_watch_pods'

require 'fluent/plugin/filter'
require 'resolv'

module Fluent::Plugin
  class KubernetesMetadataFilter < Fluent::Plugin::Filter
    K8_POD_CA_CERT = 'ca.crt'
    K8_POD_TOKEN = '<PASSWORD>'

    include KubernetesMetadata::CacheStrategy
    include KubernetesMetadata::Common
    include KubernetesMetadata::WatchNamespaces
    include KubernetesMetadata::WatchPods

    Fluent::Plugin.register_filter('kubernetes_metadata', self)

    config_param :kubernetes_url, :string, default: nil
    config_param :cache_size, :integer, default: 1000
    config_param :cache_ttl, :integer, default: 60 * 60
    config_param :watch, :bool, default: true
    config_param :apiVersion, :string, default: 'v1'
    config_param :client_cert, :string, default: nil
    config_param :client_key, :string, default: nil
    config_param :ca_file, :string, default: nil
    config_param :verify_ssl, :bool, default: true
    config_param :tag_to_kubernetes_name_regexp,
                 :string,
                 :default => 'var\.log\.containers\.(?<pod_name>[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*)_(?<namespace>[^_]+)_(?<container_name>.+)-(?<docker_id>[a-z0-9]{64})\.log$'
    config_param :bearer_token_file, :string, default: nil
    config_param :secret_dir, :string, default: '/var/run/secrets/kubernetes.io/serviceaccount'
    config_param :de_dot, :bool, default: true
    config_param :de_dot_separator, :string, default: '_'
    # if reading from the journal, the record will contain the following fields in the following
    # format:
    # CONTAINER_NAME=k8s_$containername.$containerhash_$podname_$namespacename_$poduuid_$rand32bitashex
    # CONTAINER_FULL_ID=dockeridassha256hexvalue
    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
    # parse format is defined here: https://github.com/kubernetes/kubernetes/blob/release-1.6/pkg/kubelet/dockertools/docker.go#L317
    config_param :container_name_to_kubernetes_regexp,
                 :string,
                 :default => '^(?<name_prefix>[^_]+)_(?<container_name>[^\._]+)(\.(?<container_hash>[^_]+))?_(?<pod_name>[^_]+)_(?<namespace>[^_]+)_[^_]+_[^_]+$'

    config_param :annotation_match, :array, default: []
    config_param :stats_interval, :integer, default: 30
    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
    # if `ca_file` is for an intermediate CA, or otherwise we do not have the root CA and want
    # to trust the intermediate CA certs we do have, set this to `true` - this corresponds to
    # the openssl s_client -partial_chain flag and X509_V_FLAG_PARTIAL_CHAIN
    config_param :ssl_partial_chain, :bool, default: false
    config_param :skip_labels, :bool, default: false
    config_param :skip_container_metadata, :bool, default: false
    config_param :skip_master_url, :bool, default: false
    config_param :skip_namespace_metadata, :bool, default: false
    # The time interval in seconds for retry backoffs when watch connections fail.
    config_param :watch_retry_interval, :integer, default: 1
    # The base number of exponential backoff for retries.
    config_param :watch_retry_exponential_backoff_base, :integer, default: 2
    # The maximum number of times to retry pod and namespace watches.
    config_param :watch_retry_max_times, :integer, default: 10

    def fetch_pod_metadata(namespace_name, pod_name)
      log.trace("fetching pod metadata: #{namespace_name}/#{pod_name}") if log.trace?
      pod_object = @client.get_pod(pod_name, namespace_name)
      log.trace("raw metadata for #{namespace_name}/#{pod_name}: #{pod_object}") if log.trace?
      metadata = parse_pod_metadata(pod_object)
      @stats.bump(:pod_cache_api_updates)
      log.trace("parsed metadata for #{namespace_name}/#{pod_name}: #{metadata}") if log.trace?
      @cache[metadata['pod_id']] = metadata
    rescue Exception => e
      @stats.bump(:pod_cache_api_nil_error)
      log.debug "Exception '#{e}' encountered fetching pod metadata from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}"
      {}
    end

    def dump_stats
      @curr_time = Time.now
      return if @curr_time.to_i - @prev_time.to_i < @stats_interval
      @prev_time = @curr_time
      @stats.set(:pod_cache_size, @cache.count)
      @stats.set(:namespace_cache_size, @namespace_cache.count) if @namespace_cache
      log.info(@stats)
      if log.level == Fluent::Log::LEVEL_TRACE
        log.trace("       id cache: #{@id_cache.to_a}")
        log.trace("      pod cache: #{@cache.to_a}")
        log.trace("namespace cache: #{@namespace_cache.to_a}")
      end
    end

    def fetch_namespace_metadata(namespace_name)
      log.trace("fetching namespace metadata: #{namespace_name}") if log.trace?
      namespace_object = @client.get_namespace(namespace_name)
      log.trace("raw metadata for #{namespace_name}: #{namespace_object}") if log.trace?
      metadata = parse_namespace_metadata(namespace_object)
      @stats.bump(:namespace_cache_api_updates)
      log.trace("parsed metadata for #{namespace_name}: #{metadata}") if log.trace?
      @namespace_cache[metadata['namespace_id']] = metadata
    rescue Exception => kube_error
      @stats.bump(:namespace_cache_api_nil_error)
      log.debug "Exception '#{kube_error}' encountered fetching namespace metadata from Kubernetes API #{@apiVersion} endpoint #{@kubernetes_url}"
{'fraction_non_alphanumeric': 0.10817357602602257, 'fraction_numerical': 0.008374281957228873, 'mean_word_length': 4.033089515848136, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 32, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3825516', 'n_tokens_mistral': 4467, 'n_tokens_neox': 4330, 'n_words': 1142}