class Fluent::Plugin::ElasticsearchOutputDynamic

Constants

DYNAMIC_PARAM_NAMES
DYNAMIC_PARAM_SYMBOLS
RequestInfo

Attributes

dynamic_config[R]

Public Instance Methods

client(host = nil) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 38
def client(host = nil)
  # check here to see if we already have a client connection for the given host
  connection_options = get_connection_options(host)

  @_es = nil unless is_existing_connection(connection_options[:hosts])

  @_es ||= begin
    @current_config = connection_options[:hosts].clone
    adapter_conf = lambda {|f| f.adapter @http_backend, @backend_options }
    transport = Elasticsearch::Transport::Transport::HTTP::Faraday.new(connection_options.merge(
                                                                        options: {
                                                                          reload_connections: @reload_connections,
                                                                          reload_on_failure: @reload_on_failure,
                                                                          resurrect_after: @resurrect_after,
                                                                          logger: @transport_logger,
                                                                          transport_options: {
                                                                            headers: { 'Content-Type' => @content_type.to_s },
                                                                            request: { timeout: @request_timeout },
                                                                            ssl: { verify: @ssl_verify, ca_file: @ca_file, version: @ssl_version }
                                                                          },
                                                                          http: {
                                                                            user: @user,
                                                                            password: @password
                                                                          }
                                                                        }), &adapter_conf)
    Elasticsearch::Client.new transport: transport
  end
end
configure(conf) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 20
def configure(conf)
  super

  # evaluate all configurations here
  @dynamic_config = {}
  DYNAMIC_PARAM_SYMBOLS.each_with_index { |var, i|
    value = expand_param(self.instance_variable_get(var), nil, nil, nil)
    key = DYNAMIC_PARAM_NAMES[i]
    @dynamic_config[key] = value.to_s
  }
  # end eval all configs
end
connection_options_description(host) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 100
def connection_options_description(host)
  get_connection_options(host)[:hosts].map do |host_info|
    attributes = host_info.dup
    attributes[:password] = 'obfuscated' if attributes.has_key?(:password)
    attributes.inspect
  end.join(', ')
end
create_meta_config_map() click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 33
def create_meta_config_map
  {'id_key' => '_id', 'parent_key' => '_parent', 'routing_key' => @routing_key_name}
end
eval_or_val(var) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 224
def eval_or_val(var)
  return var unless var.is_a?(String)
  eval(var)
end
expand_param(param, tag, time, record) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 229
def expand_param(param, tag, time, record)
  # check for '${ ... }'
  #   yes => `eval`
  #   no  => return param
  return param if (param =~ /\${.+}/).nil?

  # check for 'tag_parts[]'
    # separated by a delimiter (default '.')
  tag_parts = tag.split(@delimiter) unless (param =~ /tag_parts\[.+\]/).nil? || tag.nil?

  # pull out section between ${} then eval
  inner = param.clone
  while inner.match(/\${.+}/)
    to_eval = inner.match(/\${(.+?)}/){$1}

    if !(to_eval =~ /record\[.+\]/).nil? && record.nil?
      return to_eval
    elsif !(to_eval =~/tag_parts\[.+\]/).nil? && tag_parts.nil?
      return to_eval
    elsif !(to_eval =~/time/).nil? && time.nil?
      return to_eval
    else
      inner.sub!(/\${.+?}/, eval( to_eval ))
    end
  end
  inner
end
get_connection_options(con_host) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 67
def get_connection_options(con_host)
  raise "`password` must be present if `user` is present" if @user && !@password

  hosts = if con_host || @hosts
    (con_host || @hosts).split(',').map do |host_str|
      # Support legacy hosts format host:port,host:port,host:port...
      if host_str.match(%r{^[^:]+(\:\d+)?$})
        {
          host:   host_str.split(':')[0],
          port:   (host_str.split(':')[1] || @port).to_i,
          scheme: @scheme.to_s
        }
      else
        # New hosts format expects URLs such as http://logs.foo.com,https://john:pass@logs2.foo.com/elastic
        uri = URI(get_escaped_userinfo(host_str))
        %w(user password path).inject(host: uri.host, port: uri.port, scheme: uri.scheme) do |hash, key|
          hash[key.to_sym] = uri.public_send(key) unless uri.public_send(key).nil? || uri.public_send(key) == ''
          hash
        end
      end
    end.compact
  else
    [{host: @host, port: @port.to_i, scheme: @scheme.to_s}]
  end.each do |host|
    host.merge!(user: @user, password: @password) if !host[:user] && @user
    host.merge!(path: @path) if !host[:path] && @path
  end

  {
    hosts: hosts
  }
