Method: Fluent::Plugin::ElasticsearchInput#client

Defined in:

lib/fluent/plugin/in_elasticsearch.rb

#client(host = nil) ⇒ Object

# File 'lib/fluent/plugin/in_elasticsearch.rb', line 207

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 }
    local_reload_connections = @reload_connections
    if local_reload_connections && @reload_after > DEFAULT_RELOAD_AFTER
      local_reload_connections = @reload_after
    end

    headers = { 'Content-Type' => "application/json" }.merge(@custom_headers)

    transport = TRANSPORT_CLASS::Transport::HTTP::Faraday.new(
      connection_options.merge(
        options: {
          reload_connections: local_reload_connections,
          reload_on_failure: @reload_on_failure,
          resurrect_after: @resurrect_after,
          logger: @transport_logger,
          transport_options: {
            headers: headers,
            request: { timeout: @request_timeout },
            ssl: { verify: @ssl_verify, ca_file: @ca_file, version: @ssl_version }
          },
          http: {
            user: @user,
            password: @password
          },
          sniffer_class: @sniffer_class,
        }), &adapter_conf)
    Elasticsearch::Client.new transport: transport
  end
end