github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 14:47:19 -04:00
Code Issues Projects Releases Packages Wiki Activity

replace direct access of hidden indices with system indices api (#12279)

Browse source 
* replace direct hidden indices access with system indices api

* fulfill backward compatibility

* fix log msg, rename class, simplify response handling

* modularise fetcher
This commit is contained in:
kaisecheng 2020-10-06 21:19:36 +02:00 • committed by Kaise Cheng
parent c606e9f5b1
commit a93d454688
5 changed files with 479 additions and 256 deletions
Download patch file Download diff file Expand all files Collapse all files

120
x-pack/lib/config_management/elasticsearch_source.rb
View file

@ -18,7 +18,6 @@ module LogStash
class RemoteConfigError < LogStash::Error; end
PIPELINE_INDEX = ".logstash"
# exclude basic
VALID_LICENSES = %w(trial standard gold platinum enterprise)
FEATURE_INTERNAL = 'management'
@ -50,6 +49,21 @@ module LogStash
false
end
# decide using system indices api (7.10+) or legacy api (< 7.10) base on elasticsearch server version
def get_pipeline_fetcher
response = client.get("/")
if response["error"]
raise RemoteConfigError, "Cannot find elasticsearch version, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
end
logger.debug("Reading configuration from Elasticsearch version {}", response["version"]["number"])
version_number = response["version"]["number"].split(".")
first = version_number[0].to_i
second = version_number[1].to_i
(first >= 8 || (first == 7 && second >= 10))? SystemIndicesFetcher.new: LegacyHiddenIndicesFetcher.new
end
def pipeline_configs
logger.trace("Fetch remote config pipeline", :pipeline_ids => pipeline_ids)
@ -63,33 +77,21 @@ module LogStash
end
end
response = fetch_config(pipeline_ids)
fetcher = get_pipeline_fetcher
fetcher.fetch_config(pipeline_ids, client)
if response["error"]
raise RemoteConfigError, "Cannot find find configuration for pipeline_id: #{pipeline_ids}, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
end
if response["docs"].nil?
logger.debug("Server returned an unknown or malformed document structure", :response => response)
raise RemoteConfigError, "Elasticsearch returned an unknown or malformed document structure"
end
# Cache pipelines to handle the case where a remote configuration error can render a pipeline unusable
# it is not reloadable
@cached_pipelines = response["docs"].collect do |document|
get_pipeline(document)
@cached_pipelines = pipeline_ids.collect do |pid|
get_pipeline(pid, fetcher)
end.compact
end
def get_pipeline(response)
pipeline_id = response["_id"]
if response["found"] == false
def get_pipeline(pipeline_id, fetcher)
unless fetcher.config_exist?(pipeline_id)
logger.debug("Could not find a remote configuration for a specific `pipeline_id`", :pipeline_id => pipeline_id)
return nil
end
config_string = response.fetch("_source", {})["pipeline"]
config_string = fetcher.get_single_pipeline_setting(pipeline_id)["pipeline"]
raise RemoteConfigError, "Empty configuration for pipeline_id: #{pipeline_id}" if config_string.nil? || config_string.empty?
@ -100,7 +102,7 @@ module LogStash
settings.set("pipeline.id", pipeline_id)
# override global settings with pipeline settings from ES, if any
pipeline_settings = response["_source"]["pipeline_settings"]
pipeline_settings = fetcher.get_single_pipeline_setting(pipeline_id)["pipeline_settings"]
unless pipeline_settings.nil?
pipeline_settings.each do |setting, value|
if SUPPORTED_PIPELINE_SETTINGS.include? setting
@ -127,15 +129,6 @@ module LogStash
es.build_client
end
def fetch_config(pipeline_ids)
request_body_string = LogStash::Json.dump({ "docs" => pipeline_ids.collect { |pipeline_id| { "_id" => pipeline_id } } })
client.post(config_path, {}, request_body_string)
end
def config_path
"#{PIPELINE_INDEX}/_mget"
end
def populate_license_state(xpack_info)
if xpack_info.failed?
{
@ -193,5 +186,72 @@ module LogStash
@client ||= build_client
end
end
module Fetcher
def config_exist?(pipeline_id)
@response.has_key?(pipeline_id)
end
def fetch_config(pipeline_ids, client) end
def get_single_pipeline_setting(pipeline_id) end
end
class SystemIndicesFetcher
include LogStash::Util::Loggable, Fetcher
SYSTEM_INDICES_API_PATH = "_logstash/pipeline"
def fetch_config(pipeline_ids, client)
path_ids = pipeline_ids.join(",")
response = client.get("#{SYSTEM_INDICES_API_PATH}/#{path_ids}")
if response["error"]
raise ElasticsearchSource::RemoteConfigError, "Cannot find find configuration for pipeline_id: #{pipeline_ids}, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
end
@response = response
end
def get_single_pipeline_setting(pipeline_id)
@response.fetch(pipeline_id, {})
end
end
# clean up LegacyHiddenIndicesFetcher https://github.com/elastic/logstash/issues/12291
class LegacyHiddenIndicesFetcher
include LogStash::Util::Loggable, Fetcher
PIPELINE_INDEX = ".logstash"
def fetch_config(pipeline_ids, client)
request_body_string = LogStash::Json.dump({ "docs" => pipeline_ids.collect { |pipeline_id| { "_id" => pipeline_id } } })
response = client.post("#{PIPELINE_INDEX}/_mget", {}, request_body_string)
if response["error"]
raise ElasticsearchSource::RemoteConfigError, "Cannot find find configuration for pipeline_id: #{pipeline_ids}, server returned status: `#{response["status"]}`, message: `#{response["error"]}`"
end
if response["docs"].nil?
logger.debug("Server returned an unknown or malformed document structure", :response => response)
raise ElasticsearchSource::RemoteConfigError, "Elasticsearch returned an unknown or malformed document structure"
end
@response = format_response(response)
end
def get_single_pipeline_setting(pipeline_id)
@response.fetch(pipeline_id, {}).fetch("_source", {})
end
private
# transform legacy response to be similar to system indices response
def format_response(response)
response["docs"].map { |pipeline|
{pipeline["_id"] => pipeline} if pipeline.fetch("found", false)
}.compact
.reduce({}, :merge)
end
end
end
end

2
x-pack/qa/integration/management/multiple_pipelines_spec.rb
View file

@ -17,7 +17,7 @@ describe "Read configuration from elasticsearch" do
"hello" => nil
}
cleanup_elasticsearch(".logstash*")
cleanup_system_indices(@pipelines.keys)
cleanup_elasticsearch(".monitoring-logstash*")
@pipelines.each do |pipeline_id, config|

2
x-pack/qa/integration/management/read_configuration_spec.rb
View file

@ -31,7 +31,7 @@ describe "Read configuration from elasticsearch" do
def start_services(elasticsearch_options, logstash_options)
@elasticsearch_service = elasticsearch(elasticsearch_options)
cleanup_elasticsearch(".logstash*")
cleanup_system_indices([PIPELINE_ID])
config = "input { generator { count => 100 } tcp { port => 6000 } } output { null {} }"
push_elasticsearch_config(PIPELINE_ID, config)

27
x-pack/qa/integration/support/helpers.rb
View file

@ -8,6 +8,7 @@ require "elasticsearch"
require "fileutils"
require "stud/try"
require "open3"
require "time"
VERSIONS_YML_PATH = File.join(File.dirname(__FILE__), "..", "..", "..", "..", "versions.yml")
VERSION_PATH = File.join(File.dirname(__FILE__), "..", "..", "..", "VERSION")
@ -94,8 +95,19 @@ def elasticsearch_client(options = { :url => "http://elastic:#{elastic_password}
Elasticsearch::Client.new(options)
end
def es_version
response = elasticsearch_client.perform_request(:get, "")
response.body["version"]["number"].gsub(/(\d+\.\d+)\..+/, '\1').to_f
end
def push_elasticsearch_config(pipeline_id, config)
elasticsearch_client.index :index => '.logstash', :type => "_doc", id: pipeline_id, :body => { :pipeline => config }
if es_version >= 7.10
elasticsearch_client.perform_request(:put, "_logstash/pipeline/#{pipeline_id}", {},
{ :pipeline => config, :username => "log.stash", :pipeline_metadata => {:version => "1" },
:pipeline_settings => {"pipeline.batch.delay": "50"}, :last_modified => Time.now.utc.iso8601})
else
elasticsearch_client.index :index => '.logstash', :type => "_doc", id: pipeline_id, :body => { :pipeline => config }
end
end
def cleanup_elasticsearch(index = MONITORING_INDEXES)
@ -103,6 +115,17 @@ def cleanup_elasticsearch(index = MONITORING_INDEXES)
elasticsearch_client.indices.refresh
end
def cleanup_system_indices(pipeline_ids)
pipeline_ids.each do |id|
begin
elasticsearch_client.perform_request(:delete, "_logstash/pipeline/#{id}")
rescue Elasticsearch::Transport::Transport::Errors::NotFound => e
puts ".logstash can be empty #{e.message}"
end
end
elasticsearch_client.indices.refresh
end
def logstash_command_append(cmd, argument, value)
if cmd !~ /#{Regexp.escape(argument)}/
cmd << " #{argument} #{value}"
@ -136,4 +159,4 @@ def verify_response!(cmd, response)
unless response.successful?
raise "Something went wrong when installing xpack,\ncmd: #{cmd}\nresponse: #{response}"
end
end
end

584
x-pack/spec/config_management/elasticsearch_source_spec.rb
View file

@ -12,6 +12,8 @@ require "monitoring/monitoring"
require "stud/temporary"
describe LogStash::ConfigManagement::ElasticsearchSource do
let(:system_indices_api) { LogStash::ConfigManagement::SystemIndicesFetcher::SYSTEM_INDICES_API_PATH }
let(:system_indices_url_regex) { Regexp.new("^#{system_indices_api}") }
let(:elasticsearch_url) { ["https://localhost:9898"] }
let(:elasticsearch_username) { "elastictest" }
let(:elasticsearch_password) { "testchangeme" }
@ -41,7 +43,6 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
}"
}
let(:valid_xpack_response) {
{
"license" => {
@ -83,7 +84,6 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
}")
}
let(:settings) do
{
"xpack.management.enabled" => true,
@ -94,6 +94,38 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
}
end
let(:es_version_response) { es_version_8_response }
let(:es_version_8_response) { generate_es_version_response("8.0.0-SNAPSHOT") }
let(:es_version_7_9_response) { generate_es_version_response("7.9.1") }
let(:elasticsearch_7_9_err_response) {
{"error"=>
{"root_cause"=>
[{"type"=>"parse_exception",
"reason"=>"request body or source parameter is required"}],
"type"=>"parse_exception",
"reason"=>"request body or source parameter is required"},
"status"=>400}
}
let(:elasticsearch_8_err_response) {
{"error"=>
{"root_cause"=>
[{"type"=>"index_not_found_exception",
"reason"=>"no such index [.logstash]",
"resource.type"=>"index_expression",
"resource.id"=>".logstash",
"index_uuid"=>"_na_",
"index"=>".logstash"}],
"type"=>"index_not_found_exception",
"reason"=>"no such index [.logstash]",
"resource.type"=>"index_expression",
"resource.id"=>".logstash",
"index_uuid"=>"_na_",
"index"=>".logstash"},
"status"=>404}
}
before do
extension.additionals_settings(system_settings)
apply_settings(settings, system_settings)
@ -162,22 +194,85 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
end
end
describe "#config_path" do
before do
# we are testing the arguments here, not the behavior of the elasticsearch output
allow_any_instance_of(described_class).to receive(:build_client).and_return(nil)
end
describe LogStash::ConfigManagement::SystemIndicesFetcher do
subject { described_class.new }
let(:pipeline_id) { "foobar" }
let(:settings) do
{
"xpack.management.pipeline.id" => pipeline_id,
"xpack.management.elasticsearch.password" => "testpassword"
describe "system indices api" do
let(:mock_client) { double("http_client") }
let(:config) { "input { generator { count => 100 } tcp { port => 6005 } } output { }}" }
let(:pipeline_id) { "super_generator" }
let(:elasticsearch_response) { {"#{pipeline_id}"=> {"pipeline"=> "#{config}"}} }
it "#fetch_config" do
expect(mock_client).to receive(:get).with("#{described_class::SYSTEM_INDICES_API_PATH}/#{pipeline_id}").and_return(elasticsearch_response)
expect(subject.fetch_config([pipeline_id], mock_client)).to eq(elasticsearch_response)
expect(subject.get_single_pipeline_setting(pipeline_id)).to eq({"pipeline"=> "#{config}"})
end
it "#fetch_config should raise error" do
expect(mock_client).to receive(:get).with("#{described_class::SYSTEM_INDICES_API_PATH}/apache,nginx").and_return(elasticsearch_8_err_response)
expect{ subject.fetch_config(["apache", "nginx"], mock_client) }.to raise_error(LogStash::ConfigManagement::ElasticsearchSource::RemoteConfigError)
end
end
end
describe LogStash::ConfigManagement::LegacyHiddenIndicesFetcher do
subject { described_class.new }
describe "legacy api" do
let(:mock_client) { double("http_client") }
let(:config) { "input { generator { count => 100 } tcp { port => 6005 } } output { }}" }
let(:another_config) { "input { generator { count => 100 } tcp { port => 6006 } } output { }}" }
let(:pipeline_id) { "super_generator" }
let(:another_pipeline_id) { "another_generator" }
let(:elasticsearch_response) {
{"docs"=>
[{"_index"=>".logstash",
"_id"=>"#{pipeline_id}",
"_version"=>2,
"_seq_no"=>2,
"_primary_term"=>1,
"found"=>true,
"_source"=>
{"pipeline"=> "#{config}"}},
{"_index"=>".logstash",
"_id"=>"#{another_pipeline_id}",
"_version"=>2,
"_seq_no"=>3,
"_primary_term"=>1,
"found"=>true,
"_source"=>
{"pipeline"=> "#{another_config}"}},
{"_index"=>".logstash", "_id"=>"not_exists", "found"=>false}]}
}
end
it "generates the path to get the configuration" do
expect(subject.config_path).to eq("#{described_class::PIPELINE_INDEX}/_mget")
let(:formatted_es_response) {
{"super_generator"=>{"_index"=>".logstash", "_id"=>"super_generator", "_version"=>2, "_seq_no"=>2, "_primary_term"=>1, "found"=>true, "_source"=>{"pipeline"=>"input { generator { count => 100 } tcp { port => 6005 } } output { }}"}}}
}
it "#fetch_config" do
expect(mock_client).to receive(:post).with("#{described_class::PIPELINE_INDEX}/_mget", {}, "{\"docs\":[{\"_id\":\"#{pipeline_id}\"},{\"_id\":\"#{another_pipeline_id}\"}]}").and_return(elasticsearch_response)
expect(subject.fetch_config([pipeline_id, another_pipeline_id], mock_client).size).to eq(2)
expect(subject.get_single_pipeline_setting(pipeline_id)).to eq({"pipeline" => "#{config}"})
expect(subject.get_single_pipeline_setting(another_pipeline_id)).to eq({"pipeline" => "#{another_config}"})
end
it "#fetch_config should raise error" do
expect(mock_client).to receive(:post).with("#{described_class::PIPELINE_INDEX}/_mget", {}, "{\"docs\":[{\"_id\":\"#{pipeline_id}\"},{\"_id\":\"#{another_pipeline_id}\"}]}").and_return(elasticsearch_7_9_err_response)
expect{ subject.fetch_config([pipeline_id, another_pipeline_id], mock_client) }.to raise_error(LogStash::ConfigManagement::ElasticsearchSource::RemoteConfigError)
end
it "#fetch_config should raise error when response is empty" do
expect(mock_client).to receive(:post).with("#{described_class::PIPELINE_INDEX}/_mget", {}, "{\"docs\":[{\"_id\":\"#{pipeline_id}\"},{\"_id\":\"#{another_pipeline_id}\"}]}").and_return(LogStash::Json.load("{}"))
expect{ subject.fetch_config([pipeline_id, another_pipeline_id], mock_client) }.to raise_error(LogStash::ConfigManagement::ElasticsearchSource::RemoteConfigError)
end
it "#format_response should return pipelines" do
result = subject.send(:format_response, elasticsearch_response)
expect(result.size).to eq(2)
expect(result.has_key?(pipeline_id)).to be_truthy
expect(result.has_key?(another_pipeline_id)).to be_truthy
end
end
end
@ -216,267 +311,312 @@ describe LogStash::ConfigManagement::ElasticsearchSource do
let(:pipeline_id) { "apache" }
let(:mock_client) { double("http_client") }
let(:settings) { super.merge({ "xpack.management.pipeline.id" => pipeline_id }) }
let(:es_path) { "#{described_class::PIPELINE_INDEX}/_mget" }
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { elasticsearch_8_response }
let(:elasticsearch_8_response) {
"{\"#{pipeline_id}\":{
\"username\":\"log.stash\",
\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",
\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\"},
\"pipeline\":\"#{config}\",
\"pipeline_settings\":{\"pipeline.batch.delay\":\"50\", \"pipeline.workers\":\"99\", \"pipeline.output.workers\":\"99\", \"nonsensical.invalid.setting\":\"-9999\"}}}" }
let(:elasticsearch_7_9_response) {
"{ \"docs\":[{
\"_index\":\".logstash\",
\"_type\":\"pipelines\",
\"_id\":\"#{pipeline_id}\",
\"_version\":8,
\"found\":true,
\"_source\":{
\"id\":\"apache\",
\"description\":\"Process apache logs\",
\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",
\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},
\"pipeline\":\"#{config}\",
\"pipeline_settings\":{\"pipeline.workers\":\"99\", \"pipeline.output.workers\":\"99\", \"nonsensical.invalid.setting\":\"-9999\"}}}]}" }
let(:es_path) { ".logstash/_mget" }
let(:request_body_string) { LogStash::Json.dump({ "docs" => [{ "_id" => pipeline_id }] }) }
before do
allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_response))
allow(mock_client).to receive(:get).with(system_indices_url_regex).and_return(LogStash::Json.load(elasticsearch_response))
allow(mock_client).to receive(:get).with("/").and_return(es_version_response)
allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
allow(mock_license_client).to receive(:get).with('_xpack').and_return(valid_xpack_response)
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:client).and_return(mock_license_client)
end
context "with one `pipeline_id` configured" do
context "when successfully fetching a remote configuration" do
describe "system indices [8] and legacy api [7.9]" do
[8, 7.9].each { |es_version|
let(:elasticsearch_response) { (es_version >= 8)? elasticsearch_8_response: elasticsearch_7_9_response }
before :each do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
allow(mock_client).to receive(:get).with("/").and_return(es_version >= 8? es_version_response: es_version_7_9_response)
end
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:whitelisted_pipeline_setting_name) {"pipeline.workers"}
let(:whitelisted_pipeline_setting_value) {"99"}
let(:non_whitelisted_pipeline_setting_name) {"pipeline.output.workers"}
let(:non_whitelisted_pipeline_setting_value) {"99"}
let(:invalid_pipeline_setting) {"nonsensical.invalid.setting"}
let(:elasticsearch_response) {
%{
{ "docs": [{
"_index":".logstash",
"_type":"pipelines",
"_id":"#{pipeline_id}",
"_version":8,
"found":true,
"_source": {
"id":"apache",
"description":"Process apache logs",
"modified_timestamp":"2017-02-28T23:02:17.023Z",
"pipeline_metadata":{
"version":5,
"type":"logstash_pipeline",
"username":"elastic"
},
"pipeline":"#{config}",
"pipeline_settings": {
"#{whitelisted_pipeline_setting_name}":#{whitelisted_pipeline_setting_value},
"#{non_whitelisted_pipeline_setting_name}":#{non_whitelisted_pipeline_setting_value},
"#{invalid_pipeline_setting}":-9999
}
context "with one `pipeline_id` configured [#{es_version}]" do
context "when successfully fetching a remote configuration" do
before :each do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_7_9_response))
end
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
end
it "ignores non-whitelisted and invalid settings" do
pipeline_config = subject.pipeline_configs
settings_hash = pipeline_config[0].settings.to_hash
expect(settings_hash["pipeline.workers"]).to eq(99)
expect(settings_hash["pipeline.output.workers"]).not_to eq(99)
expect(settings_hash["nonsensical.invalid.setting"]).to be_falsey
end
end
context "when the license has expired [#{es_version}]" do
let(:license_status) { 'expired'}
let(:license_expiry_date) { Time.now - (60 * 60 * 24)}
before :each do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
end
end
context "when the license server is not available [#{es_version}]" do
before :each do
allow(mock_license_client).to receive(:get).with('_xpack').and_raise("An error is here")
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
context "when the xpack is not installed [#{es_version}]" do
before :each do
expect(mock_license_client).to receive(:get).with('_xpack').and_return(no_xpack_response)
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
describe "security enabled/disabled in Elasticsearch [#{es_version}]" do
let(:xpack_response) do
{
"license"=> {
"status"=> license_status,
"uid"=> "9a48c67c-ce2c-4169-97bf-37d324b8ab80",
"type"=> license_type,
"expiry_date_in_millis"=> license_expiry_in_millis
},
"features" => {
"security" => {
"description" => "Security for the Elastic Stack",
"available" => true,
"enabled" => security_enabled
}
}
}
}]
}
}
}
end
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
before :each do
allow(mock_license_client).to receive(:get).with('_xpack').and_return(xpack_response)
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
end
context "when security is disabled in Elasticsearch [#{es_version}]" do
let(:security_enabled) { false }
it 'should raise an error' do
expect { subject.pipeline_configs }.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
it "ignores non-whitelisted and invalid settings" do
pipeline_config = subject.pipeline_configs
settings_hash = pipeline_config[0].settings.to_hash
context "when security is enabled in Elasticsearch [#{es_version}]" do
let(:security_enabled) { true }
it 'should not raise an error' do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
expect { subject.pipeline_configs }.not_to raise_error
end
end
end
expect(settings_hash[whitelisted_pipeline_setting_name]).to eq(whitelisted_pipeline_setting_value.to_i)
expect(settings_hash[non_whitelisted_pipeline_setting_name]).not_to eq(non_whitelisted_pipeline_setting_value.to_i)
expect(settings_hash[invalid_pipeline_setting]).to be_falsey
end
end
context 'when the license has expired' do
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
let(:license_status) { 'expired'}
let(:license_expiry_date) { Time.now - (60 * 60 * 24)}
context "With an invalid basic license, it should raise an error [#{es_version}]" do
let(:license_type) { 'basic' }
before :each do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
# config management can be used with any license type except basic
(::LogStash::LicenseChecker::LICENSE_TYPES - ["basic"]).each do |license_type|
context "With a valid #{license_type} license, it should return a pipeline [#{es_version}]" do
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
end
end
let(:license_type) { license_type }
context 'when the license server is not available' do
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
before :each do
allow(mock_license_client).to receive(:get).with('_xpack').and_raise("An error is here")
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
context 'when the xpack is not installed' do
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
before :each do
expect(mock_license_client).to receive(:get).with('_xpack').and_return(no_xpack_response)
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
describe 'security enabled/disabled in Elasticsearch' do
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
let(:xpack_response) do
{
"license"=> {
"status"=> license_status,
"uid"=> "9a48c67c-ce2c-4169-97bf-37d324b8ab80",
"type"=> license_type,
"expiry_date_in_millis"=> license_expiry_in_millis
},
"features" => {
"security" => {
"description" => "Security for the Elastic Stack",
"available" => true,
"enabled" => security_enabled
}
}
}
end
before :each do
allow(mock_license_client).to receive(:get).with('_xpack').and_return(xpack_response)
allow_any_instance_of(LogStash::LicenseChecker::LicenseReader).to receive(:build_client).and_return(mock_license_client)
end
context 'when security is disabled in Elasticsearch' do
let(:security_enabled) { false }
it 'should raise an error' do
expect { subject.pipeline_configs }.to raise_error(LogStash::LicenseChecker::LicenseError)
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
end
end
end
end
context 'when security is enabled in Elasticsearch' do
let(:security_enabled) { true }
it 'should not raise an error' do
context "with multiples `pipeline_id` configured [#{es_version}]" do
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
expect { subject.pipeline_configs }.not_to raise_error
end
context "when successfully fetching multiple remote configuration" do
let(:pipelines) do
{
"apache" => config_apache,
"firewall" => config_firewall
}
end
let(:pipeline_id) { pipelines.keys }
let(:config_apache) { "input { generator { id => '123'} } filter { mutate {} } output { }" }
let(:config_firewall) { "input { generator { id => '321' } } filter { mutate {} } output { }" }
let(:elasticsearch_response) do
content = "{"
content << pipelines.collect do |pipeline_id, config|
"\"#{pipeline_id}\":{\"username\":\"log.stash\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\"},\"pipeline\":\"#{config}\",\"pipeline_settings\":{\"pipeline.batch.delay\":\"50\"}}"
end.join(",")
content << "}"
content
end
let(:elasticsearch_7_9_response) do
content = "{ \"docs\":["
content << pipelines.collect do |pipeline_id, config|
"{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}"
end.join(",")
content << "]}"
content
end
let(:request_body_string) { LogStash::Json.dump({ "docs" => pipeline_id.collect { |pipeline_id| { "_id" => pipeline_id } } }) }
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
expect(pipeline_config.collect(&:config_string)).to include(*pipelines.values)
expect(pipeline_config.map(&:pipeline_id).collect(&:to_sym)).to include(*pipelines.keys.collect(&:to_sym))
end
end
end
end
context "With an invalid basic license, it should raise an error" do
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
let(:license_type) { 'basic' }
it 'should raise an error' do
expect{subject.pipeline_configs}.to raise_error(LogStash::LicenseChecker::LicenseError)
end
end
# config management can be used with any license type except basic
(::LogStash::LicenseChecker::LICENSE_TYPES - ["basic"]).each do |license_type|
context "With a valid #{license_type} license, it should return a pipeline" do
context "when the configuration is not found [#{es_version}]" do
let(:elasticsearch_8_response) { "{}" }
let(:elasticsearch_7_9_response) { "{ \"docs\": [{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"donotexist\",\"found\":false}]}" }
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
let(:config) { "input { generator {} } filter { mutate {} } output { }" }
let(:elasticsearch_response) { "{ \"docs\":[{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}]}" }
let(:license_type) { license_type }
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
expect(pipeline_config.first.config_string).to match(config)
expect(pipeline_config.first.pipeline_id.to_sym).to eq(pipeline_id.to_sym)
it "returns no pipeline config" do
expect(subject.pipeline_configs).to be_empty
end
end
end
context "when any error returned from elasticsearch [#{es_version}]" do
let(:elasticsearch_8_response){"{\"error\" : \"no handler found for uri [/_logstash/pipelines?pretty] and method [GET]\"}" }
let(:elasticsearch_7_9_response) { '{ "error":{"root_cause":[{"type":"illegal_argument_exception","reason":"No endpoint or operation is available at [testing_ph]"}],"type":"illegal_argument_exception","reason":"No endpoint or operation is available at [testing_ph]"},"status":400}' }
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it "raises a `RemoteConfigError`" do
expect { subject.pipeline_configs }.to raise_error LogStash::ConfigManagement::ElasticsearchSource::RemoteConfigError
end
end
}
end
context "with multiples `pipeline_id` configured" do
before do
allow(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_return(LogStash::Json.load(elasticsearch_response))
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
context "when successfully fetching multiple remote configuration" do
let(:pipelines) do
{
"apache" => config_apache,
"firewall" => config_firewall
}
end
let(:config_apache) { "input { generator { id => '123'} } filter { mutate {} } output { }" }
let(:config_firewall) { "input { generator { id => '321' } } filter { mutate {} } output { }" }
let(:elasticsearch_response) do
content = "{ \"docs\":["
content << pipelines.collect do |pipeline_id, config|
"{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"#{pipeline_id}\",\"_version\":8,\"found\":true,\"_source\":{\"id\":\"apache\",\"description\":\"Process apache logs\",\"modified_timestamp\":\"2017-02-28T23:02:17.023Z\",\"pipeline_metadata\":{\"version\":5,\"type\":\"logstash_pipeline\",\"username\":\"elastic\"},\"pipeline\":\"#{config}\"}}"
end.join(",")
content << "]}"
content
end
it "returns a valid pipeline config" do
pipeline_config = subject.pipeline_configs
expect(pipeline_config.collect(&:config_string)).to include(*pipelines.values)
expect(pipeline_config.map(&:pipeline_id).collect(&:to_sym)).to include(*pipelines.keys.collect(&:to_sym))
end
end
end
context "when the configuration is not found" do
let(:elasticsearch_response) { "{ \"docs\": [{\"_index\":\".logstash\",\"_type\":\"pipelines\",\"_id\":\"donotexist\",\"found\":false}]}" }
describe "create pipeline fetcher by es version" do
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it "returns no pipeline config" do
expect(subject.pipeline_configs).to be_empty
it "should give SystemIndicesFetcher in [8]" do
allow(mock_client).to receive(:get).with("/").and_return(es_version_response)
expect(subject.get_pipeline_fetcher).to be_an_instance_of LogStash::ConfigManagement::SystemIndicesFetcher
end
it "should give SystemIndicesFetcher in [7.10]" do
allow(mock_client).to receive(:get).with("/").and_return(generate_es_version_response("7.10.0-SNAPSHOT"))
expect(subject.get_pipeline_fetcher).to be_an_instance_of LogStash::ConfigManagement::SystemIndicesFetcher
end
it "should give LegacyHiddenIndicesFetcher in [7.9]" do
allow(mock_client).to receive(:get).with("/").and_return(es_version_7_9_response)
expect(subject.get_pipeline_fetcher).to be_an_instance_of LogStash::ConfigManagement::LegacyHiddenIndicesFetcher
end
end
context "when any error returned from elasticsearch" do
let(:elasticsearch_response){'{ "error":{"root_cause":[{"type":"illegal_argument_exception","reason":"No endpoint or operation is available at [testing_ph]"}],"type":"illegal_argument_exception","reason":"No endpoint or operation is available at [testing_ph]"},"status":400}' }
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it "raises a `RemoteConfigError`" do
expect { subject.pipeline_configs }.to raise_error /illegal_argument_exception/
end
end
context "when exception occur" do
describe "when exception occur" do
let(:elasticsearch_response) { "" }
before do
expect_any_instance_of(described_class).to receive(:build_client).and_return(mock_client)
end
it "raises the exception upstream" do
it "raises the exception upstream in [8]" do
allow(mock_client).to receive(:get).with("/").and_return(es_version_response)
allow(mock_client).to receive(:get).with(system_indices_url_regex).and_raise("Something bad")
expect { subject.pipeline_configs }.to raise_error /Something bad/
end
it "raises the exception upstream in [7.9]" do
allow(mock_client).to receive(:get).with("/").and_return(es_version_7_9_response)
expect(mock_client).to receive(:post).with(es_path, {}, request_body_string).and_raise("Something bad")
expect { subject.pipeline_configs }.to raise_error /Something bad/
end
end
end
def generate_es_version_response(version)
{"name"=>"MacBook-Pro",
"cluster_name"=>"elasticsearch",
"cluster_uuid"=>"YgpKq8VkTJuGTSb9aidlIA",
"version"=>
{"number"=>"#{version}",
"build_flavor"=>"default",
"build_type"=>"tar",
"build_hash"=>"26eb422dc55236a1c5625e8a73e5d866e54610a2",
"build_date"=>"2020-09-24T09:37:06.459350Z",
"build_snapshot"=>true,
"lucene_version"=>"8.7.0",
"minimum_wire_compatibility_version"=>"7.10.0",
"minimum_index_compatibility_version"=>"7.0.0"},
"tagline"=>"You Know, for Search"}
end
end