Include geo db version in metadata

Fixes #537

.circleci/config.yml

@@ -99,9 +99,8 @@ jobs:
           - maven-repo-v1-{{ .Branch }}-
           - maven-repo-v1-
     - run: &download_beam_resources
-        name: Download Geolite2 and schemas
+        name: Download resources needed for Beam tests
         command: |
-          ingestion-beam/bin/download-geolite2
           ingestion-beam/bin/download-schemas
     - run:
         name: Maven Test

ingestion-beam/.gitignore

@@ -7,6 +7,7 @@ derby.log
 .DS_Store
 tmp/
 GeoLite2-City.mmdb
+cities15000.txt
 schemas.tar.gz
 avro-schema.tar.gz
 bq-schemas/

ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/GeoCityLookup.java

@@ -23,6 +23,8 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -116,6 +118,9 @@ public class GeoCityLookup
         }
 
         try {
+          attributes.put("geo_db_version", DateTimeFormatter.ISO_INSTANT
+              .format(Instant.ofEpochMilli(geoIP2City.getMetadata().getBuildDate().getTime())));
+
           // Throws UnknownHostException
           InetAddress ipAddress = InetAddress.getByName(ip);
           foundIp.inc();

ingestion-beam/src/main/java/com/mozilla/telemetry/decoder/ParseProxy.java

@@ -50,7 +50,7 @@ public class ParseProxy extends PTransform<PCollection<PubsubMessage>, PCollecti
       message = PubsubConstraints.ensureNonNull(message);
 
       // Copy attributes
-      Map<String, String> attributes = new HashMap<String, String>(message.getAttributeMap());
+      Map<String, String> attributes = new HashMap<>(message.getAttributeMap());
 
       String xpp = attributes.get(X_PIPELINE_PROXY);
       if (xpp != null) {

ingestion-beam/src/test/java/com/mozilla/telemetry/DecoderMainTest.java

@@ -63,8 +63,8 @@ public class DecoderMainTest {
     Decoder.main(new String[] { "--inputFileFormat=json", "--inputType=file", "--input=" + input,
         "--outputFileFormat=json", "--outputType=file", "--output=" + output,
         "--errorOutputType=file", "--errorOutput=" + errorOutput,
-        "--geoCityDatabase=GeoLite2-City.mmdb", "--schemasLocation=schemas.tar.gz",
-        "--errorOutputFileCompression=UNCOMPRESSED" });
+        "--geoCityDatabase=src/test/resources/cityDB/GeoIP2-City-Test.mmdb",
+        "--schemasLocation=schemas.tar.gz", "--errorOutputFileCompression=UNCOMPRESSED" });
 
     List<String> errorOutputLines = Lines.files(errorOutput + "*.ndjson");
     assertThat(errorOutputLines, Matchers.hasSize(1));
@@ -82,8 +82,8 @@ public class DecoderMainTest {
         "--outputFileFormat=json", "--outputType=file", "--output=" + output,
         "--errorOutputType=file", "--errorOutput=" + errorOutput, "--includeStackTrace=false",
         "--outputFileCompression=UNCOMPRESSED", "--errorOutputFileCompression=UNCOMPRESSED",
-        "--geoCityDatabase=GeoLite2-City.mmdb", "--schemasLocation=schemas.tar.gz",
-        "--redisUri=" + redis.uri });
+        "--geoCityDatabase=src/test/resources/cityDB/GeoIP2-City-Test.mmdb",
+        "--schemasLocation=schemas.tar.gz", "--redisUri=" + redis.uri });
 
     List<String> outputLines = Lines.files(output + "*.ndjson");
     List<String> expectedOutputLines = Lines.files(resourceDir + "/output.ndjson");
