fix database manager with multiple pipelines and expiry action

.travis.yml

@@ -2,8 +2,6 @@ import:
   - logstash-plugins/.ci:travis/travis.yml@1.x
 
 env:
-  - DISTRIBUTION=default ELASTIC_STACK_VERSION=6.x
-  - DISTRIBUTION=default ELASTIC_STACK_VERSION=7.11.0
   - DISTRIBUTION=default ELASTIC_STACK_VERSION=7.x
   - DISTRIBUTION=default ELASTIC_STACK_VERSION=7.x SNAPSHOT=true
   - DISTRIBUTION=default ELASTIC_STACK_VERSION=8.x SNAPSHOT=true

CHANGELOG.md

@@ -1,3 +1,9 @@
+## 7.2.0
+  - Add EULA GeoIP2 Database with auto-update  [#181](https://github.com/logstash-plugins/logstash-filter-geoip/pull/181)
+    Available in Logstash 7.14+
+  - Support multiple pipelines using the same database
+  - Add EULA doc
+
 ## 7.1.3
   - Fixed resolving wrong `fields` name `AUTONOMOUS_SYSTEM_NUMBER` and `AUTONOMOUS_SYSTEM_ORGANIZATION` [#185](https://github.com/logstash-plugins/logstash-filter-geoip/pull/185)
 

docs/index.asciidoc

@@ -37,6 +37,26 @@ can be https://dev.maxmind.com/geoip/geoip2/geolite2[downloaded from here].
 
 If you would like to get Autonomous System Number(ASN) information, you can use the GeoLite2-ASN database.
 
+[id="plugins-{type}s-{plugin}-database_license"]
+==== Database License
+
+https://www.maxmind.com[MaxMind] changed from releasing the GeoIP database under
+a Creative Commons (CC) license to a proprietary end-user license agreement
+(EULA).  The MaxMind EULA requires Logstash to update the MaxMind database
+within 30 days of a database update. If Logstash fails to download the database
+for 30 days, the geoip filter will stop enriching events in order to maintain compliance.
+Events will be tagged with `_geoip_expired_database` tag to facilitate the handling of this situation.
+
+The GeoIP filter plugin can manage the database for users running the Logstash default
+distribution, or you can manage
+database updates on your own. The behavior is controlled by the `database` setting.
+When you use the default `database` setting, the auto-update feature ensures that the plugin is
+using the latest version of the database.
+Otherwise, you are responsible for maintaining compliance.
+
+The Logstash open source distribution uses the MaxMind Creative Commons license
+database by default.
+
 ==== Details
 
 A `[geoip][location]` field is created if
@@ -110,14 +130,16 @@ number of cache misses and waste memory.
 ===== `database`
 
   * Value type is <<path,path>>
-  * There is no default value for this setting.
+  * If not specified, the database defaults to the GeoLite2 City database that ships with Logstash.
 
 The path to MaxMind's database file that Logstash should use. The default database is GeoLite2-City.
 GeoLite2-City, GeoLite2-Country, GeoLite2-ASN are the free databases from MaxMind that are supported.
 GeoIP2-City, GeoIP2-ISP, GeoIP2-Country are the commercial databases from MaxMind that are supported.
 
-If not specified, this will default to the GeoLite2 City database that ships
-with Logstash.
+Database auto-update applies to default distribution. When `database` points to user's database path,
+auto-update will be disabled.
+See
+<<plugins-{type}s-{plugin}-database_license,Database License>> for more information.
 
 [id="plugins-{type}s-{plugin}-default_database_type"]
 ===== `default_database_type`

lib/logstash/filters/geoip.rb

@@ -144,54 +144,63 @@ def auto_target_from_source!
                                        "requires a `target` when `source` is not an `ip` sub-field, eg. [client][ip]")
   end
 
-
   def setup_filter(database_path)
-    @healthy_database = true
+    @healthy_database = !database_path.nil?
+    return if database_path.nil?
+
     @database = database_path
-    @logger.info("Using geoip database", :path => @database, :healthy_database => @healthy_database)
     @geoipfilter = org.logstash.filters.geoip.GeoIPFilter.new(@source, @target, @fields, @database, @cache_size, ecs_compatibility.to_s)
   end
 
   # call by DatabaseManager
-  def expire_action
-    fail_filter
+  def update_filter(action, *args)
+    @logger.trace("update filter", :action => action, :args => args) if @logger.trace?
+
+    case action
+    when :update
+      setup_filter(*args)
+    when :expire
+      fail_filter
+    else
+      @logger.warn("invalid action: #{action}")
+    end
   end
 
   def fail_filter
     @healthy_database = false
-    @logger.warn("geoip plugin will stop filtering and will tag all events with the '_geoip_expired_database' tag.",
-                 :healthy_database => @healthy_database)
-  end
-
-  def terminate_filter
-    @logger.info("geoip plugin is terminating")
-    pipeline_id = execution_context.pipeline_id
-    execution_context.agent.stop_pipeline(pipeline_id)
   end
 
   def close
-    @database_manager.close unless @database_manager.nil?
+    @database_manager.unsubscribe_database_path(@default_database_type, self) if @database_manager
   end
 
   def select_database_path
-    vendor_path = ::File.expand_path(::File.join("..", "..", "..", "..", "vendor"), __FILE__)
-
-    if load_database_manager?
-      @database_manager = LogStash::Filters::Geoip::DatabaseManager.new(self, @database, @default_database_type, vendor_path)
-      @database_manager.database_path
-    else
-      @database.nil? ? ::File.join(vendor_path, "GeoLite2-#{@default_database_type}.mmdb") : @database
-    end
+    path =
+      if load_database_manager?
+        @database_manager = LogStash::Filters::Geoip::DatabaseManager.instance
+        @database_manager.subscribe_database_path(@default_database_type, @database, self)
+      else
+        vendor_path = ::File.expand_path(::File.join("..", "..", "..", "..", "vendor"), __FILE__)
+        @database.nil? ? ::File.join(vendor_path, "GeoLite2-#{@default_database_type}.mmdb") : @database
+      end
+
+    @logger.info("Using geoip database", :path => path)
+    path
   end
 
   def load_database_manager?
     begin
       require_relative ::File.join(LogStash::Environment::LOGSTASH_HOME, "x-pack", "lib", "filters", "geoip", "database_manager")
-      true
+      compatible_logstash_version?
     rescue LoadError => e
       @logger.info("DatabaseManager is not in classpath", :version => LOGSTASH_VERSION, :exception => e)
       false
     end
   end
 
+  MINIMUM_LOGSTASH_VERSION=">= 7.14.0".freeze
+  def compatible_logstash_version?
+    Gem::Requirement.new(MINIMUM_LOGSTASH_VERSION).satisfied_by?(Gem::Version.new(LOGSTASH_VERSION))
+  end
+
 end # class LogStash::Filters::GeoIP

logstash-filter-geoip.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-filter-geoip'
-  s.version         = '7.1.3'
+  s.version         = '7.2.0'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "Adds geographical information about an IP address"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
@@ -26,4 +26,6 @@ Gem::Specification.new do |s|
   s.add_development_dependency 'logstash-devutils'
   s.add_development_dependency 'insist'
   s.add_development_dependency 'benchmark-ips'
+  # only compatible with 7.14+ because of the dependency of DatabaseManager
+  s.add_runtime_dependency "logstash-core", ">= 7.14.0"
 end

spec/filters/geoip_offline_spec.rb

@@ -195,6 +195,22 @@
       end
     end
 
+    describe "empty database path" do
+      let(:plugin) { LogStash::Filters::GeoIP.new("source" => "message") }
+      let(:event) { LogStash::Event.new("message" => "8.8.8.8") }
+
+      context "when database manager give nil database path" do
+        it "should tag expired database" do
+          expect(plugin).to receive(:select_database_path).and_return(nil)
+
+          plugin.register
+          plugin.filter(event)
+
+          expect(event.get("tags")).to include("_geoip_expired_database")
+        end
+      end
+    end
+
     describe "filter method outcomes" do
       let(:plugin) { LogStash::Filters::GeoIP.new("source" => "message", "add_tag" => "done", "database" => CITYDB) }
       let(:event) { LogStash::Event.new("message" => ipstring) }

spec/filters/geoip_online_spec.rb

@@ -10,7 +10,7 @@
     ::File.delete(METADATA_PATH) if ::File.exist?(METADATA_PATH)
   end
 
-  xdescribe "config without database path in LS >= 7.13", :aggregate_failures do
+  describe "config without database path in LS >= 7.14", :aggregate_failures do
     before(:each) do
       dir_path = Stud::Temporary.directory
       File.open(dir_path + '/uuid', 'w') { |f| f.write(SecureRandom.uuid) }
@@ -33,20 +33,20 @@
         plugin.register
         plugin.filter(event)
         plugin.close
-        first_database_name = get_metadata_database_name
+        first_dirname = get_metadata_city_database_name
         plugin.register
         plugin.filter(event2)
         plugin.close
-        second_database_name = get_metadata_database_name
+        second_dirname = get_metadata_city_database_name
 
-        expect(first_database_name).not_to be_nil
-        expect(first_database_name).to eq(second_database_name)
-        expect(::File.exist?(get_file_path(first_database_name))).to be_truthy
+        expect(first_dirname).not_to be_nil
+        expect(first_dirname).to eq(second_dirname)
+        expect(File).to exist(get_file_path(first_dirname))
       end
     end
-  end if MAJOR >= 8 || (MAJOR == 7 && MINOR >= 13)
+  end if MAJOR >= 8 || (MAJOR == 7 && MINOR >= 14)
 
-  describe "config without database path in LS < 7.13" do
+  describe "config without database path in LS < 7.14" do
     context "should run in offline mode" do
       config <<-CONFIG
       filter {
@@ -61,5 +61,5 @@
         expect(::File.exist?(METADATA_PATH)).to be_falsey
       end
     end
-  end if MAJOR < 7 || (MAJOR == 7 && MINOR <= 13)
+  end if MAJOR < 7 || (MAJOR == 7 && MINOR < 14)
 end

spec/filters/geoip_spec.rb

@@ -19,13 +19,13 @@
       end
     end
 
-    describe "> 7.13" do
+    describe ">= 7.14" do
       it "load_database_manager? should be true" do
         expect(plugin.load_database_manager?).to be_truthy
       end
-    end if MAJOR >= 8 || (MAJOR == 7 && MINOR >= 13)
+    end if MAJOR >= 8 || (MAJOR == 7 && MINOR >= 14)
 
-    describe "<= 7.12" do
+    describe "<= 7.13" do
       it "load_database_manager? should be false" do
         expect(plugin.load_database_manager?).to be_falsey
       end
@@ -37,6 +37,6 @@
           expect(plugin.select_database_path).to eql(DEFAULT_CITY_DB_PATH)
         end
       end
-    end if MAJOR < 7 || (MAJOR == 7 && MINOR <= 12)
+    end if MAJOR < 7 || (MAJOR == 7 && MINOR <= 13)
   end
 end

spec/filters/test_helper.rb

@@ -1,5 +1,6 @@
 require "logstash-core/logstash-core"
 require "digest"
+require "csv"
 
 def get_vendor_path(filename)
   ::File.join(::File.expand_path("../../vendor/", ::File.dirname(__FILE__)), filename)
@@ -13,8 +14,13 @@ def get_file_path(filename)
   ::File.join(get_data_dir, filename)
 end
 
-def get_metadata_database_name
-  ::File.exist?(METADATA_PATH) ? ::File.read(METADATA_PATH).split(",").last[0..-2] : nil
+def get_metadata_city_database_name
+  if ::File.exist?(METADATA_PATH)
+    city = ::CSV.read(METADATA_PATH, headers: false).select { |row| row[0].eql?("City") }.last
+    city[3]
+  else
+    nil
+  end
 end
 
 METADATA_PATH = get_file_path("metadata.csv")