geoip: don't store geoinfo if nothing can be resolved

plugins/ingest/build.gradle

@@ -48,7 +48,7 @@ task copyDefaultGeoIp2DatabaseFiles(type: Copy) {
 
 project.bundlePlugin.dependsOn(copyDefaultGeoIp2DatabaseFiles)
 
-compileJava.options.compilerArgs << "-Xlint:-rawtypes,-unchecked"
+compileJava.options.compilerArgs << "-Xlint:-rawtypes,-unchecked,-serial"
 compileTestJava.options.compilerArgs << "-Xlint:-rawtypes,-unchecked"
 
 bundlePlugin {

plugins/ingest/src/main/java/org/elasticsearch/ingest/processor/geoip/GeoIpProcessor.java

@@ -20,10 +20,12 @@
 package org.elasticsearch.ingest.processor.geoip;
 
 import com.maxmind.geoip2.DatabaseReader;
+import com.maxmind.geoip2.exception.AddressNotFoundException;
 import com.maxmind.geoip2.exception.GeoIp2Exception;
 import com.maxmind.geoip2.model.CityResponse;
 import com.maxmind.geoip2.model.CountryResponse;
 import com.maxmind.geoip2.record.*;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.SpecialPermission;
 import org.elasticsearch.common.network.NetworkAddress;
 import org.elasticsearch.ingest.Data;
@@ -38,6 +40,7 @@ import java.nio.file.Path;
 import java.nio.file.StandardOpenOption;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -67,13 +70,21 @@ public final class GeoIpProcessor implements Processor {
             throw new RuntimeException(e);
         }
 
-        final Map<String, Object> geoData;
+        Map<String, Object> geoData;
         switch (dbReader.getMetadata().getDatabaseType()) {
             case "GeoLite2-City":
+                try {
                     geoData = retrieveCityGeoData(ipAddress);
+                } catch (AddressNotFoundRuntimeException e) {
+                    geoData = Collections.emptyMap();
+                }
                 break;
             case "GeoLite2-Country":
+                try {
                     geoData = retrieveCountryGeoData(ipAddress);
+                } catch (AddressNotFoundRuntimeException e) {
+                    geoData = Collections.emptyMap();
+                }
                 break;
             default:
                 throw new IllegalStateException("Unsupported database type [" + dbReader.getMetadata().getDatabaseType() + "]");
@@ -98,15 +109,14 @@ public final class GeoIpProcessor implements Processor {
         if (sm != null) {
             sm.checkPermission(new SpecialPermission());
         }
-        CityResponse response = AccessController.doPrivileged(new PrivilegedAction<CityResponse>() {
+        CityResponse response = AccessController.doPrivileged((PrivilegedAction<CityResponse>) () -> {
-            @Override
-            public CityResponse run() {
             try {
                 return dbReader.city(ipAddress);
-                } catch (IOException | GeoIp2Exception e) {
+            } catch (AddressNotFoundException e) {
+                throw new AddressNotFoundRuntimeException(e);
+            } catch (Exception e) {
                 throw new RuntimeException(e);
             }
-            }
         });
 
         Country country = response.getCountry();
@@ -136,15 +146,14 @@ public final class GeoIpProcessor implements Processor {
         if (sm != null) {
             sm.checkPermission(new SpecialPermission());
         }
-        CountryResponse response = AccessController.doPrivileged(new PrivilegedAction<CountryResponse>() {
+        CountryResponse response = AccessController.doPrivileged((PrivilegedAction<CountryResponse>) () -> {
-            @Override
-            public CountryResponse run() {
             try {
                 return dbReader.country(ipAddress);
-                } catch (IOException | GeoIp2Exception e) {
+            } catch (AddressNotFoundException e) {
+                throw new AddressNotFoundRuntimeException(e);
+            } catch (Exception e) {
                 throw new RuntimeException(e);
             }
-            }
         });
 
         Country country = response.getCountry();
@@ -190,4 +199,14 @@ public final class GeoIpProcessor implements Processor {
         }
     }
 
+    // Geoip2's AddressNotFoundException is checked and due to the fact that we need run their code
+    // inside a PrivilegedAction code block, we are forced to catch any checked exception and rethrow
+    // it with an unchecked exception.
+    private final static class AddressNotFoundRuntimeException extends RuntimeException {
+
+        public AddressNotFoundRuntimeException(Throwable cause) {
+            super(cause);
+        }
+    }
+
 }

plugins/ingest/src/test/java/org/elasticsearch/ingest/processor/geoip/GeoIpProcessorTests.java

@@ -77,4 +77,17 @@ public class GeoIpProcessorTests extends ESTestCase {
         assertThat(geoData.get("continent_name"), equalTo("Europe"));
     }
 
+    public void testAddressIsNotInTheDatabase() throws Exception {
+        InputStream database = GeoIpProcessor.class.getResourceAsStream("/GeoLite2-City.mmdb");
+        GeoIpProcessor processor = new GeoIpProcessor("source_field", new DatabaseReader.Builder(database).build(), "target_field");
+
+        Map<String, Object> document = new HashMap<>();
+        document.put("source_field", "202.45.11.11");
+        Data data = new Data("_index", "_type", "_id", document);
+        processor.execute(data);
+        @SuppressWarnings("unchecked")
+        Map<String, Object> geoData = (Map<String, Object>) data.getDocument().get("target_field");
+        assertThat(geoData.size(), equalTo(0));
+    }
+
 }