@@ -108,8 +108,8 @@ public class DecoderMainTest {
         "--outputFileFormat=json", "--outputType=file", "--output=" + output,
         "--outputFileCompression=UNCOMPRESSED", "--errorOutputType=file",
         "--errorOutput=" + errorOutput, "--includeStackTrace=false",
-        "--geoCityDatabase=GeoLite2-City.mmdb", "--schemasLocation=schemas.tar.gz",
-        "--redisUri=" + redis.uri });
+        "--geoCityDatabase=src/test/resources/cityDB/GeoIP2-City-Test.mmdb",
+        "--schemasLocation=schemas.tar.gz", "--redisUri=" + redis.uri });
 
     List<String> outputLines = Lines.files(output + "*.ndjson");
     List<String> expectedOutputLines = Lines.files(resourceDir + "/output.ndjson");

ingestion-beam/src/test/java/com/mozilla/telemetry/decoder/GeoCityLookupTest.java

@@ -29,6 +29,8 @@ import org.junit.rules.ExpectedException;
 
 public class GeoCityLookupTest {
 
+  private static final String MMDB = "src/test/resources/cityDB/GeoIP2-City-Test.mmdb";
+
   @Rule
   public final transient TestPipeline pipeline = TestPipeline.create();
 
@@ -37,31 +39,35 @@ public class GeoCityLookupTest {
 
   @Test
   public void testOutput() {
+    // Some of the IPs below are chosen specifically because they are contained in the test city
+    // database; see the json source for the test db in:
+    // https://github.com/maxmind/MaxMind-DB/blob/664aeeb08bb50f53a1fdceac763c37f6465e44a4/source-data/GeoIP2-City-Test.json
     final List<String> input = Arrays.asList(
         "{\"attributeMap\":{\"host\":\"test\"},\"payload\":\"dGVzdA==\"}", //
-        "{\"attributeMap\":{\"remote_addr\":\"8.8.8.8\"},\"payload\":\"\"}", //
+        "{\"attributeMap\":{\"remote_addr\":\"202.196.224.0\"},\"payload\":\"\"}", //
         "{\"attributeMap\":" //
             + "{\"remote_addr\":\"10.0.0.2\"" //
-            + ",\"x_forwarded_for\":\"192.168.1.2, 63.245.208.195, 60.1.1.1\"" //
+            + ",\"x_forwarded_for\":\"192.168.1.2, 216.160.83.56, 60.1.1.1\"" //
             + "},\"payload\":\"\"}");
 
     final List<String> expected = Arrays.asList(
-        "{\"attributeMap\":{\"host\":\"test\"},\"payload\":\"dGVzdA==\"}", //
+        "{\"attributeMap\":{\"host\":\"test\"" + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\""
+            + "},\"payload\":\"dGVzdA==\"}", //
+        "{\"attributeMap\":" //
+            + "{\"geo_country\":\"PH\"" //
+            + ",\"normalized_country_code\":\"PH\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" + "},\"payload\":\"\"}", //
         "{\"attributeMap\":" //
             + "{\"geo_country\":\"US\"" //
             + ",\"normalized_country_code\":\"US\"" //
-            + "},\"payload\":\"\"}", //
-        "{\"attributeMap\":" //
-            + "{\"geo_country\":\"US\"" //
-            + ",\"normalized_country_code\":\"US\"" //
-            + ",\"geo_city\":\"Sacramento\"" //
-            + ",\"geo_subdivision1\":\"CA\"" //
-            + "},\"payload\":\"\"}");
+            + ",\"geo_city\":\"Milton\"" //
+            + ",\"geo_subdivision1\":\"WA\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" + "},\"payload\":\"\"}");
 
     final PCollection<String> output = pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"), null))
+        .apply(GeoCityLookup.of(pipeline.newProvider(MMDB), null))
         .apply(OutputFileFormat.json.encode());
 
     PAssert.that(output).containsInAnyOrder(expected);
@@ -79,18 +85,21 @@ public class GeoCityLookupTest {
 
   @Test
   public void testCityRejected() {
-    final List<String> input = Arrays.asList("{\"attributeMap\":" + "{\"remote_addr\":\"10.0.0.2\""
-        + ",\"x_forwarded_for\":\"192.168.1.2, 63.245.208.195, 60.1.1.1\"" + "},\"payload\":\"\"}");
+    final List<String> input = Arrays.asList("{\"attributeMap\":" //
+        + "{\"remote_addr\":\"10.0.0.2\"" //
+        + ",\"x_forwarded_for\":\"192.168.1.2, 216.160.83.56, 60.1.1.1\"" //
+        + "},\"payload\":\"\"}");
 
     final List<String> expected = Arrays.asList(
         "{\"attributeMap\":" + "{\"geo_country\":\"US\"" + ",\"normalized_country_code\":\"US\"" //
-            + ",\"geo_subdivision1\":\"CA\"" + "},\"payload\":\"\"}");
+            + ",\"geo_subdivision1\":\"WA\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"},\"payload\":\"\"}");
 
     final PCollection<String> output = pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"),
-            pipeline.newProvider("src/test/resources/cityFilters/milton.txt")))
+        .apply(GeoCityLookup.of(pipeline.newProvider(MMDB),
+            pipeline.newProvider("src/test/resources/cityFilters/sacramento.txt")))
         .apply(OutputFileFormat.json.encode());
 
     PAssert.that(output).containsInAnyOrder(expected);
