Module: LogStash::Outputs::OpenSearch::HttpClientBuilder

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

Class Method Summary collapse

Class Method Details

.build(logger, hosts, params) ⇒ Object

Raises:

  • (LogStash::ConfigurationError) 


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
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/logstash/outputs/opensearch/http_client_builder.rb', line 15

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"],
    :http_compression => params["http_compression"],
    :headers => params["custom_headers"] || {},
    :legacy_template => params["legacy_template"]
  }
  
  client_settings[:proxy] = params["proxy"] if params["proxy"]
  
  common_options = {
    :client_settings => client_settings,
    :metric => params["metric"],
    :resurrect_delay => params["resurrect_delay"],
    :default_server_major_version => params["default_server_major_version"]
  }

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

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

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

  common_options[:bulk_path] = if params["bulk_path"]
     dedup_slashes("/#{params["bulk_path"]}")
  else
     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)

  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."
  ) 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,
                                          :auth_type => params["auth_type"]
                                          ))
end

.create_http_client(options) ⇒ Object 



118
119
120
 
# File 'lib/logstash/outputs/opensearch/http_client_builder.rb', line 118

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

.setup_basic_auth(logger, params) ⇒ Object 



171
172
173
174
175
176
177
178
179
180
 
# File 'lib/logstash/outputs/opensearch/http_client_builder.rb', line 171

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 



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
166
167
168
169
 
# File 'lib/logstash/outputs/opensearch/http_client_builder.rb', line 122

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

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

  cacert, truststore, truststore_password, keystore, keystore_password, tls_client_cert, tls_client_key =
    params.values_at('cacert', 'truststore', 'truststore_password', 'keystore', 'keystore_password', 'tls_certificate', 'tls_key')

  if cacert && truststore
    raise(LogStash::ConfigurationError, "Use either \"cacert\" or \"truststore\" when configuring the CA certificate") if truststore
  end

  if (tls_client_cert && !tls_client_key)
    raise(LogStash::ConfigurationError, "\"tls_key\" is missing")
  end

  if (!tls_client_cert && tls_client_key)
    raise(LogStash::ConfigurationError, "\"tls_certificate\" is missing")
  end

  ssl_options = {:enabled => true}

  if cacert
    ssl_options[:ca_file] = cacert
  elsif truststore
    ssl_options[:truststore_password] = truststore_password.value if truststore_password
  end

  ssl_options[:truststore] = truststore if truststore
  if keystore
    ssl_options[:keystore] = keystore
    ssl_options[:keystore_password] = keystore_password.value if keystore_password
  end
  if (tls_client_cert && tls_client_key)
    ssl_options[:client_cert] = tls_client_cert
    ssl_options[:client_key] = tls_client_key
  end
  if !params["ssl_certificate_verification"]
    logger.warn [
                   "** WARNING ** Detected UNSAFE options in opensearch output configuration!",
                   "** WARNING ** You have enabled encryption but DISABLED certificate verification.",
                   "** WARNING ** To make sure your data is secure change :ssl_certificate_verification to true"
                 ].join("\n")
    ssl_options[:verify] = false
  end
  { ssl: ssl_options }
end