Method: Fluent::Plugin::Zebrium#get_request_headers

Defined in:

lib/fluent/plugin/out_zebrium.rb

#get_request_headers(chunk_tag, record) ⇒ Object

# File 'lib/fluent/plugin/out_zebrium.rb', line 513

def get_request_headers(chunk_tag, record)
  headers = {}
  ids = {}
  cfgs = {}
  tags = {}

  # Sometimes 'record' appears to be a simple number, which causes an exception when
  # used as a hash. Until the underlying issue is addressed detect this and log.
  if record.class.name != "Hash"  or not record.respond_to?(:key?)
    log.error("Record is not a hash, unable to process (class: ${record.class.name}).")
    return false, nil, nil
  end

  if record.key?("docker") and not record.fetch("docker").nil?
      container_id = record["docker"]["container_id"]
      if record.key?("kubernetes") and not record.fetch("kubernetes").nil?
        cfgs["container_id"] = container_id
      else
        ids["container_id"] = container_id
      end
  end

  is_container_log = true
  log_type = ""
  forwarded_log = false
  user_mapping = false
  fpath = ""
  override_deployment = ""
  override_deployment_from_ns_svcgrp_map = ""

  record_host = ""
  if record.key?("host") and not record["host"].empty?
    record_host = record["host"]
  end
  has_container_keys = false
  if record.key?("container_id") and record.key?("container_name")
    has_container_keys = true
  end
  if chunk_tag =~ /^sysloghost\./ or chunk_tag =~ /^#{ze_forward_tag}\./
    if record_host.empty? and ze_host_in_logpath > 0 and record.key?("tailed_path")
  tailed_path = record["tailed_path"]
  path_components = tailed_path.split("/")
  if path_components.length() < ze_host_in_logpath 
log.info("Cannot find host at index #{ze_host_in_logpath} in '#{tailed_path}'")
  else