@@ -100,18 +109,20 @@ public class GeoCityLookupTest {
 
   @Test
   public void testCityAllowed() {
-    final List<String> input = Arrays.asList("{\"attributeMap\":" + "{\"remote_addr\":\"10.0.0.2\""
-        + ",\"x_forwarded_for\":\"192.168.1.2, 63.245.208.195, 60.1.1.1\"" + "},\"payload\":\"\"}");
-
-    final List<String> expected = Arrays.asList("{\"attributeMap\":" + "{\"geo_country\":\"US\""
-        + ",\"normalized_country_code\":\"US\"" //
-        + ",\"geo_city\":\"Sacramento\"" + ",\"geo_subdivision1\":\"CA\"" + "},\"payload\":\"\"}");
+    final List<String> input = Arrays.asList("{\"attributeMap\":" //
+        + "{\"remote_addr\":\"10.0.0.2\"" //
+        + ",\"x_forwarded_for\":\"192.168.1.2, 216.160.83.56, 60.1.1.1\"" //
+        + "},\"payload\":\"\"}");
 
+    final List<String> expected = Arrays.asList(
+        "{\"attributeMap\":" + "{\"geo_country\":\"US\"" + ",\"normalized_country_code\":\"US\"" //
+            + ",\"geo_city\":\"Milton\",\"geo_subdivision1\":\"WA\""
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"},\"payload\":\"\"}");
     final PCollection<String> output = pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"),
-            pipeline.newProvider("src/test/resources/cityFilters/sacramento.txt")))
+        .apply(GeoCityLookup.of(pipeline.newProvider(MMDB),
+            pipeline.newProvider("src/test/resources/cityFilters/milton.txt")))
         .apply(OutputFileFormat.json.encode());
 
     PAssert.that(output).containsInAnyOrder(expected);
@@ -144,8 +155,8 @@ public class GeoCityLookupTest {
     pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"),
-            pipeline.newProvider("missing-file.txt")));
+        .apply(
+            GeoCityLookup.of(pipeline.newProvider(MMDB), pipeline.newProvider("missing-file.txt")));
 
     pipeline.run();
   }