end
is_valid_expand_param_type(param) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 257
def is_valid_expand_param_type(param)
  return false if [:@buffer_type].include?(param)
  return self.instance_variable_get(param).is_a?(String)
end
multi_workers_ready?() click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 108
def multi_workers_ready?
  true
end
send_bulk(data, host, index) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 211
def send_bulk(data, host, index)
  begin
    response = client(host).bulk body: data, index: index
    if response['errors']
      log.error "Could not push log to Elasticsearch: #{response}"
    end
  rescue => e
    @_es = nil if @reconnect_on_error
    # FIXME: identify unrecoverable errors and raise UnrecoverableRequestFailure instead
    raise RecoverableRequestFailure, "could not push logs to Elasticsearch cluster (#{connection_options_description(host)}): #{e.message}"
  end
end
write(chunk) click to toggle source
# File lib/fluent/plugin/out_elasticsearch2_dynamic.rb, line 112
def write(chunk)
  bulk_message = Hash.new { |h,k| h[k] = '' }
  dynamic_conf = @dynamic_config.clone

  headers = {
    UPDATE_OP => {},
    UPSERT_OP => {},
    CREATE_OP => {},
    INDEX_OP => {}
  }

  tag = chunk.metadata.tag

  chunk.msgpack_each do |time, record|
    next unless record.is_a? Hash

    begin
      # evaluate all configurations here
      DYNAMIC_PARAM_SYMBOLS.each_with_index { |var, i|
        k = DYNAMIC_PARAM_NAMES[i]
        v = self.instance_variable_get(var)
        # check here to determine if we should evaluate
        if dynamic_conf[k] != v
          value = expand_param(v, tag, time, record)
          dynamic_conf[k] = value
        end
      }
    # end eval all configs
    rescue => e
      # handle dynamic parameters misconfigurations
      router.emit_error_event(tag, time, record, e)
      next
    end

    if eval_or_val(dynamic_conf['logstash_format']) || eval_or_val(dynamic_conf['include_timestamp'])
      if record.has_key?("@timestamp")
        time = Time.parse record["@timestamp"]
      elsif record.has_key?(dynamic_conf['time_key'])
        time = Time.parse record[dynamic_conf['time_key']]
        record['@timestamp'] = record[dynamic_conf['time_key']] unless time_key_exclude_timestamp
      else
        record.merge!({"@timestamp" => Time.at(time).iso8601(@time_precision)})
      end
    end

    if eval_or_val(dynamic_conf['logstash_format'])
      if eval_or_val(dynamic_conf['utc_index'])
        target_index = "#{dynamic_conf['logstash_prefix']}#{@logstash_prefix_separator}#{Time.at(time).getutc.strftime("#{dynamic_conf['logstash_dateformat']}")}"
      else
        target_index = "#{dynamic_conf['logstash_prefix']}#{@logstash_prefix_separator}#{Time.at(time).strftime("#{dynamic_conf['logstash_dateformat']}")}"
      end
    else
      target_index = dynamic_conf['index_name']
    end

    # Change target_index to lower-case since Elasticsearch doesn't
    # allow upper-case characters in index names.
    target_index = target_index.downcase

    if @include_tag_key
      record.merge!(dynamic_conf['tag_key'] => tag)
    end

    if dynamic_conf['hosts']
      host = dynamic_conf['hosts']
    else
      host = "#{dynamic_conf['host']}:#{dynamic_conf['port']}"
    end

    if @include_index_in_url
      key = RequestInfo.new(host, target_index)
      meta = {"_type" => dynamic_conf['type_name']}
    else
      key = RequestInfo.new(host, nil)
      meta = {"_index" => target_index, "_type" => dynamic_conf['type_name']}
    end

    @meta_config_map.each_pair do |config_name, meta_key|
      if dynamic_conf[config_name] && accessor = record_accessor_create(dynamic_conf[config_name])
        if raw_value = accessor.call(record)
          meta[meta_key] = raw_value
        end
      end
    end

    if @remove_keys
      @remove_keys.each { |key| record.delete(key) }
    end

    write_op = dynamic_conf["write_operation"]
    append_record_to_messages(write_op, meta, headers[write_op], record, bulk_message[key])
  end

  bulk_message.each do |info, msgs|
    send_bulk(msgs, info.host, info.index) unless msgs.empty?
    msgs.clear
  end
end