Geoip database management cache invalidation (#16222) (#16223)

* geoip: failing specs demonstrating #16221

* geoip: invalidate cached db state when receiving updates/expiries

(cherry picked from commit 801f0f4)

Co-authored-by: Ry Biesemeyer <[email protected]>

x-pack/lib/filters/geoip/database_manager.rb

@@ -136,17 +136,23 @@ def database_path(database_type)
   end
 
   def update!(database_type, updated_db_info)
-    new_database_path = updated_db_info.path
-    notify_plugins(database_type, :update, new_database_path) do |db_type, ids|
-      logger.info("geoip filter plugin will use database #{new_database_path}",
-                  :database_type => db_type, :pipeline_ids => ids) unless ids.empty?
+    @trigger_lock.synchronize do
+      new_database_path = updated_db_info.path
+      @states[database_type].database_path = new_database_path
+      notify_plugins(database_type, :update, new_database_path) do |db_type, ids|
+        logger.info("geoip filter plugin will use database #{new_database_path}",
+                    :database_type => db_type, :pipeline_ids => ids) unless ids.empty?
+      end
     end
   end
 
   def expire!(database_type)
-    notify_plugins(database_type, :expire) do |db_type, ids|
-      logger.warn("geoip filter plugin will stop filtering and will tag all events with the '_geoip_expired_database' tag.",
-                  :database_type => db_type, :pipeline_ids => ids)
+    @trigger_lock.synchronize do
+      @states[database_type].database_path = nil
+      notify_plugins(database_type, :expire) do |db_type, ids|
+        logger.warn("geoip filter plugin will stop filtering and will tag all events with the '_geoip_expired_database' tag.",
+                    :database_type => db_type, :pipeline_ids => ids)
+      end
     end
   end
 

x-pack/spec/filters/geoip/database_manager_spec.rb

@@ -8,11 +8,14 @@
 describe LogStash::Filters::Geoip do
   describe 'DatabaseManager', :aggregate_failures do
     let(:pipeline_id) { SecureRandom.hex(16) }
-    let(:mock_geoip_plugin) do
-      double("LogStash::Filters::Geoip").tap do |c|
-        allow(c).to receive(:execution_context).and_return(double("EC", pipeline_id: pipeline_id))
-        allow(c).to receive(:update_filter).with(anything)
-      end
+    let(:mock_geoip_plugin) { mock_geoip_plugin_factory.call }
+    let(:mock_geoip_plugin_factory) do
+      ->() {
+        double("LogStash::Filters::Geoip").tap do |c|
+          allow(c).to receive(:execution_context).and_return(double("EC", pipeline_id: pipeline_id))
+          allow(c).to receive(:update_filter).with(anything)
+        end
+      }
     end
 
     let(:eula_database_infos) { Hash.new { LogStash::GeoipDatabaseManagement::DbInfo::PENDING } }
@@ -119,10 +122,19 @@
 
           shared_examples "subscribed to expire notifications" do
             context "when the manager expires the db" do
-              it "notifies the plugin" do
+              before(:each) do
                 db_manager.eula_subscription("City").notify(LogStash::GeoipDatabaseManagement::DbInfo::EXPIRED)
+              end
+              it "notifies the plugin" do
                 expect(mock_geoip_plugin).to have_received(:update_filter).with(:expire)
               end
+              context "subsequent subscriptions" do
+                it "are given the nil path" do
+                  plugin2 = mock_geoip_plugin_factory.call
+                  path2 = db_manager.subscribe_database_path("City", nil, plugin2)
+                  expect(path2).to be_nil
+                end
+              end
             end
             context "when the manager expires a different DB" do
               it 'does not notify the plugin' do
@@ -135,10 +147,19 @@
           shared_examples "subscribed to update notifications" do
             context "when the manager updates the db" do
               let(:updated_db_path) { "/this/that/another.mmdb" }
-              it "notifies the plugin" do
+              before(:each) do
                 db_manager.eula_subscription("City").notify(LogStash::GeoipDatabaseManagement::DbInfo.new(path: updated_db_path))
+              end
+              it "notifies the plugin" do
                 expect(mock_geoip_plugin).to have_received(:update_filter).with(:update, updated_db_path)
               end
+              context "subsequent subscriptions" do
+                it "are given the updated path" do
+                  plugin2 = mock_geoip_plugin_factory.call
+                  path2 = db_manager.subscribe_database_path("City", nil, plugin2)
+                  expect(path2).to eq updated_db_path
+                end
+              end
             end
             context "when the manager updates a different DB" do
               let(:updated_db_path) { "/this/that/another.mmdb" }