@@ -160,7 +171,7 @@ public class GeoCityLookupTest {
     pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"),
+        .apply(GeoCityLookup.of(pipeline.newProvider(MMDB),
             pipeline.newProvider("src/test/resources/cityFilters/invalid.txt")));
 
     pipeline.run();

ingestion-beam/src/test/java/com/mozilla/telemetry/decoder/ParseProxyTest.java

@@ -97,38 +97,42 @@ public class ParseProxyTest {
     final List<String> input = Arrays.asList(//
         "{\"attributeMap\":{},\"payload\":\"\"}", //
         "{\"attributeMap\":" //
-            + "{\"x_forwarded_for\":\"_, 8.8.8.8, _\"" //
+            + "{\"x_forwarded_for\":\"_, 202.196.224.0, _\"" //
             + "},\"payload\":\"notProxied++\"}",
         "{\"attributeMap\":" //
             + "{\"x_pipeline_proxy\":1" //
-            + ",\"x_forwarded_for\":\"_, 8.8.8.8, _, _\"" //
+            + ",\"x_forwarded_for\":\"_, 202.196.224.0, _, _\"" //
             + "},\"payload\":\"proxied+\"}",
         "{\"attributeMap\":" //
             + "{\"x_pipeline_proxy\":\"2000-01-01T00:00:00.000000Z\"" //
-            + ",\"x_forwarded_for\":\"_, 8.8.8.8, _, _\"" //
+            + ",\"x_forwarded_for\":\"_, 202.196.224.0, _, _\"" //
             + "},\"payload\":\"proxiedWithTimestamp\"}");
 
     final List<String> expected = Arrays.asList(//
-        "{\"attributeMap\":{},\"payload\":\"\"}", //
+        "{\"attributeMap\":{\"geo_db_version\":\"2019-01-03T21:26:19Z\"},\"payload\":\"\"}", //
         "{\"attributeMap\":" //
-            + "{\"geo_country\":\"US\"" //
-            + ",\"normalized_country_code\":\"US\"" //
+            + "{\"geo_country\":\"PH\"" //
+            + ",\"normalized_country_code\":\"PH\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" //
             + "},\"payload\":\"notProxied++\"}",
         "{\"attributeMap\":" //
-            + "{\"geo_country\":\"US\"" //
-            + ",\"normalized_country_code\":\"US\"" //
+            + "{\"geo_country\":\"PH\"" //
+            + ",\"normalized_country_code\":\"PH\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" //
             + "},\"payload\":\"proxied+\"}",
         "{\"attributeMap\":" //
-            + "{\"geo_country\":\"US\"" //
-            + ",\"normalized_country_code\":\"US\"" //
+            + "{\"geo_country\":\"PH\"" //
+            + ",\"normalized_country_code\":\"PH\"" //
             + ",\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" //
+            + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" //
             + "},\"payload\":\"proxiedWithTimestamp\"}");
 
     final PCollection<String> output = pipeline //
         .apply(Create.of(input)) //
         .apply(InputFileFormat.json.decode()).output() //
         .apply(ParseProxy.of()) //
-        .apply(GeoCityLookup.of(pipeline.newProvider("GeoLite2-City.mmdb"), null))
+        .apply(GeoCityLookup
+            .of(pipeline.newProvider("src/test/resources/cityDB/GeoIP2-City-Test.mmdb"), null))
         .apply(OutputFileFormat.json.encode());
 
     PAssert.that(output).containsInAnyOrder(expected);

ingestion-beam/src/test/java/com/mozilla/telemetry/integration/StorageIntegrationTest.java

@@ -101,8 +101,8 @@ public class StorageIntegrationTest {
         "--outputFileFormat=json", "--outputType=file", "--output=" + output,
         "--outputFileCompression=UNCOMPRESSED", "--errorOutputFileCompression=UNCOMPRESSED",
         "--errorOutputType=file", "--errorOutput=" + errorOutput, "--includeStackTrace=false",
-        "--geoCityDatabase=GeoLite2-City.mmdb", "--schemasLocation=schemas.tar.gz",
-        "--redisUri=" + redis.uri });
+        "--geoCityDatabase=src/test/resources/cityDB/GeoIP2-City-Test.mmdb",
+        "--schemasLocation=schemas.tar.gz", "--redisUri=" + redis.uri });
 
     tempFolder.newFolder("out");
     tempFolder.newFolder("error");
