require_relative "../../../spec/es_spec_helper" shared_examples_for 'an ILM enabled Logstash' do context 'with a policy with a maximum number of documents' do let (:policy) { small_max_doc_policy } let (:ilm_policy_name) { "logstash-policy-custom"} let (:settings) { super().merge("ilm_policy" => ilm_policy_name)} it 'should rollover when the policy max docs is reached' do put_policy(@es, ilm_policy_name, policy) subject.register subject.multi_receive([ LogStash::Event.new("message" => "sample message here"), LogStash::Event.new("somemessage" => { "message" => "sample nested message here" }), LogStash::Event.new("somevalue" => 100), ]) sleep(6) subject.multi_receive([ LogStash::Event.new("country" => "us"), LogStash::Event.new("country" => "at"), LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0 ] }) ]) sleep(6) subject.multi_receive([ LogStash::Event.new("country" => "uk"), LogStash::Event.new("country" => "fr"), LogStash::Event.new("geoip" => { "location" => [ 0.1, 1.0 ] }) ]) @es.indices.refresh # Wait or fail until everything's indexed. Stud::try(20.times) do r = @es.search(index: "#{expected_index}-*") expect(r).to have_hits(9) end indexes_written = @es.search(index: "#{expected_index}-*")['hits']['hits'].each_with_object(Hash.new(0)) do |x, res| index_written = x['_index'] res[index_written] += 1 end expect(indexes_written.count).to eq(3) expect(indexes_written["#{expected_index}-#{todays_date}-000001"]).to eq(3) expect(indexes_written["#{expected_index}-#{todays_date}-000002"]).to eq(3) expect(indexes_written["#{expected_index}-#{todays_date}-000003"]).to eq(3) end end context 'with a policy where the maximum number of documents is not reached' do let (:policy) { large_max_doc_policy } let (:ilm_policy_name) { "logstash-policy-custom-policy"} let (:settings) { super().merge("ilm_policy" => ilm_policy_name)} it 'should ingest into a single index when max docs is not reached' do put_policy(@es,ilm_policy_name, policy) subject.register subject.multi_receive([ LogStash::Event.new("message" => "sample message here"), LogStash::Event.new("somemessage" => { "message" => "sample nested message here" }), LogStash::Event.new("somevalue" => 100), ]) sleep(6) subject.multi_receive([ LogStash::Event.new("country" => "us"), LogStash::Event.new("country" => "at"), LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0 ] }) ]) @es.indices.refresh # Wait or fail until everything's indexed. Stud::try(20.times) do r = @es.search(index: "#{expected_index}-*") expect(r).to have_hits(6) end indexes_written = @es.search(index: "#{expected_index}-*")['hits']['hits'].each_with_object(Hash.new(0)) do |x, res| index_written = x['_index'] res[index_written] += 1 end expect(indexes_written.count).to eq(1) expect(indexes_written["#{expected_index}-#{todays_date}-000001"]).to eq(6) end end end shared_examples_for 'an ILM disabled Logstash' do it 'should not create a rollover alias' do expect(@es.indices.exists_alias(name: "logstash")).to be_falsey subject.register sleep(1) expect(@es.indices.exists_alias(name: "logstash")).to be_falsey end it 'should not install the default policy' do subject.register sleep(1) expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) end it 'should not write the ILM settings into the template' do subject.register sleep(1) template = get_template(@es, "logstash") expect(template).to have_index_pattern("logstash-*") if ESHelper.es_version_satisfies?(">= 2") expect(get_template_settings(template)['index']['lifecycle']).to be_nil end end context 'with an existing policy that will roll over' do let (:policy) { small_max_doc_policy } let (:ilm_policy_name) { "logstash-policy-3_docs"} let (:settings) { super().merge("ilm_policy" => ilm_policy_name)} it 'should not roll over indices' do subject.register subject.multi_receive([ LogStash::Event.new("message" => "sample message here"), LogStash::Event.new("somemessage" => { "message" => "sample nested message here" }), LogStash::Event.new("somevalue" => 100), ]) sleep(6) subject.multi_receive([ LogStash::Event.new("country" => "us"), LogStash::Event.new("country" => "at"), LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0 ] }) ]) @es.indices.refresh # Wait or fail until everything's indexed. Stud::try(20.times) do r = @es.search(index: 'logstash-*') expect(r).to have_hits(6) end indexes_written = @es.search(index: 'logstash-*')['hits']['hits'].each_with_object(Hash.new(0)) do |x, res| index_written = x['_index'] res[index_written] += 1 end expect(indexes_written.count).to eq(1) expect(indexes_written.values.first).to eq(6) end end context 'with a custom template name' do let (:template_name) { "logstash_custom_template_name" } let (:settings) { super().merge('template_name' => template_name)} it 'should not write the ILM settings into the template' do subject.register sleep(1) template = get_template(@es, template_name) expect(template).to have_index_pattern("logstash-*") if ESHelper.es_version_satisfies?(">= 2") expect(get_template_settings(template)['index']['lifecycle']).to be_nil end end end end shared_examples_for 'an Elasticsearch instance that does not support index lifecycle management' do require "logstash/outputs/elasticsearch" let (:ilm_enabled) { false } let (:settings) { { "hosts" => "#{get_host_port()}" } } before :each do require "elasticsearch" # Clean ES of data before we start. @es = get_client clean(@es) end after :each do clean(@es) end subject { LogStash::Outputs::ElasticSearch.new(settings) } context 'when ilm is enabled in Logstash' do let (:settings) { super().merge!({ 'ilm_enabled' => true }) } it 'should raise a configuration error' do expect do begin subject.register sleep(1) ensure subject.stop_template_installer end end.to raise_error(LogStash::ConfigurationError) end end context 'when ilm is disabled in Logstash' do let (:settings) { super().merge!({ 'ilm_enabled' => false }) } it_behaves_like 'an ILM disabled Logstash' end context 'when ilm is set to auto in Logstash' do let (:settings) { super().merge!({ 'ilm_enabled' => 'auto' }) } it_behaves_like 'an ILM disabled Logstash' end context 'when ilm is not set in Logstash' do it_behaves_like 'an ILM disabled Logstash' end end if ESHelper.es_version_satisfies?("<= 6.5") describe 'Pre-ILM versions of Elasticsearch', :integration => true do it_behaves_like 'an Elasticsearch instance that does not support index lifecycle management' end end if ESHelper.es_version_satisfies?(">= 6.6") describe 'OSS Elasticsearch', :distribution => 'oss', :integration => true do it_behaves_like 'an Elasticsearch instance that does not support index lifecycle management' end describe 'Elasticsearch has index lifecycle management enabled', :distribution => 'xpack', :integration => true do DEFAULT_INTERVAL = '600s' let (:ilm_enabled) { true } let (:settings) { { "ilm_enabled" => ilm_enabled, "hosts" => "#{get_host_port()}" } } let (:small_max_doc_policy) { max_docs_policy(3) } let (:large_max_doc_policy) { max_docs_policy(1000000) } let (:expected_index) { elasticsearch_output_plugin.default_ilm_rollover_alias } subject(:elasticsearch_output_plugin) { LogStash::Outputs::ElasticSearch.new(settings) } before :each do # Delete all templates first. require "elasticsearch" # Clean ES of data before we start. @es = get_client clean(@es) # Set the poll interval for lifecycle management to be short so changes get picked up in time. set_cluster_settings(@es, { "persistent" => { "indices.lifecycle.poll_interval" => "1s" } }) end after :each do # Set poll interval back to default set_cluster_settings(@es, { "persistent" => { "indices.lifecycle.poll_interval" => DEFAULT_INTERVAL } }) clean(@es) end context 'with ilm enabled' do let (:ilm_enabled) { true } context 'when using the default policy' do context 'with a custom pattern' do let (:settings) { super().merge("ilm_pattern" => "000001")} it 'should create a rollover alias' do expect(@es.indices.exists_alias(name: "logstash")).to be_falsey subject.register sleep(1) expect(@es.indices.exists_alias(name: "logstash")).to be_truthy expect(@es.get_alias(name: "logstash")).to include("logstash-000001") end end it 'should install it if it is not present' do expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) subject.register sleep(1) expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.not_to raise_error end it 'should create the default rollover alias' do expect(@es.indices.exists_alias(name: "logstash")).to be_falsey subject.register sleep(1) expect(@es.indices.exists_alias(name: "logstash")).to be_truthy expect(@es.get_alias(name: "logstash")).to include("logstash-#{todays_date}-000001") end it 'should ingest into a single index' do subject.register subject.multi_receive([ LogStash::Event.new("message" => "sample message here"), LogStash::Event.new("somemessage" => { "message" => "sample nested message here" }), LogStash::Event.new("somevalue" => 100), ]) sleep(6) subject.multi_receive([ LogStash::Event.new("country" => "us"), LogStash::Event.new("country" => "at"), LogStash::Event.new("geoip" => { "location" => [ 0.0, 0.0 ] }) ]) @es.indices.refresh # Wait or fail until everything's indexed. Stud::try(20.times) do r = @es.search(index: "logstash-*") expect(r).to have_hits(6) end indexes_written = @es.search(index: "logstash-*")['hits']['hits'].each_with_object(Hash.new(0)) do |x, res| index_written = x['_index'] res[index_written] += 1 end expect(indexes_written.count).to eq(1) expect(indexes_written["logstash-#{todays_date}-000001"]).to eq(6) end end context 'when not using the default policy' do let (:ilm_policy_name) {"logstash-policy-small"} let (:settings) { super().merge("ilm_policy" => ilm_policy_name)} let (:policy) { small_max_doc_policy } before do expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) put_policy(@es,ilm_policy_name, policy) end it 'should not install the default policy if it is not used' do subject.register sleep(1) expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) end end context 'when using a time based policy' do let (:ilm_policy_name) {"logstash-policy-time"} let (:settings) { super().merge("ilm_policy" => ilm_policy_name)} let (:policy) { max_age_policy("1d") } before do expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) put_policy(@es,ilm_policy_name, policy) end it 'should not install the default policy if it is not used' do subject.register sleep(1) expect{get_policy(@es, LogStash::Outputs::ElasticSearch::DEFAULT_POLICY)}.to raise_error(Elasticsearch::Transport::Transport::Errors::NotFound) end end context 'with the default template' do it 'should create the rollover alias' do expect(@es.indices.exists_alias(name: expected_index)).to be_falsey subject.register sleep(1) expect(@es.indices.exists_alias(name: expected_index)).to be_truthy expect(@es.get_alias(name: expected_index)).to include("#{expected_index}-#{todays_date}-000001") end it 'should write the ILM settings into the template' do subject.register sleep(1) template = get_template(@es, "logstash") expect(template).to have_index_pattern("logstash-*") expect(get_template_settings(template)['index']['lifecycle']['name']).to eq("logstash-policy") expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq("logstash") end it_behaves_like 'an ILM enabled Logstash' end context 'with a set index and a custom index pattern' do if ESHelper.es_version_satisfies?(">= 8.0") let (:template) { "spec/fixtures/template-with-policy-es8x.json" } elsif ESHelper.es_version_satisfies?(">= 7.0") let (:template) { "spec/fixtures/template-with-policy-es7x.json" } else let (:template) { "spec/fixtures/template-with-policy-es6x.json" } end let (:settings) { super().merge("template" => template, "index" => "overwrite-4")} it 'should not overwrite the index patterns' do subject.register sleep(1) template = get_template(@es, "logstash") expect(template).to have_index_pattern("overwrite-*") end end context 'with a custom template' do let (:ilm_rollover_alias) { "logstash_the_cat_in_the_hat" } let (:index) { ilm_rollover_alias } let(:expected_index) { index } let (:settings) { super().merge("ilm_policy" => ilm_policy_name, "template" => template, "ilm_rollover_alias" => ilm_rollover_alias)} if ESHelper.es_version_satisfies?(">= 8.0") let (:template) { "spec/fixtures/template-with-policy-es8x.json" } elsif ESHelper.es_version_satisfies?(">= 7.0") let (:template) { "spec/fixtures/template-with-policy-es7x.json" } else let (:template) { "spec/fixtures/template-with-policy-es6x.json" } end let (:ilm_enabled) { true } let (:ilm_policy_name) { "logstash-policy-custom-policy" } let (:policy) { small_max_doc_policy } before :each do put_policy(@es,ilm_policy_name, policy) end it_behaves_like 'an ILM enabled Logstash' it 'should create the rollover alias' do expect(@es.indices.exists_alias(name: ilm_rollover_alias)).to be_falsey subject.register sleep(1) expect(@es.indices.exists_alias(name: ilm_rollover_alias)).to be_truthy expect(@es.get_alias(name: ilm_rollover_alias)).to include("#{ilm_rollover_alias}-#{todays_date}-000001") end context 'when the custom rollover alias already exists' do it 'should ignore the already exists error' do expect(@es.indices.exists_alias(name: ilm_rollover_alias)).to be_falsey put_alias(@es, "#{ilm_rollover_alias}-#{todays_date}-000001", ilm_rollover_alias) expect(@es.indices.exists_alias(name: ilm_rollover_alias)).to be_truthy subject.register sleep(1) expect(@es.get_alias(name: ilm_rollover_alias)).to include("#{ilm_rollover_alias}-#{todays_date}-000001") end end it 'should write the ILM settings into the template' do subject.register sleep(1) template = get_template(@es, ilm_rollover_alias) expect(template).to have_index_pattern("#{ilm_rollover_alias}-*") expect(get_template_settings(template)['index']['lifecycle']['name']).to eq(ilm_policy_name) expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias) end context 'with a different template_name' do let (:template_name) { "logstash_custom_template_name" } let (:settings) { super().merge('template_name' => template_name)} it_behaves_like 'an ILM enabled Logstash' it 'should write the ILM settings into the template' do subject.register sleep(1) template = get_template(@es, template_name) expect(template).to have_index_pattern("#{ilm_rollover_alias}-*") expect(get_template_settings(template)['index']['lifecycle']['name']).to eq(ilm_policy_name) expect(get_template_settings(template)['index']['lifecycle']['rollover_alias']).to eq(ilm_rollover_alias) end end end end context 'when ilm_enabled is set to "auto"' do let (:ilm_enabled) { 'auto' } if ESHelper.es_version_satisfies?(">=7.0") context 'when Elasticsearch is version 7 or above' do it_behaves_like 'an ILM enabled Logstash' end end if ESHelper.es_version_satisfies?('< 7.0') context 'when Elasticsearch is version 7 or below' do it_behaves_like 'an ILM disabled Logstash' end end end context 'when ilm_enabled is the default' do let (:settings) { super().tap{|x|x.delete('ilm_enabled')}} if ESHelper.es_version_satisfies?(">=7.0") context 'when Elasticsearch is version 7 or above' do it_behaves_like 'an ILM enabled Logstash' end end if ESHelper.es_version_satisfies?('< 7.0') context 'when Elasticsearch is version 7 or below' do it_behaves_like 'an ILM disabled Logstash' end end end context 'with ilm disabled' do let (:settings) { super().merge('ilm_enabled' => false )} it_behaves_like 'an ILM disabled Logstash' end context 'with ilm disabled using a string' do let (:settings) { super().merge('ilm_enabled' => 'false' )} it_behaves_like 'an ILM disabled Logstash' end end end