Module: LogStash::Outputs::ElasticSearch::HttpClientBuilder

Defined in:
lib/logstash/outputs/elasticsearch/http_client_builder.rb

Class Method Summary collapse

Class Method Details

.build(logger, hosts, params) ⇒ Object

Raises:

  • (LogStash::ConfigurationError) 


6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 6

def self.build(logger, hosts, params)
  client_settings = {
    :pool_max => params["pool_max"],
    :pool_max_per_route => params["pool_max_per_route"],
    :check_connection_timeout => params["validate_after_inactivity"],
    :compression_level => params["compression_level"],
    :headers => params["custom_headers"] || {}
  }
  
  client_settings[:proxy] = params["proxy"] if params["proxy"]
  
  common_options = {
    :license_checker => params["license_checker"],
    :client_settings => client_settings,
    :metric => params["metric"],
    :resurrect_delay => params["resurrect_delay"]
  }

  if params["sniffing"]
    common_options[:sniffing] = true
    common_options[:sniffer_delay] = params["sniffing_delay"]
  end

  common_options[:timeout] = params["timeout"] if params["timeout"]

  if params["path"]
    client_settings[:path] = dedup_slashes("/#{params["path"]}/")
  end

  common_options[:bulk_path] = if params["bulk_path"]
    resolve_filter_path(dedup_slashes("/#{params["bulk_path"]}"))
  else
    resolve_filter_path(dedup_slashes("/#{params["path"]}/_bulk"))
  end

  common_options[:sniffing_path] = if params["sniffing_path"]
     dedup_slashes("/#{params["sniffing_path"]}")
  else
     dedup_slashes("/#{params["path"]}/_nodes/http")
  end

  common_options[:healthcheck_path] = if params["healthcheck_path"]
     dedup_slashes("/#{params["healthcheck_path"]}")
  else
     dedup_slashes("/#{params["path"]}")
  end

  if params["parameters"]
    client_settings[:parameters] = params["parameters"]
  end

  logger.debug? && logger.debug("Normalizing http path", :path => params["path"], :normalized => client_settings[:path])

  client_settings.merge! setup_ssl(logger, params)
  common_options.merge! setup_basic_auth(logger, params)
  client_settings[:headers].merge! setup_api_key(logger, params)

  external_version_types = ["external", "external_gt", "external_gte"]
  # External Version validation
  raise(
    LogStash::ConfigurationError,
    "External versioning requires the presence of a version number."
  ) if external_version_types.include?(params.fetch('version_type', '')) and params.fetch("version", nil) == nil
 

  # Create API setup
  raise(
    LogStash::ConfigurationError,
    "External versioning is not supported by the create action."
  ) if params['action'] == 'create' and external_version_types.include?(params.fetch('version_type', ''))

  # Update API setup
  raise( LogStash::ConfigurationError,
    "doc_as_upsert and scripted_upsert are mutually exclusive."
  ) if params["doc_as_upsert"] and params["scripted_upsert"]

  raise(
    LogStash::ConfigurationError,
    "Specifying action => 'update' needs a document_id."
  ) if params['action'] == 'update' and params.fetch('document_id', '') == ''

  raise(
    LogStash::ConfigurationError,
    "External versioning is not supported by the update action. See https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-update.html."
  ) if params['action'] == 'update' and external_version_types.include?(params.fetch('version_type', ''))

  # Update API setup
  update_options = {
    :doc_as_upsert => params["doc_as_upsert"],
    :script_var_name => params["script_var_name"],
    :script_type => params["script_type"],
    :script_lang => params["script_lang"],
    :scripted_upsert => params["scripted_upsert"]
  }
  common_options.merge! update_options if params["action"] == 'update'

  create_http_client(common_options.merge(:hosts => hosts, :logger => logger))
end

.create_http_client(options) ⇒ Object 



105
106
107
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 105

def self.create_http_client(options)
  LogStash::Outputs::ElasticSearch::HttpClient.new(options)
end

.setup_api_key(logger, params) ⇒ Object 



188
189
190
191
192
193
194
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 188