@@ -131,7 +131,8 @@ public class StorageIntegrationTest {
         "--templateLocation=" + gcsPath + "/templates/TestTemplate",
         "--stagingLocation=" + gcsPath + "/temp/staging", "--inputFileFormat=json",
         "--inputType=file", "--outputFileFormat=json", "--outputType=file",
-        "--errorOutputType=file", "--geoCityDatabase=GeoLite2-City.mmdb",
+        "--errorOutputType=file",
+        "--geoCityDatabase=src/test/resources/cityDB/GeoIP2-City-Test.mmdb",
         "--schemasLocation=schemas.tar.gz", "--schemaAliasesLocation=" + aliases });
 
   }

ingestion-beam/src/test/resources/cityDB/README.md

@@ -0,0 +1,6 @@
+The files are copied from `MaxMind`'s db format repository (latest as of May 2019):
+
+- https://github.com/maxmind/MaxMind-DB/blob/664aeeb08bb50f53a1fdceac763c37f6465e44a4/test-data/GeoIP2-City-Test.mmdb
+- https://github.com/maxmind/MaxMind-DB/blob/664aeeb08bb50f53a1fdceac763c37f6465e44a4/source-data/GeoIP2-City-Test.json
+
+The JSON file gives the readable source of the `.mmdb` file.

ingestion-beam/src/test/resources/testdata/decoder-integration/error-input.ndjson

@@ -1,4 +1,4 @@
 {}
