|
|
@@ -21,6 +21,7 @@ import org.elasticsearch.common.settings.Settings;
|
|
|
import org.elasticsearch.index.VersionType;
|
|
|
import org.elasticsearch.ingest.IngestDocument;
|
|
|
import org.elasticsearch.ingest.IngestService;
|
|
|
+import org.elasticsearch.ingest.Processor;
|
|
|
import org.elasticsearch.ingest.RandomDocumentPicks;
|
|
|
import org.elasticsearch.persistent.PersistentTasksCustomMetadata;
|
|
|
import org.elasticsearch.test.ESTestCase;
|
|
|
@@ -35,6 +36,7 @@ import java.io.ByteArrayInputStream;
|
|
|
import java.io.IOException;
|
|
|
import java.nio.file.Files;
|
|
|
import java.nio.file.Path;
|
|
|
+import java.nio.file.StandardCopyOption;
|
|
|
import java.util.ArrayList;
|
|
|
import java.util.Collections;
|
|
|
import java.util.EnumSet;
|
|
|
@@ -45,7 +47,8 @@ import java.util.Map;
|
|
|
import java.util.Set;
|
|
|
|
|
|
import static org.hamcrest.Matchers.equalTo;
|
|
|
-import static org.hamcrest.Matchers.notNullValue;
|
|
|
+import static org.hamcrest.Matchers.instanceOf;
|
|
|
+import static org.hamcrest.Matchers.nullValue;
|
|
|
import static org.hamcrest.Matchers.sameInstance;
|
|
|
import static org.mockito.Mockito.mock;
|
|
|
import static org.mockito.Mockito.when;
|
|
|
@@ -53,22 +56,24 @@ import static org.mockito.Mockito.when;
|
|
|
public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
|
|
|
private Path geoipTmpDir;
|
|
|
+ private Path geoIpConfigDir;
|
|
|
+ private LocalDatabases localDatabases;
|
|
|
private DatabaseRegistry databaseRegistry;
|
|
|
private ClusterService clusterService;
|
|
|
|
|
|
@Before
|
|
|
public void loadDatabaseReaders() throws IOException {
|
|
|
- final Path geoIpDir = createTempDir();
|
|
|
final Path configDir = createTempDir();
|
|
|
- final Path geoIpConfigDir = configDir.resolve("ingest-geoip");
|
|
|
+ geoIpConfigDir = configDir.resolve("ingest-geoip");
|
|
|
Files.createDirectories(geoIpConfigDir);
|
|
|
- copyDatabaseFiles(geoIpDir);
|
|
|
|
|
|
Client client = mock(Client.class);
|
|
|
GeoIpCache cache = new GeoIpCache(1000);
|
|
|
- LocalDatabases localDatabases = new LocalDatabases(geoIpDir, geoIpConfigDir, new GeoIpCache(1000));
|
|
|
+ localDatabases = new LocalDatabases(geoIpConfigDir, new GeoIpCache(1000));
|
|
|
+ copyDatabaseFiles(geoIpConfigDir, localDatabases);
|
|
|
geoipTmpDir = createTempDir();
|
|
|
databaseRegistry = new DatabaseRegistry(geoipTmpDir, client, cache, localDatabases, Runnable::run);
|
|
|
+ databaseRegistry.initialize("nodeId", mock(ResourceWatcherService.class), mock(IngestService.class));
|
|
|
clusterService = mock(ClusterService.class);
|
|
|
when(clusterService.state()).thenReturn(ClusterState.EMPTY_STATE);
|
|
|
}
|
|
|
@@ -86,7 +91,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
config.put("field", "_field");
|
|
|
String processorTag = randomAlphaOfLength(10);
|
|
|
|
|
|
- GeoIpProcessor processor = factory.create(null, processorTag, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
|
|
|
assertThat(processor.getTag(), equalTo(processorTag));
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
assertThat(processor.getTargetField(), equalTo("geoip"));
|
|
|
@@ -103,7 +108,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
config.put("ignore_missing", true);
|
|
|
String processorTag = randomAlphaOfLength(10);
|
|
|
|
|
|
- GeoIpProcessor processor = factory.create(null, processorTag, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
|
|
|
assertThat(processor.getTag(), equalTo(processorTag));
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
assertThat(processor.getTargetField(), equalTo("geoip"));
|
|
|
@@ -120,7 +125,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
config.put("database_file", "GeoLite2-Country.mmdb");
|
|
|
String processorTag = randomAlphaOfLength(10);
|
|
|
|
|
|
- GeoIpProcessor processor = factory.create(null, processorTag, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
|
|
|
|
|
|
assertThat(processor.getTag(), equalTo(processorTag));
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
@@ -138,7 +143,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
config.put("database_file", "GeoLite2-ASN.mmdb");
|
|
|
String processorTag = randomAlphaOfLength(10);
|
|
|
|
|
|
- GeoIpProcessor processor = factory.create(null, processorTag, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
|
|
|
|
|
|
assertThat(processor.getTag(), equalTo(processorTag));
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
@@ -153,7 +158,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("target_field", "_field");
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
assertThat(processor.getTargetField(), equalTo("_field"));
|
|
|
assertFalse(processor.isIgnoreMissing());
|
|
|
@@ -164,7 +169,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("database_file", "GeoLite2-Country.mmdb");
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
assertThat(processor.getTargetField(), equalTo("geoip"));
|
|
|
assertThat(processor.getDatabaseType(), equalTo("GeoLite2-Country"));
|
|
|
@@ -201,6 +206,9 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
}
|
|
|
|
|
|
public void testBuildNonExistingDbFile() throws Exception {
|
|
|
+ Files.copy(GeoIpProcessorFactoryTests.class.getResourceAsStream("/GeoLite2-City-Test.mmdb"),
|
|
|
+ geoipTmpDir.resolve("GeoLite2-City.mmdb"));
|
|
|
+ databaseRegistry.updateDatabase("GeoLite2-City.mmdb", "md5", geoipTmpDir.resolve("GeoLite2-City.mmdb"));
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
@@ -210,6 +218,16 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
assertThat(e.getMessage(), equalTo("[database_file] database file [does-not-exist.mmdb] doesn't exist"));
|
|
|
}
|
|
|
|
|
|
+ public void testBuildNoDatabasesDownloaded() throws Exception {
|
|
|
+ GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
+
|
|
|
+ Map<String, Object> config = new HashMap<>();
|
|
|
+ config.put("field", "_field");
|
|
|
+ config.put("database_file", "does-not-exist-yet.mmdb");
|
|
|
+ Processor processor = factory.create(null, null, null, config);
|
|
|
+ assertThat(processor, instanceOf(GeoIpProcessor.DatabaseUnavailableProcessor.class));
|
|
|
+ }
|
|
|
+
|
|
|
public void testBuildFields() throws Exception {
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
|
|
|
@@ -228,7 +246,7 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("properties", fieldNames);
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
|
|
|
assertThat(processor.getField(), equalTo("_field"));
|
|
|
assertThat(processor.getProperties(), equalTo(properties));
|
|
|
assertFalse(processor.isIgnoreMissing());
|
|
|
@@ -252,21 +270,20 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
}
|
|
|
|
|
|
public void testLazyLoading() throws Exception {
|
|
|
- final Path geoIpDir = createTempDir();
|
|
|
final Path configDir = createTempDir();
|
|
|
final Path geoIpConfigDir = configDir.resolve("ingest-geoip");
|
|
|
Files.createDirectories(geoIpConfigDir);
|
|
|
- copyDatabaseFiles(geoIpDir);
|
|
|
+ GeoIpCache cache = new GeoIpCache(1000);
|
|
|
+ LocalDatabases localDatabases = new LocalDatabases(geoIpConfigDir, cache);
|
|
|
+ copyDatabaseFiles(geoIpConfigDir, localDatabases);
|
|
|
|
|
|
// Loading another database reader instances, because otherwise we can't test lazy loading as the
|
|
|
// database readers used at class level are reused between tests. (we want to keep that otherwise running this
|
|
|
// test will take roughly 4 times more time)
|
|
|
Client client = mock(Client.class);
|
|
|
- GeoIpCache cache = new GeoIpCache(1000);
|
|
|
- LocalDatabases localDatabases = new LocalDatabases(geoIpDir, geoIpConfigDir, cache);
|
|
|
DatabaseRegistry databaseRegistry = new DatabaseRegistry(createTempDir(), client, cache, localDatabases, Runnable::run);
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
- for (DatabaseReaderLazyLoader lazyLoader : localDatabases.getAllDatabases()) {
|
|
|
+ for (DatabaseReaderLazyLoader lazyLoader : localDatabases.getConfigDatabases().values()) {
|
|
|
assertNull(lazyLoader.databaseReader.get());
|
|
|
}
|
|
|
|
|
|
@@ -276,43 +293,43 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("database_file", "GeoLite2-City.mmdb");
|
|
|
- final GeoIpProcessor city = factory.create(null, "_tag", null, config);
|
|
|
+ final GeoIpProcessor city = (GeoIpProcessor) factory.create(null, "_tag", null, config);
|
|
|
|
|
|
// these are lazy loaded until first use so we expect null here
|
|
|
- assertNull(databaseRegistry.getDatabase("GeoLite2-City.mmdb", true).databaseReader.get());
|
|
|
+ assertNull(databaseRegistry.getDatabase("GeoLite2-City.mmdb").databaseReader.get());
|
|
|
city.execute(document);
|
|
|
// the first ingest should trigger a database load
|
|
|
- assertNotNull(databaseRegistry.getDatabase("GeoLite2-City.mmdb", true).databaseReader.get());
|
|
|
+ assertNotNull(databaseRegistry.getDatabase("GeoLite2-City.mmdb").databaseReader.get());
|
|
|
|
|
|
config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("database_file", "GeoLite2-Country.mmdb");
|
|
|
- final GeoIpProcessor country = factory.create(null, "_tag", null, config);
|
|
|
+ final GeoIpProcessor country = (GeoIpProcessor) factory.create(null, "_tag", null, config);
|
|
|
|
|
|
// these are lazy loaded until first use so we expect null here
|
|
|
- assertNull(databaseRegistry.getDatabase("GeoLite2-Country.mmdb", true).databaseReader.get());
|
|
|
+ assertNull(databaseRegistry.getDatabase("GeoLite2-Country.mmdb").databaseReader.get());
|
|
|
country.execute(document);
|
|
|
// the first ingest should trigger a database load
|
|
|
- assertNotNull(databaseRegistry.getDatabase("GeoLite2-Country.mmdb", true).databaseReader.get());
|
|
|
+ assertNotNull(databaseRegistry.getDatabase("GeoLite2-Country.mmdb").databaseReader.get());
|
|
|
|
|
|
config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("database_file", "GeoLite2-ASN.mmdb");
|
|
|
- final GeoIpProcessor asn = factory.create(null, "_tag", null, config);
|
|
|
+ final GeoIpProcessor asn = (GeoIpProcessor) factory.create(null, "_tag", null, config);
|
|
|
|
|
|
// these are lazy loaded until first use so we expect null here
|
|
|
- assertNull(databaseRegistry.getDatabase("GeoLite2-ASN.mmdb", true).databaseReader.get());
|
|
|
+ assertNull(databaseRegistry.getDatabase("GeoLite2-ASN.mmdb").databaseReader.get());
|
|
|
asn.execute(document);
|
|
|
// the first ingest should trigger a database load
|
|
|
- assertNotNull(databaseRegistry.getDatabase("GeoLite2-ASN.mmdb", true).databaseReader.get());
|
|
|
+ assertNotNull(databaseRegistry.getDatabase("GeoLite2-ASN.mmdb").databaseReader.get());
|
|
|
}
|
|
|
|
|
|
public void testLoadingCustomDatabase() throws IOException {
|
|
|
- final Path geoIpDir = createTempDir();
|
|
|
final Path configDir = createTempDir();
|
|
|
final Path geoIpConfigDir = configDir.resolve("ingest-geoip");
|
|
|
Files.createDirectories(geoIpConfigDir);
|
|
|
- copyDatabaseFiles(geoIpDir);
|
|
|
+ LocalDatabases localDatabases = new LocalDatabases(geoIpConfigDir, new GeoIpCache(1000));
|
|
|
+ copyDatabaseFiles(geoIpConfigDir, localDatabases);
|
|
|
// fake the GeoIP2-City database
|
|
|
copyDatabaseFile(geoIpConfigDir, "GeoLite2-City.mmdb");
|
|
|
Files.move(geoIpConfigDir.resolve("GeoLite2-City.mmdb"), geoIpConfigDir.resolve("GeoIP2-City.mmdb"));
|
|
|
@@ -323,13 +340,12 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
*/
|
|
|
ThreadPool threadPool = new TestThreadPool("test");
|
|
|
ResourceWatcherService resourceWatcherService = new ResourceWatcherService(Settings.EMPTY, threadPool);
|
|
|
- LocalDatabases localDatabases = new LocalDatabases(geoIpDir, geoIpConfigDir, new GeoIpCache(1000));
|
|
|
Client client = mock(Client.class);
|
|
|
GeoIpCache cache = new GeoIpCache(1000);
|
|
|
DatabaseRegistry databaseRegistry = new DatabaseRegistry(createTempDir(), client, cache, localDatabases, Runnable::run);
|
|
|
databaseRegistry.initialize("nodeId", resourceWatcherService, mock(IngestService.class));
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
- for (DatabaseReaderLazyLoader lazyLoader : localDatabases.getAllDatabases()) {
|
|
|
+ for (DatabaseReaderLazyLoader lazyLoader : localDatabases.getConfigDatabases().values()) {
|
|
|
assertNull(lazyLoader.databaseReader.get());
|
|
|
}
|
|
|
|
|
|
@@ -339,35 +355,24 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "_field");
|
|
|
config.put("database_file", "GeoIP2-City.mmdb");
|
|
|
- final GeoIpProcessor city = factory.create(null, "_tag", null, config);
|
|
|
+ final GeoIpProcessor city = (GeoIpProcessor) factory.create(null, "_tag", null, config);
|
|
|
|
|
|
// these are lazy loaded until first use so we expect null here
|
|
|
- assertNull(databaseRegistry.getDatabase("GeoIP2-City.mmdb", true).databaseReader.get());
|
|
|
+ assertNull(databaseRegistry.getDatabase("GeoIP2-City.mmdb").databaseReader.get());
|
|
|
city.execute(document);
|
|
|
// the first ingest should trigger a database load
|
|
|
- assertNotNull(databaseRegistry.getDatabase("GeoIP2-City.mmdb", true).databaseReader.get());
|
|
|
+ assertNotNull(databaseRegistry.getDatabase("GeoIP2-City.mmdb").databaseReader.get());
|
|
|
resourceWatcherService.close();
|
|
|
threadPool.shutdown();
|
|
|
}
|
|
|
|
|
|
public void testFallbackUsingDefaultDatabases() throws Exception {
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
- {
|
|
|
- Map<String, Object> config = new HashMap<>();
|
|
|
- config.put("field", "source_field");
|
|
|
- config.put("fallback_to_default_databases", false);
|
|
|
- Exception e = expectThrows(ElasticsearchParseException.class, () -> factory.create(null, null, null, config));
|
|
|
- assertThat(e.getMessage(), equalTo("[database_file] database file [GeoLite2-City.mmdb] doesn't exist"));
|
|
|
- }
|
|
|
- {
|
|
|
- Map<String, Object> config = new HashMap<>();
|
|
|
- config.put("field", "source_field");
|
|
|
- if (randomBoolean()) {
|
|
|
- config.put("fallback_to_default_databases", true);
|
|
|
- }
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
- assertThat(processor, notNullValue());
|
|
|
- }
|
|
|
+ Map<String, Object> config = new HashMap<>();
|
|
|
+ config.put("field", "source_field");
|
|
|
+ config.put("fallback_to_default_databases", randomBoolean());
|
|
|
+ factory.create(null, null, null, config);
|
|
|
+ assertWarnings(GeoIpProcessor.DEFAULT_DATABASES_DEPRECATION_MESSAGE);
|
|
|
}
|
|
|
|
|
|
public void testDefaultDatabaseWithTaskPresent() throws Exception {
|
|
|
@@ -385,63 +390,93 @@ public class GeoIpProcessorFactoryTests extends ESTestCase {
|
|
|
config.put("field", "_field");
|
|
|
String processorTag = randomAlphaOfLength(10);
|
|
|
|
|
|
- GeoIpProcessor processor = factory.create(null, processorTag, null, config);
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
|
|
|
|
|
|
processor.execute(RandomDocumentPicks.randomIngestDocument(random(), Map.of("_field", "89.160.20.128")));
|
|
|
}
|
|
|
|
|
|
- public void testFallbackUsingDefaultDatabasesWhileIngesting() throws Exception {
|
|
|
- copyDatabaseFile(geoipTmpDir, "GeoLite2-City-Test.mmdb");
|
|
|
+ public void testUpdateDatabaseWhileIngesting() throws Exception {
|
|
|
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
- // fallback_to_default_databases=true, first use default city db then a custom city db:
|
|
|
+ Map<String, Object> config = new HashMap<>();
|
|
|
+ config.put("field", "source_field");
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
|
|
|
+ Map<String, Object> document = new HashMap<>();
|
|
|
+ document.put("source_field", "89.160.20.128");
|
|
|
{
|
|
|
- Map<String, Object> config = new HashMap<>();
|
|
|
- config.put("field", "source_field");
|
|
|
- if (randomBoolean()) {
|
|
|
- config.put("fallback_to_default_databases", true);
|
|
|
- }
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
- Map<String, Object> document = new HashMap<>();
|
|
|
- document.put("source_field", "89.160.20.128");
|
|
|
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
processor.execute(ingestDocument);
|
|
|
Map<?, ?> geoData = (Map<?, ?>) ingestDocument.getSourceAndMetadata().get("geoip");
|
|
|
assertThat(geoData.get("city_name"), equalTo("Tumba"));
|
|
|
-
|
|
|
+ }
|
|
|
+ {
|
|
|
+ copyDatabaseFile(geoipTmpDir, "GeoLite2-City-Test.mmdb");
|
|
|
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
databaseRegistry.updateDatabase("GeoLite2-City.mmdb", "md5", geoipTmpDir.resolve("GeoLite2-City-Test.mmdb"));
|
|
|
- ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
processor.execute(ingestDocument);
|
|
|
- geoData = (Map<?, ?>) ingestDocument.getSourceAndMetadata().get("geoip");
|
|
|
+ Map<?, ?> geoData = (Map<?, ?>) ingestDocument.getSourceAndMetadata().get("geoip");
|
|
|
assertThat(geoData.get("city_name"), equalTo("Linköping"));
|
|
|
}
|
|
|
- // fallback_to_default_databases=false, first use a custom city db then remove the custom db and expect failure:
|
|
|
+ {
|
|
|
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
+ databaseRegistry.removeStaleEntries(List.of("GeoLite2-City.mmdb"));
|
|
|
+ localDatabases.updateDatabase(geoIpConfigDir.resolve("GeoLite2-City.mmdb"), false);
|
|
|
+ Exception e = expectThrows(ResourceNotFoundException.class, () -> processor.execute(ingestDocument));
|
|
|
+ assertThat(e.getMessage(), equalTo("database file [GeoLite2-City.mmdb] doesn't exist"));
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ public void testDatabaseNotReadyYet() throws Exception {
|
|
|
+ GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseRegistry, clusterService);
|
|
|
+
|
|
|
{
|
|
|
Map<String, Object> config = new HashMap<>();
|
|
|
config.put("field", "source_field");
|
|
|
- config.put("fallback_to_default_databases", false);
|
|
|
- GeoIpProcessor processor = factory.create(null, null, null, config);
|
|
|
+ config.put("database_file", "GeoLite2-City-Test.mmdb");
|
|
|
+
|
|
|
Map<String, Object> document = new HashMap<>();
|
|
|
document.put("source_field", "89.160.20.128");
|
|
|
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
+
|
|
|
+ GeoIpProcessor.DatabaseUnavailableProcessor processor =
|
|
|
+ (GeoIpProcessor.DatabaseUnavailableProcessor) factory.create(null, null, null, config);
|
|
|
processor.execute(ingestDocument);
|
|
|
+ assertThat(ingestDocument.getSourceAndMetadata().get("geoip"), nullValue());
|
|
|
+ assertThat(ingestDocument.getSourceAndMetadata().get("tags"),
|
|
|
+ equalTo(List.of("_geoip_database_unavailable_GeoLite2-City-Test.mmdb")));
|
|
|
+ }
|
|
|
+
|
|
|
+ copyDatabaseFile(geoipTmpDir, "GeoLite2-City-Test.mmdb");
|
|
|
+ databaseRegistry.updateDatabase("GeoLite2-City-Test.mmdb", "md5", geoipTmpDir.resolve("GeoLite2-City-Test.mmdb"));
|
|
|
+
|
|
|
+ {
|
|
|
+ Map<String, Object> config = new HashMap<>();
|
|
|
+ config.put("field", "source_field");
|
|
|
+ config.put("database_file", "GeoLite2-City-Test.mmdb");
|
|
|
+
|
|
|
+ Map<String, Object> document = new HashMap<>();
|
|
|
+ document.put("source_field", "89.160.20.128");
|
|
|
+ IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
|
|
|
+
|
|
|
+ GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
|
|
|
+ processor.execute(ingestDocument);
|
|
|
+ assertThat(ingestDocument.getSourceAndMetadata().get("tags"), nullValue());
|
|
|
Map<?, ?> geoData = (Map<?, ?>) ingestDocument.getSourceAndMetadata().get("geoip");
|
|
|
assertThat(geoData.get("city_name"), equalTo("Linköping"));
|
|
|
- databaseRegistry.removeStaleEntries(List.of("GeoLite2-City.mmdb"));
|
|
|
- Exception e = expectThrows(ResourceNotFoundException.class, () -> processor.execute(ingestDocument));
|
|
|
- assertThat(e.getMessage(), equalTo("database file [GeoLite2-City.mmdb] doesn't exist"));
|
|
|
}
|
|
|
}
|
|
|
|
|
|
private static void copyDatabaseFile(final Path path, final String databaseFilename) throws IOException {
|
|
|
Files.copy(
|
|
|
new ByteArrayInputStream(StreamsUtils.copyToBytesFromClasspath("/" + databaseFilename)),
|
|
|
- path.resolve(databaseFilename)
|
|
|
+ path.resolve(databaseFilename),
|
|
|
+ StandardCopyOption.REPLACE_EXISTING
|
|
|
);
|
|
|
}
|
|
|
|
|
|
- static void copyDatabaseFiles(final Path path) throws IOException {
|
|
|
+ static void copyDatabaseFiles(final Path path, LocalDatabases localDatabases) throws IOException {
|
|
|
for (final String databaseFilename : IngestGeoIpPlugin.DEFAULT_DATABASE_FILENAMES) {
|
|
|
copyDatabaseFile(path, databaseFilename);
|
|
|
+ localDatabases.updateDatabase(path.resolve(databaseFilename), true);
|
|
|
}
|
|
|
}
|
|
|
|
Martijn van Groningen