# note .split has empty first element from initial '/'
record_host = path_components[ze_host_in_logpath]
  end
    end
    log_type = "syslog"
    forwarded_log = true
    logbasename = "syslog"
    ids["app"] = logbasename
    ids["host"] = record_host
    is_container_log = false
  elsif record.key?("kubernetes") and not record.fetch("kubernetes").nil?
    kubernetes = record["kubernetes"]
    if kubernetes.key?("namespace_name") and not kubernetes.fetch("namespace_name").nil?
      namespace = kubernetes.fetch("namespace_name")
      if namespace.casecmp?("orphaned") or namespace.casecmp?(".orphaned")
        return false, nil, nil
      end
    end
    fpath = kubernetes["container_name"]
    keys = [ "namespace_name", "host", "container_name" ]
    for k in keys do
        if kubernetes.key?(k) and not kubernetes.fetch(k).nil?
          ids[k] = kubernetes[k]
          if k == "host" and @k8s_hostname.empty?
             @k8s_hostname = kubernetes[k]
          end
          # Requirement for ZS-2185 add cmdb_role, based on namespace_name
          if k == "namespace_name" 
              cfgs["cmdb_role"] = kubernetes[k].gsub("-","_")
              if @ns_to_svcgrp_mappings.active
                if @ns_to_svcgrp_mappings.svcgrps.key?(kubernetes[k]) and not @ns_to_svcgrp_mappings.svcgrps.fetch(kubernetes[k]).nil?
                  override_deployment_from_ns_svcgrp_map = @ns_to_svcgrp_mappings.svcgrps[kubernetes[k]]
                end
              end
          end
        end
    end

    for pattern in [ @pod_name_to_deployment_name_regexp_long_compiled, @pod_name_to_deployment_name_regexp_short_compiled ] do
        match_data = kubernetes["pod_name"].match(pattern)
        if match_data
            ids["deployment_name"] = match_data["deployment_name"]
            break
        end
    end
    keys = [ "namespace_id", "container_name", "pod_name", "pod_id", "container_image", "container_image_id" ]
    for k in keys do
        if kubernetes.key?(k) and not kubernetes.fetch(k).nil?
          cfgs[k] = kubernetes[k]
        end
    end
    unless kubernetes["labels"].nil?
      cfgs.merge!(kubernetes["labels"])
    end
    # Allow Cluster name to be set via an env variable
    if !@cluster_name.nil? and !@cluster_name.empty?
      cfgs["cluster_name"] = @cluster_name
    end

    # At this point k8s config should be set. Save these so a subsequent file-log
    # record for the same pod_id can use them.
    save_kubernetes_cfgs(cfgs)
    unless kubernetes["namespace_annotations"].nil?
      tags = kubernetes["namespace_annotations"]
      for t in tags.keys
        if t == "zebrium.com/ze_service_group" and not tags[t].empty?
          override_deployment = tags[t]
        end
      end
    end

    unless kubernetes["annotations"].nil?
      tags = kubernetes["annotations"]
      for t in tags.keys
        if t == "zebrium.com/ze_logtype" and not tags[t].empty?
          user_mapping = true
          logbasename = tags[t]
        end
        if t == "zebrium.com/ze_service_group" and not tags[t].empty?
          override_deployment = tags[t]
        end
      end
    end

    unless kubernetes["labels"].nil?
      for k in kubernetes["labels"].keys
        if k == "zebrium.com/ze_logtype" and not kubernetes["labels"][k].empty?
          user_mapping = true
          logbasename = kubernetes["labels"][k]
        end
        if k == "zebrium.com/ze_service_group" and not kubernetes["labels"][k].empty?
          override_deployment = kubernetes["labels"][k]
        end
      end
    end
    if not user_mapping
      logbasename = kubernetes["container_name"]
    end
  elsif chunk_tag =~ /^containers\./
    if record.key?("tailed_path")
      fpath = record["tailed_path"]
      fname = File.basename(fpath)
      ary = fname.split('-')
      container_id = ""
      if ary.length == 2
        container_id = ary[0]
        cm = get_container_meta_data(container_id)
        if cm.nil?
          return false, headers, nil
        end
        cfgs["container_id"] = container_id
        cfgs["container_name"] = cm['name']
        labels = cm['labels']
        for k in labels.keys do
          cfgs[k] = labels[k]
          if k == "zebrium.com/ze_logtype" and not labels[k].empty?
            user_mapping = true
            logbasename = labels[k]
          end
          if k == "zebrium.com/ze_service_group" and not labels[k].empty?
            override_deployment = labels[k]
          end
        end
        if not user_mapping
          logbasename = cm['name']
        end
        ids["app"] = logbasename
        cfgs["image"] = cm['image']
      else
        log.error("Wrong container log file: ", fpath)
      end
    else
      log.error("Missing tailed_path on logs with containers.* tag")
    end
  elsif has_container_keys
    logbasename = record['container_name'].sub(/^\//, '')
    ids["app"] = logbasename
    cfgs["container_id"] = record['container_id']
    cfgs["container_name"] = logbasename
  else
    is_container_log = false
    if record.key?("tailed_path")
      fpath = record["tailed_path"]
      fbname = File.basename(fpath, ".*")
      if @file_mappings.key?(fpath)
        logbasename = @file_mappings[fpath]
        user_mapping = true
        ids["ze_logname"] = fbname
      else
        logbasename = fbname.split('.')[0]
        if logbasename != fbname
          ids["ze_logname"] = fbname
        end
      end
    elsif record.key?("_SYSTEMD_UNIT")
      logbasename = record["_SYSTEMD_UNIT"].gsub(/\.service$/, '')
    elsif chunk_tag =~ /^k8s\.events/
      logbasename = "zk8s-events"
      # Allow Cluster name to be set via an env variable
      if !@cluster_name.nil? and !@cluster_name.empty?
        cfgs["cluster_name"] = @cluster_name
      end
    elsif chunk_tag =~ /^ztcp\.events\./
      ids["host"] = record_host.empty? ? "ztcp_host": record["host"]
      logbasename = record["logbasename"] ? record["logbasename"] : "ztcp_stream"
      forwarded_log = true
      log_type = "tcp_forward"
    elsif chunk_tag =~ /^zhttp\.events\./
      ids["host"] = record_host.empty? ? "ztttp_host" : record["host"]
      logbasename = record["logbasename"] ? record["logbasename"] : "zhttp_stream"
      forwarded_log = true
      log_type = "http_forward"
    else
      # Default goes to zlog-collector. Usually there are fluentd generated message
      # and our own log messages
      # for these generic messages, we will send as json messages
      return true, {}, nil
    end
    ids["app"] = logbasename
  end
  cfgs["ze_file_path"] = fpath
  if not ids.key?("host") or ids.fetch("host").nil?
    if record_host.empty?
      ids["host"] = get_host()
    else
      ids["host"] = record_host
    end
  end
  unless @ze_deployment_name.empty?
    ids["ze_deployment_name"] = @ze_deployment_name
  end
  unless override_deployment_from_ns_svcgrp_map.empty?
    log.debug("Updating ze_deployment_name ns_svcgrp_map '#{override_deployment_from_ns_svcgrp_map}'")
    ids["ze_deployment_name"] = override_deployment_from_ns_svcgrp_map
  end
  unless override_deployment.empty?
    log.debug("Updating ze_deployment_name to '#{override_deployment}'")
    ids["ze_deployment_name"] = override_deployment
  end
  for k in @ze_tags.keys do
    tags[k] = @ze_tags[k]
  end
  tags["fluentd_tag"] = chunk_tag
  
  id_key = ""
  keys = ids.keys.sort
  keys.each do |k|
    if ids.key?(k)
      if id_key.empty?
        id_key = k + "=" + ids[k]
      else
        id_key = id_key + "," + k + "=" + ids[k]
      end
    end
  end

  if record.key?("tailed_path")
    map_path_ids(record["tailed_path"], ids, cfgs, tags)
    add_kubernetes_cfgs_for_pod_id(cfgs)
  end

  # host should be handled as a config element instead of an id.
  # This is used when host changes frequently, causing issues with
  # detection. The actual host is stored in the cfgs metadata, and
  # a constant is stored in the ids metadata.
  # Note that a host entry must be present in ids for correct backend
  # processing, it is simply a constant at this point.
  if ze_handle_host_as_config && ids.key?("host")
    cfgs["host"] = ids["host"]
    ids["host"] = "host_in_config"
  end

  has_stream_token = false
  if @stream_tokens.key?(id_key)
      # Make sure there is no meta data change. If there is change, new stream token
      # must be requested.
      cfgs_tags_match = true
      if (cfgs.length == @stream_tokens[id_key]['cfgs'].length &&
              tags.length == @stream_tokens[id_key]['tags'].length)
          @stream_tokens[id_key]['cfgs'].keys.each do |k|
              old_cfg = @stream_tokens[id_key]['cfgs'][k]
              if old_cfg != cfgs[k]
                  log.info("Stream " + id_key + " config has changed: old " + old_cfg + ", new " + cfgs[k])
                  cfgs_tags_match = false
                  break
              end
          end
          @stream_tokens[id_key]['tags'].keys.each do |k|
              old_tag = @stream_tokens[id_key]['tags'][k]
              if old_tag !=  tags[k]
                  log.info("Stream " + id_key + " config has changed: old " + old_tag + ", new " + tags[k])
                  cfgs_tags_match = false
                  break
              end
          end
      else
          log.info("Stream " + id_key + " number of config or tag has changed")
          cfgs_tags_match = false
      end
      if cfgs_tags_match
          has_stream_token = true
      end
  end

  if has_stream_token
      stream_token = @stream_tokens[id_key]["token"]
  else
      log.info("Request new stream token with key " + id_key)
      stream_token = get_stream_token(ids, cfgs, tags, logbasename, is_container_log, user_mapping,
                                      log_type, forwarded_log)
      @stream_tokens[id_key] = {
                                 "token" => stream_token,
                                 "cfgs"  => cfgs,
                                 "tags"  => tags
                               }
  end

  # User can use node label on pod to override "host" meta data from kubernetes
  headers["authtoken"] = stream_token
  headers["Content-Type"] = "application/json"
  headers["Transfer-Encoding"] = "chunked"
  return true, headers, stream_token
end