Fix GeoIpDownloader startup during rolling upgrade (#84000) (#84033)

If rolling upgrade was used from version prior GeoIPv2 (<`7.14`) then
geoip downloader wouldn't be started so no new databases were
downloaded. This is especially troubling in `8.x` as we no longer
provide default databases inside ES so after upgrade no geoip enrichment
can take place until downloader is started with workaround (setting
`ingest.geoip.downloader.enabled` to `false` and `true` again). This is
because logic that was used to lower number of requests / cluster update
listeners at the startup was too optimistic about order of actions / who
can be elected master at what time.  This change fixes that and also
cleans up logs when there are some ignorable errors and adds debug
logging on start and stop of the task to ease up troubleshooting. It
also adds rolling upgrade test to make sure the fix works.

Author: probakowski

docs/changelog/84000.yaml

@@ -0,0 +1,5 @@
+pr: 84000
+summary: Fix `GeoIpDownloader` startup during rolling upgrade
+area: Ingest
+type: bug
+issues: []

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

@@ -66,9 +66,15 @@ public class GeoIpDownloader extends AllocatedPersistentTask {
         Property.Dynamic,
         Property.NodeScope
     );
+
+    // for overriding in tests
+    private static final String DEFAULT_ENDPOINT = System.getProperty(
+        "ingest.geoip.downloader.endpoint.default",
+        "https://geoip.elastic.co/v1/database"
+    );
     public static final Setting<String> ENDPOINT_SETTING = Setting.simpleString(
         "ingest.geoip.downloader.endpoint",
-        "https://geoip.elastic.co/v1/database",
+        DEFAULT_ENDPOINT,
         Property.NodeScope
     );
 
@@ -258,6 +264,7 @@ void runDownloader() {
         try {
             updateDatabases();
         } catch (Exception e) {
+            stats = stats.failedDownload();
             logger.error("exception during geoip databases update", e);
         }
         try {

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

@@ -12,11 +12,13 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ResourceAlreadyExistsException;
 import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.cluster.ClusterChangedEvent;
 import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
@@ -29,6 +31,7 @@
 import org.elasticsearch.persistent.PersistentTasksService;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.RemoteTransportException;
 
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
@@ -128,14 +131,18 @@ public void clusterChanged(ClusterChangedEvent event) {
             // wait for state recovered
             return;
         }
-        // bootstrap downloader after first cluster start
+
+        DiscoveryNode masterNode = event.state().nodes().getMasterNode();
+        if (masterNode == null || masterNode.getVersion().before(Version.V_7_14_0)) {
+            // wait for master to be upgraded so it understands geoip task
+            return;
+        }
+
         clusterService.removeListener(this);
-        if (event.localNodeMaster()) {
-            if (ENABLED_SETTING.get(event.state().getMetadata().settings(), settings)) {
-                startTask(() -> clusterService.addListener(this));
-            } else {
-                stopTask(() -> clusterService.addListener(this));
-            }
+        if (ENABLED_SETTING.get(event.state().getMetadata().settings(), settings)) {
+            startTask(() -> clusterService.addListener(this));
+        } else {
+            stopTask(() -> clusterService.addListener(this));
         }
     }
 
@@ -144,8 +151,9 @@ private void startTask(Runnable onFailure) {
             GEOIP_DOWNLOADER,
             GEOIP_DOWNLOADER,
             new GeoIpTaskParams(),
-            ActionListener.wrap(r -> {}, e -> {
-                if (e instanceof ResourceAlreadyExistsException == false) {
+            ActionListener.wrap(r -> logger.debug("Started geoip downloader task"), e -> {
+                Throwable t = e instanceof RemoteTransportException ? e.getCause() : e;
+                if (t instanceof ResourceAlreadyExistsException == false) {
                     logger.error("failed to create geoip downloader task", e);
                     onFailure.run();
                 }
@@ -154,18 +162,23 @@ private void startTask(Runnable onFailure) {
     }
 
     private void stopTask(Runnable onFailure) {
-        ActionListener<PersistentTasksCustomMetadata.PersistentTask<?>> listener = ActionListener.wrap(r -> {}, e -> {
-            if (e instanceof ResourceNotFoundException == false) {
-                logger.error("failed to remove geoip downloader task", e);
-                onFailure.run();
+        ActionListener<PersistentTasksCustomMetadata.PersistentTask<?>> listener = ActionListener.wrap(
+            r -> logger.debug("Stopped geoip downloader task"),
+            e -> {
+                Throwable t = e instanceof RemoteTransportException ? e.getCause() : e;
+                if (t instanceof ResourceNotFoundException == false) {
+                    logger.error("failed to remove geoip downloader task", e);
+                    onFailure.run();
+                }
             }
-        });
+        );
         persistentTasksService.sendRemoveRequest(
             GEOIP_DOWNLOADER,
             ActionListener.runAfter(
                 listener,
                 () -> client.admin().indices().prepareDelete(DATABASES_INDEX).execute(ActionListener.wrap(rr -> {}, e -> {
-                    if (e instanceof ResourceNotFoundException == false) {
+                    Throwable t = e instanceof RemoteTransportException ? e.getCause() : e;
+                    if (t instanceof ResourceNotFoundException == false) {
                         logger.warn("failed to remove " + DATABASES_INDEX, e);
                     }
                 }))

x-pack/qa/rolling-upgrade/build.gradle

@@ -41,6 +41,13 @@ BuildParams.bwcVersions.withWireCompatible { bwcVersion, baseName ->
     versions = [oldVersion, project.version]
     numberOfNodes = 3
 
+    systemProperty 'ingest.geoip.downloader.enabled.default', 'true'
+    //we don't want to hit real service from each test
+    systemProperty 'ingest.geoip.downloader.endpoint.default', 'http://invalid.endpoint'
+    if (bwcVersion.onOrAfter('7.14.0')) {
+      setting 'ingest.geoip.downloader.endpoint', 'http://invalid.endpoint'
+    }
+
     setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
     setting 'path.repo', "['${buildDir}/cluster/shared/repo/${baseName}', '${searchableSnapshotRepository}']"
     setting 'xpack.license.self_generated.type', 'trial'

x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/GeoIpUpgradeIT.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.upgrades;
+
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.hamcrest.Matchers;
+
+import java.nio.charset.StandardCharsets;
+
+public class GeoIpUpgradeIT extends AbstractUpgradeTestCase {
+
+    public void testGeoIpDownloader() throws Exception {
+        if (CLUSTER_TYPE == ClusterType.UPGRADED) {
+            assertBusy(() -> {
+                Response response = client().performRequest(new Request("GET", "_cat/tasks"));
+                String tasks = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+                assertThat(tasks, Matchers.containsString("geoip-downloader"));
+            });
+            assertBusy(() -> {
+                Response response = client().performRequest(new Request("GET", "_ingest/geoip/stats"));
+                String tasks = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
+                assertThat(tasks, Matchers.containsString("failed_downloads\":1"));
+            });
+        }
+    }
+}