def self.setup_api_key(logger, params)
  api_key = params["api_key"]

  return {} unless (api_key && api_key.value)

  { "Authorization" => "ApiKey " + Base64.strict_encode64(api_key.value) }
end

.setup_basic_auth(logger, params) ⇒ Object 



177
178
179
180
181
182
183
184
185
186
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 177

def self.setup_basic_auth(logger, params)
  user, password = params["user"], params["password"]
  
  return {} unless user && password && password.value

  {
    :user => CGI.escape(user),
    :password => CGI.escape(password.value)
  }
end

.setup_ssl(logger, params) ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 109

def self.setup_ssl(logger, params)
  params["ssl_enabled"] = true if params["hosts"].any? {|h| h.scheme == "https" }
  return {} if params["ssl_enabled"].nil?

  return {:ssl => {:enabled => false}} if params["ssl_enabled"] == false

  ssl_certificate_authorities, ssl_truststore_path, ssl_certificate, ssl_keystore_path = params.values_at('ssl_certificate_authorities', 'ssl_truststore_path', 'ssl_certificate', 'ssl_keystore_path')

  if ssl_certificate_authorities && ssl_truststore_path
    raise LogStash::ConfigurationError, 'Use either "ssl_certificate_authorities/cacert" or "ssl_truststore_path/truststore" when configuring the CA certificate'
  end

  if ssl_certificate && ssl_keystore_path
    raise LogStash::ConfigurationError, 'Use either "ssl_certificate" or "ssl_keystore_path/keystore" when configuring client certificates'
  end

  ssl_options = {:enabled => true}

  if ssl_certificate_authorities&.any?
    raise LogStash::ConfigurationError, 'Multiple values on "ssl_certificate_authorities" are not supported by this plugin' if ssl_certificate_authorities.size > 1
    ssl_options[:ca_file] = ssl_certificate_authorities.first
  end

  setup_ssl_store(ssl_options, 'truststore', params)
  setup_ssl_store(ssl_options, 'keystore', params)

  ssl_key = params["ssl_key"]
  if ssl_certificate
    raise LogStash::ConfigurationError, 'Using an "ssl_certificate" requires an "ssl_key"' unless ssl_key
    ssl_options[:client_cert] = ssl_certificate
    ssl_options[:client_key] = ssl_key
  elsif !ssl_key.nil?
    raise LogStash::ConfigurationError, 'An "ssl_certificate" is required when using an "ssl_key"'
  end

  ssl_verification_mode = params["ssl_verification_mode"]
  unless ssl_verification_mode.nil?
    case ssl_verification_mode
    when 'none'
      logger.warn "You have enabled encryption but DISABLED certificate verification, " +
                    "to make sure your data is secure set `ssl_verification_mode => full`"
      ssl_options[:verify] = :disable
    else
      # Manticore's :default maps to Apache HTTP Client's DefaultHostnameVerifier,
      # which is the modern STRICT verifier that replaces the deprecated StrictHostnameVerifier
      ssl_options[:verify] = :default
    end
  end

  ssl_options[:cipher_suites] = params["ssl_cipher_suites"] if params.include?("ssl_cipher_suites")
  ssl_options[:trust_strategy] = params["ssl_trust_strategy"] if params.include?("ssl_trust_strategy")

  protocols = params['ssl_supported_protocols']
  ssl_options[:protocols] = protocols if protocols && protocols.any?

  { ssl: ssl_options }
end

.setup_ssl_store(ssl_options, kind, params) ⇒ Object

Parameters:

  • kind —

    is a string [truststore|keystore]



168
169
170
171
172
173
174
175
 
# File 'lib/logstash/outputs/elasticsearch/http_client_builder.rb', line 168

def self.setup_ssl_store(ssl_options, kind, params)
  store_path = params["ssl_#{kind}_path"]
  if store_path
    ssl_options[kind.to_sym] = store_path
    ssl_options["#{kind}_type".to_sym] = params["ssl_#{kind}_type"] if params.include?("ssl_#{kind}_type")
    ssl_options["#{kind}_password".to_sym] = params["ssl_#{kind}_password"].value if params.include?("ssl_#{kind}_password")
  end
end