Add ToS query parameter to GeoIP downloader (#69495)

This change adds query parameter confirming that we accept ToS of GeoIP database service provided by Infra.
It also changes integration test to use lower timeout when using local fixture.

Relates to #68920

modules/ingest-geoip/src/internalClusterTest/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderIT.java

@@ -50,6 +50,8 @@
 @ClusterScope(scope = Scope.TEST, maxNumDataNodes = 1)
 public class GeoIpDownloaderIT extends AbstractGeoIpIT {
 
+    private static final String ENDPOINT = System.getProperty("geoip_endpoint");
+
     @Override
     protected Collection<Class<? extends Plugin>> nodePlugins() {
         return Arrays.asList(ReindexPlugin.class, IngestGeoIpPlugin.class, GeoIpProcessorNonIngestNodeIT.IngestGeoIpSettingsPlugin.class);
@@ -58,9 +60,8 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         Settings.Builder settings = Settings.builder().put(super.nodeSettings(nodeOrdinal));
-        String endpoint = System.getProperty("geoip_endpoint");
-        if (endpoint != null) {
-            settings.put(GeoIpDownloader.ENDPOINT_SETTING.getKey(), endpoint);
+        if (ENDPOINT != null) {
+            settings.put(GeoIpDownloader.ENDPOINT_SETTING.getKey(), ENDPOINT);
         }
         return settings.build();
     }
@@ -75,6 +76,8 @@ public void disableDownloader(){
     }
 
     public void testGeoIpDatabasesDownload() throws Exception {
+        // use short wait for local fixture, longer when we hit real service
+        int waitTime = ENDPOINT == null ? 120 : 10;
         ClusterUpdateSettingsResponse settingsResponse = client().admin().cluster()
             .prepareUpdateSettings()
             .setPersistentSettings(Settings.builder().put(GeoIpDownloaderTaskExecutor.ENABLED_SETTING.getKey(), true))
@@ -86,7 +89,7 @@ public void testGeoIpDatabasesDownload() throws Exception {
             GeoIpTaskState state = (GeoIpTaskState) task.getState();
             assertNotNull(state);
             assertEquals(Set.of("GeoLite2-ASN.mmdb", "GeoLite2-City.mmdb", "GeoLite2-Country.mmdb"), state.getDatabases().keySet());
-        }, 2, TimeUnit.MINUTES);
+        }, waitTime, TimeUnit.SECONDS);
 
         GeoIpTaskState state = (GeoIpTaskState) getTask().getState();
         for (String id : List.of("GeoLite2-ASN.mmdb", "GeoLite2-City.mmdb", "GeoLite2-Country.mmdb")) {

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpDownloader.java

@@ -106,7 +106,7 @@ void updateDatabases() throws IOException {
 
     @SuppressWarnings("unchecked")
     private <T> List<T> fetchDatabasesOverview() throws IOException {
-        byte[] data = httpClient.getBytes(endpoint + "?key=11111111-1111-1111-1111-111111111111");
+        byte[] data = httpClient.getBytes(endpoint + "?key=11111111-1111-1111-1111-111111111111&elastic_geoip_service_tos=agree");
         try (XContentParser parser = XContentType.JSON.xContent().createParser(NamedXContentRegistry.EMPTY,
             DeprecationHandler.THROW_UNSUPPORTED_OPERATION, data)) {
             return (List<T>) parser.list();

modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTests.java

@@ -323,7 +323,8 @@ public void testUpdateDatabases() throws IOException {
         builder.map(Map.of("a", 2));
         builder.endArray();
         builder.close();
-        when(httpClient.getBytes("a.b?key=11111111-1111-1111-1111-111111111111")).thenReturn(baos.toByteArray());
+        when(httpClient.getBytes("a.b?key=11111111-1111-1111-1111-111111111111&elastic_geoip_service_tos=agree"))
+            .thenReturn(baos.toByteArray());
         Iterator<Map<String, Object>> it = maps.iterator();
         geoIpDownloader = new GeoIpDownloader(client, httpClient, clusterService, threadPool,
             Settings.builder().put(ENDPOINT_SETTING.getKey(), "a.b").build(),

test/fixtures/geoip-fixture/src/main/java/fixture/geoip/GeoIpHttpFixture.java

@@ -15,7 +15,6 @@
 import java.io.OutputStreamWriter;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
-import java.net.URI;
 import java.nio.charset.StandardCharsets;
 
 public class GeoIpHttpFixture {
@@ -26,6 +25,11 @@ public class GeoIpHttpFixture {
         String rawData = new String(GeoIpHttpFixture.class.getResourceAsStream("/data.json").readAllBytes(), StandardCharsets.UTF_8);
         this.server = HttpServer.create(new InetSocketAddress(InetAddress.getByName(args[0]), Integer.parseInt(args[1])), 0);
         this.server.createContext("/", exchange -> {
+            String query = exchange.getRequestURI().getQuery();
+            if (query.contains("elastic_geoip_service_tos=agree") == false) {
+                exchange.sendResponseHeaders(400, 0);
+                return;
+            }
             String data = rawData.replace("endpoint", "http://" + exchange.getRequestHeaders().getFirst("Host"));
             exchange.sendResponseHeaders(200, data.length());
             try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(exchange.getResponseBody()))) {