-{"attributeMap":{"host":"test","remote_addr":"8.8.8.8","uri":"/submit/telemetry/main/Firefox/61.0a1/nightly/20180328030202"},"payload":""}
-{"attributeMap":{"host":"test","remote_addr":"8.8.8.8","uri":"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}
-{"attributeMap":{"host":"test","remote_addr":"8.8.8.8","uri":"/submit/test/test/1/invalid_id"},"payload":"e30="}
+{"attributeMap":{"host":"test","remote_addr":"202.196.224.0","uri":"/submit/telemetry/main/Firefox/61.0a1/nightly/20180328030202"},"payload":""}
+{"attributeMap":{"host":"test","remote_addr":"202.196.224.0","uri":"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}
+{"attributeMap":{"host":"test","remote_addr":"202.196.224.0","uri":"/submit/test/test/1/invalid_id"},"payload":"e30="}

ingestion-beam/src/test/resources/testdata/decoder-integration/error-output.ndjson

@@ -1,4 +1,4 @@
 {"attributeMap":{"error_message":"java.io.IOException: not a valid PubsubMessage.payload: null","error_type":"DecodePubsubMessages.Json","exception_class":"java.io.IOException"},"payload":"e30="}
-{"attributeMap":{"error_message":"com.mozilla.telemetry.decoder.ParseUri$InvalidUriException: Found -1 more path elements in the URI than expected for this endpoint","error_type":"ParseUri","exception_class":"com.mozilla.telemetry.decoder.ParseUri$InvalidUriException","host":"test","remote_addr":"8.8.8.8","uri":"/submit/telemetry/main/Firefox/61.0a1/nightly/20180328030202"},"payload":""}
-{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376049","document_namespace":"eng-workflow","document_type":"hgpush","document_version":"1","error_message":"org.everit.json.schema.ValidationException: #: required key [changesetID] not found","error_type":"ParsePayload","exception_class":"org.everit.json.schema.ValidationException","host":"test","remote_addr":"8.8.8.8","uri":"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}
-{"attributeMap":{"document_id":"invalid_id","document_namespace":"test","document_type":"test","document_version":"1","error_message":"java.lang.IllegalArgumentException: Invalid UUID string: invalid_id","error_type":"Deduplicate.RemoveDuplicates","exception_class":"java.lang.IllegalArgumentException","geo_country":"US","host":"test","normalized_country_code":"US","uri":"/submit/test/test/1/invalid_id"},"payload":"eyJkb2N1bWVudF9pZCI6ImludmFsaWRfaWQiLCJtZXRhZGF0YSI6eyJkb2N1bWVudF9uYW1lc3BhY2UiOiJ0ZXN0IiwiZG9jdW1lbnRfdHlwZSI6InRlc3QiLCJkb2N1bWVudF92ZXJzaW9uIjoiMSIsImdlbyI6eyJjb3VudHJ5IjoiVVMifSwiaGVhZGVyIjp7fSwidXNlcl9hZ2VudCI6e319LCJub3JtYWxpemVkX2NvdW50cnlfY29kZSI6IlVTIn0="}
+{"attributeMap":{"error_message":"com.mozilla.telemetry.decoder.ParseUri$InvalidUriException: Found -1 more path elements in the URI than expected for this endpoint","error_type":"ParseUri","exception_class":"com.mozilla.telemetry.decoder.ParseUri$InvalidUriException","host":"test","remote_addr":"202.196.224.0","uri":"/submit/telemetry/main/Firefox/61.0a1/nightly/20180328030202"},"payload":""}
+{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376049","document_namespace":"eng-workflow","document_type":"hgpush","document_version":"1","error_message":"org.everit.json.schema.ValidationException: #: required key [changesetID] not found","error_type":"ParsePayload","exception_class":"org.everit.json.schema.ValidationException","host":"test","remote_addr":"202.196.224.0","uri":"/submit/eng-workflow/hgpush/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}
+{"attributeMap":{"document_id":"invalid_id","document_namespace":"test","document_type":"test","document_version":"1","error_message":"java.lang.IllegalArgumentException: Invalid UUID string: invalid_id","error_type":"Deduplicate.RemoveDuplicates","exception_class":"java.lang.IllegalArgumentException","geo_country":"PH","geo_db_version":"2019-01-03T21:26:19Z","host":"test","normalized_country_code":"PH","uri":"/submit/test/test/1/invalid_id"},"payload":"eyJkb2N1bWVudF9pZCI6ImludmFsaWRfaWQiLCJtZXRhZGF0YSI6eyJkb2N1bWVudF9uYW1lc3BhY2UiOiJ0ZXN0IiwiZG9jdW1lbnRfdHlwZSI6InRlc3QiLCJkb2N1bWVudF92ZXJzaW9uIjoiMSIsImdlbyI6eyJjb3VudHJ5IjoiUEgiLCJkYl92ZXJzaW9uIjoiMjAxOS0wMS0wM1QyMToyNjoxOVoifSwiaGVhZGVyIjp7fSwidXNlcl9hZ2VudCI6e319LCJub3JtYWxpemVkX2NvdW50cnlfY29kZSI6IlBIIn0="}

ingestion-beam/src/test/resources/testdata/decoder-integration/gzipped.ndjson

@@ -1,2 +1,2 @@
 {"attributeMap":{"host":"test1","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376048"},"payload":"H4sIAKnBGlwAA6uuBQBDv6ajAgAAAA=="}
-{"attributeMap":{"host":"test2","remote_addr":"8.8.8.8","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"H4sIAKnBGlwAA6uuBQBDv6ajAgAAAA=="}
+{"attributeMap":{"host":"test2","remote_addr":"202.196.224.0","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"H4sIAKnBGlwAA6uuBQBDv6ajAgAAAA=="}

ingestion-beam/src/test/resources/testdata/decoder-integration/output.ndjson

@@ -1,2 +1,2 @@
-{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376048","document_namespace":"test","document_type":"test","document_version":"1","host":"test1","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376048"},"payload":"eyJkb2N1bWVudF9pZCI6IjJjM2EwNzY3LWQ4NGEtNGQwMi04YTkyLWZhNTRhMzM3NjA0OCIsIm1ldGFkYXRhIjp7ImRvY3VtZW50X25hbWVzcGFjZSI6InRlc3QiLCJkb2N1bWVudF90eXBlIjoidGVzdCIsImRvY3VtZW50X3ZlcnNpb24iOiIxIiwiZ2VvIjp7fSwiaGVhZGVyIjp7fSwidXNlcl9hZ2VudCI6e319fQ=="}
-{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376049","document_namespace":"test","document_type":"test","document_version":"1","geo_country":"US","host":"test2","normalized_country_code":"US","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"eyJkb2N1bWVudF9pZCI6IjJjM2EwNzY3LWQ4NGEtNGQwMi04YTkyLWZhNTRhMzM3NjA0OSIsIm1ldGFkYXRhIjp7ImRvY3VtZW50X25hbWVzcGFjZSI6InRlc3QiLCJkb2N1bWVudF90eXBlIjoidGVzdCIsImRvY3VtZW50X3ZlcnNpb24iOiIxIiwiZ2VvIjp7ImNvdW50cnkiOiJVUyJ9LCJoZWFkZXIiOnt9LCJ1c2VyX2FnZW50Ijp7fX0sIm5vcm1hbGl6ZWRfY291bnRyeV9jb2RlIjoiVVMifQ=="}
+{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376048","document_namespace":"test","document_type":"test","document_version":"1","geo_db_version":"2019-01-03T21:26:19Z","host":"test1","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376048"},"payload":"eyJkb2N1bWVudF9pZCI6IjJjM2EwNzY3LWQ4NGEtNGQwMi04YTkyLWZhNTRhMzM3NjA0OCIsIm1ldGFkYXRhIjp7ImRvY3VtZW50X25hbWVzcGFjZSI6InRlc3QiLCJkb2N1bWVudF90eXBlIjoidGVzdCIsImRvY3VtZW50X3ZlcnNpb24iOiIxIiwiZ2VvIjp7ImRiX3ZlcnNpb24iOiIyMDE5LTAxLTAzVDIxOjI2OjE5WiJ9LCJoZWFkZXIiOnt9LCJ1c2VyX2FnZW50Ijp7fX19"}
+{"attributeMap":{"document_id":"2c3a0767-d84a-4d02-8a92-fa54a3376049","document_namespace":"test","document_type":"test","document_version":"1","geo_country":"PH","geo_db_version":"2019-01-03T21:26:19Z","host":"test2","normalized_country_code":"PH","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"eyJkb2N1bWVudF9pZCI6IjJjM2EwNzY3LWQ4NGEtNGQwMi04YTkyLWZhNTRhMzM3NjA0OSIsIm1ldGFkYXRhIjp7ImRvY3VtZW50X25hbWVzcGFjZSI6InRlc3QiLCJkb2N1bWVudF90eXBlIjoidGVzdCIsImRvY3VtZW50X3ZlcnNpb24iOiIxIiwiZ2VvIjp7ImNvdW50cnkiOiJQSCIsImRiX3ZlcnNpb24iOiIyMDE5LTAxLTAzVDIxOjI2OjE5WiJ9LCJoZWFkZXIiOnt9LCJ1c2VyX2FnZW50Ijp7fX0sIm5vcm1hbGl6ZWRfY291bnRyeV9jb2RlIjoiUEgifQ=="}

ingestion-beam/src/test/resources/testdata/decoder-integration/valid-input.ndjson

@@ -1,2 +1,2 @@
 {"attributeMap":{"host":"test1","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376048"},"payload":"e30="}
-{"attributeMap":{"host":"test2","remote_addr":"8.8.8.8","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}
+{"attributeMap":{"host":"test2","remote_addr":"202.196.224.0","uri":"/submit/test/test/1/2c3a0767-d84a-4d02-8a92-fa54a3376049"},"payload":"e30="}