Open and Close PIT for Reindexing

modules/reindex-management/src/internalClusterTest/java/org/elasticsearch/reindex/management/ReindexCancelIT.java

@@ -91,6 +91,8 @@ public void setup() {
      * We test synchronous (<code>?wait_for_completion=true</code>) invocation of the _cancel endpoint in this test.
      */
     public void testCancelEndpointEndToEndSynchronously() throws Exception {
+        assumeFalse("scroll-based reindex uses a different code path", ReindexPlugin.REINDEX_PIT_SEARCH_ENABLED);
+
         final TaskId parentTaskId = startAsyncThrottledReindex();
 
         final TaskInfo running = getRunningTask(parentTaskId);
@@ -139,6 +141,8 @@ public void testCancelEndpointEndToEndSynchronously() throws Exception {
 
     /** Same test as above but calling _cancel asynchronously and wrapping assertions after cancellation in assertBusy. */
     public void testCancelEndpointEndToEndAsynchronously() throws Exception {
+        assumeFalse("scroll-based reindex uses a different code path", ReindexPlugin.REINDEX_PIT_SEARCH_ENABLED);
+
         final TaskId parentTaskId = startAsyncThrottledReindex();
 
         final TaskInfo running = getRunningTask(parentTaskId);

modules/reindex/src/internalClusterTest/java/org/elasticsearch/index/reindex/ReindexPluginMetricsIT.java

@@ -10,13 +10,16 @@
 package org.elasticsearch.index.reindex;
 
 import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.reindex.BulkIndexByScrollResponseMatcher;
 import org.elasticsearch.reindex.ReindexPlugin;
+import org.elasticsearch.reindex.Reindexer;
 import org.elasticsearch.reindex.TransportReindexAction;
 import org.elasticsearch.rest.root.MainRestPlugin;
 import org.elasticsearch.search.slice.SliceBuilder;
@@ -44,6 +47,7 @@
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertHitCount;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.notNullValue;
 
 @ESIntegTestCase.ClusterScope(numDataNodes = 0, numClientNodes = 0, scope = ESIntegTestCase.Scope.TEST)
 public class ReindexPluginMetricsIT extends ESIntegTestCase {
@@ -329,6 +333,124 @@ public void testReindexMetricsWithAutoSlices() throws Exception {
         });
     }
 
+    /**
+     * Verifies that remote reindex metrics record failures when the remote version lookup fails
+     * (e.g. connection refused, host unreachable).
+     */
+    public void testRemoteReindexVersionLookupFailureMetrics() throws Exception {
+        assumeTrue("PIT search must be enabled for remote version lookup path", ReindexPlugin.REINDEX_PIT_SEARCH_ENABLED);
+
+        final String dataNodeName = internalCluster().startNode();
+
+        // Use an invalid host so version lookup fails during connection
+        RemoteInfo invalidRemote = new RemoteInfo(
+            "http",
+            "invalid.invalid",
+            9200,
+            null,
+            new BytesArray("{\"match_all\":{}}"),
+            null,
+            null,
+            Map.of(),
+            TimeValue.timeValueMillis(100),
+            TimeValue.timeValueMillis(100)
+        );
+
+        final TestTelemetryPlugin testTelemetryPlugin = internalCluster().getInstance(PluginsService.class, dataNodeName)
+            .filterPlugins(TestTelemetryPlugin.class)
+            .findFirst()
+            .orElseThrow();
+
+        expectThrows(Exception.class, () -> reindex().source("source").setRemoteInfo(invalidRemote).destination("dest").get());
+
+        assertBusy(() -> {
+            testTelemetryPlugin.collect();
+            assertThat(testTelemetryPlugin.getLongHistogramMeasurement(REINDEX_TIME_HISTOGRAM).size(), equalTo(1));
+            List<Measurement> completions = testTelemetryPlugin.getLongCounterMeasurement(REINDEX_COMPLETION_COUNTER);
+            assertThat(completions.size(), equalTo(1));
+            assertThat(completions.getFirst().attributes().get(ATTRIBUTE_NAME_ERROR_TYPE), notNullValue());
+            assertThat(completions.getFirst().attributes().get(ATTRIBUTE_NAME_SOURCE), equalTo(ATTRIBUTE_VALUE_SOURCE_REMOTE));
+        });
+    }
+
+    /**
+     * Verifies that no reindex metrics are recorded when validation fails before the {@link Reindexer} runs
+     * (e.g. source index does not exist).
+     */
+    public void testLocalReindexValidationFailureNoMetrics() {
+        final String dataNodeName = internalCluster().startNode();
+
+        final TestTelemetryPlugin testTelemetryPlugin = internalCluster().getInstance(PluginsService.class, dataNodeName)
+            .filterPlugins(TestTelemetryPlugin.class)
+            .findFirst()
+            .orElseThrow();
+
+        expectThrows(Exception.class, () -> reindex().source("non_existent_index").destination("dest").get());
+
+        testTelemetryPlugin.collect();
+        assertThat(testTelemetryPlugin.getLongHistogramMeasurement(REINDEX_TIME_HISTOGRAM).size(), equalTo(0));
+        assertThat(testTelemetryPlugin.getLongCounterMeasurement(REINDEX_COMPLETION_COUNTER).size(), equalTo(0));
+    }
+
+    /**
+     * Verifies that local reindex metrics record failures when PIT open fails (e.g. source index is closed).
+     */
+    public void testLocalReindexPitOpenFailureMetrics() throws Exception {
+        assumeTrue("PIT search must be enabled for local PIT path", ReindexPlugin.REINDEX_PIT_SEARCH_ENABLED);
+
+        final String dataNodeName = internalCluster().startNode();
+
+        // Create and close the source index so PIT open fails (validation passes because index exists)
+        indexRandom(true, prepareIndex("source").setId("1").setSource("foo", "a"));
+        indicesAdmin().prepareClose("source").get();
+
+        final TestTelemetryPlugin testTelemetryPlugin = internalCluster().getInstance(PluginsService.class, dataNodeName)
+            .filterPlugins(TestTelemetryPlugin.class)
+            .findFirst()
+            .orElseThrow();
+
+        // Use STRICT_EXPAND_OPEN_CLOSED so validation resolves the closed index; PIT open will still fail
+        ReindexRequestBuilder builder = reindex().source("source").destination("dest");
+        builder.source().setIndicesOptions(IndicesOptions.STRICT_EXPAND_OPEN_CLOSED);
+        expectThrows(Exception.class, () -> builder.get());
+
+        assertBusy(() -> {
+            testTelemetryPlugin.collect();
+            assertThat(testTelemetryPlugin.getLongHistogramMeasurement(REINDEX_TIME_HISTOGRAM).size(), equalTo(1));
+            List<Measurement> completions = testTelemetryPlugin.getLongCounterMeasurement(REINDEX_COMPLETION_COUNTER);
+            assertThat(completions.size(), equalTo(1));
+            assertThat(completions.getFirst().attributes().get(ATTRIBUTE_NAME_ERROR_TYPE), notNullValue());
+            assertThat(completions.getFirst().attributes().get(ATTRIBUTE_NAME_SOURCE), equalTo(ATTRIBUTE_VALUE_SOURCE_LOCAL));
+        });
+    }
+
+    /**
+     * Verifies reindex metrics for a successful local reindex.
+     * Uses the scroll path when PIT is disabled, or the PIT path when PIT is enabled.
+     */
+    public void testLocalReindexMetrics() throws Exception {
+        final String dataNodeName = internalCluster().startNode();
+
+        indexRandom(true, prepareIndex("source").setId("1").setSource("foo", "a"), prepareIndex("source").setId("2").setSource("foo", "b"));
+        assertHitCount(prepareSearch("source").setSize(0), 2);
+
+        final TestTelemetryPlugin testTelemetryPlugin = internalCluster().getInstance(PluginsService.class, dataNodeName)
+            .filterPlugins(TestTelemetryPlugin.class)
+            .findFirst()
+            .orElseThrow();
+
+        reindex().source("source").destination("dest").get();
+
+        assertBusy(() -> {
+            testTelemetryPlugin.collect();
+            assertThat(testTelemetryPlugin.getLongHistogramMeasurement(REINDEX_TIME_HISTOGRAM).size(), equalTo(1));
+            List<Measurement> completions = testTelemetryPlugin.getLongCounterMeasurement(REINDEX_COMPLETION_COUNTER);
+            assertThat(completions.size(), equalTo(1));
+            assertNull(completions.getFirst().attributes().get(ATTRIBUTE_NAME_ERROR_TYPE));
+            assertThat(completions.getFirst().attributes().get(ATTRIBUTE_NAME_SOURCE), equalTo(ATTRIBUTE_VALUE_SOURCE_LOCAL));
+        });
+    }
+
     public void testDeleteByQueryMetrics() throws Exception {
         final String dataNodeName = internalCluster().startNode();