[Remote Store] Upload translog checkpoint as object metadata to translog

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add support for Azure Managed Identity in repository-azure ([#12423](https://github.com/opensearch-project/OpenSearch/issues/12423))
 - Add useCompoundFile index setting ([#13478](https://github.com/opensearch-project/OpenSearch/pull/13478))
 - Make outbound side of transport protocol dependent ([#13293](https://github.com/opensearch-project/OpenSearch/pull/13293))
+- [Remote Store] Upload translog checkpoint as object metadata to translog.tlog([#13637](https://github.com/opensearch-project/OpenSearch/pull/13637))
 
 ### Dependencies
 - Bump `com.github.spullara.mustache.java:compiler` from 0.9.10 to 0.9.13 ([#13329](https://github.com/opensearch-project/OpenSearch/pull/13329), [#13559](https://github.com/opensearch-project/OpenSearch/pull/13559))

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobContainer.java

@@ -78,7 +78,7 @@
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.blobstore.BlobStoreException;
 import org.opensearch.common.blobstore.DeleteResult;
-import org.opensearch.common.blobstore.FetchBlobResult;
+import org.opensearch.common.blobstore.InputStreamWithMetadata;
 import org.opensearch.common.blobstore.stream.read.ReadContext;
 import org.opensearch.common.blobstore.stream.write.WriteContext;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
@@ -143,9 +143,9 @@ public boolean blobExists(String blobName) {
 
     @ExperimentalApi
     @Override
-    public FetchBlobResult readBlobWithMetadata(String blobName) throws IOException {
+    public InputStreamWithMetadata readBlobWithMetadata(String blobName) throws IOException {
         S3RetryingInputStream s3RetryingInputStream = new S3RetryingInputStream(blobStore, buildKey(blobName));
-        return new FetchBlobResult(s3RetryingInputStream, s3RetryingInputStream.getMetadata());
+        return new InputStreamWithMetadata(s3RetryingInputStream, s3RetryingInputStream.getMetadata());
     }
 
     @Override

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3BlobStore.java

@@ -244,6 +244,11 @@ public Map<Metric, Map<String, Long>> extendedStats() {
         return extendedStats;
     }
 
+    @Override
+    public boolean isBlobMetadataEnabled() {
+        return true;
+    }
+
     public ObjectCannedACL getCannedACL() {
         return cannedACL;
     }

server/src/internalClusterTest/java/org/opensearch/remotemigration/RemoteMigrationIndexMetadataUpdateIT.java

@@ -512,5 +512,6 @@ private void assertCustomIndexMetadata(String index) {
         logger.info("---> Asserting custom index metadata");
         IndexMetadata iMd = internalCluster().client().admin().cluster().prepareState().get().getState().metadata().index(index);
         assertNotNull(iMd.getCustomData(IndexMetadata.REMOTE_STORE_CUSTOM_KEY));
+        assertNotNull(iMd.getCustomData(IndexMetadata.REMOTE_STORE_CUSTOM_KEY).get(IndexMetadata.TRANSLOG_METADATA_KEY));
     }
 }

server/src/internalClusterTest/java/org/opensearch/remotestore/translogmetadata/TranslogMetadataRemoteStoreIT.java

@@ -0,0 +1,124 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.remotestore.translogmetadata;
+
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.blobstore.BlobPath;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.remotestore.RemoteStoreIT;
+import org.opensearch.remotestore.translogmetadata.mocks.MockFsMetadataSupportedRepositoryPlugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.test.transport.MockTransportService;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.stream.Stream;
+
+import static org.opensearch.index.remote.RemoteStoreEnums.DataCategory.TRANSLOG;
+import static org.opensearch.index.remote.RemoteStoreEnums.DataType.METADATA;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class TranslogMetadataRemoteStoreIT extends RemoteStoreIT {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(MockTransportService.TestPlugin.class, MockFsMetadataSupportedRepositoryPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(
+                remoteStoreClusterSettings(
+                    REPOSITORY_NAME,
+                    segmentRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD,
+                    REPOSITORY_2_NAME,
+                    translogRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD
+                )
+            )
+            .build();
+    }
+
+    // Test local only translog files which are not uploaded to remote store (no metadata present in remote)
+    // Without the cleanup change in RemoteFsTranslog.createEmptyTranslog, this test fails with NPE.
+    public void testLocalOnlyTranslogCleanupOnNodeRestart() throws Exception {
+        clusterSettingsSuppliedByTest = true;
+
+        // Overriding settings to use AsyncMultiStreamBlobContainer
+        Settings settings = Settings.builder()
+            .put(super.nodeSettings(1))
+            .put(
+                remoteStoreClusterSettings(
+                    REPOSITORY_NAME,
+                    segmentRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD,
+                    REPOSITORY_2_NAME,
+                    translogRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD
+                )
+            )
+            .build();
+
+        internalCluster().startClusterManagerOnlyNode(settings);
+        String dataNode = internalCluster().startDataOnlyNode(settings);
+
+        // 1. Create index with 0 replica
+        createIndex(INDEX_NAME, remoteStoreIndexSettings(0, 10000L, -1));
+        ensureGreen(INDEX_NAME);
+
+        // 2. Index docs
+        int searchableDocs = 0;
+        for (int i = 0; i < randomIntBetween(1, 5); i++) {
+            indexBulk(INDEX_NAME, 15);
+            refresh(INDEX_NAME);
+            searchableDocs += 15;
+        }
+        indexBulk(INDEX_NAME, 15);
+
+        assertHitCount(client(dataNode).prepareSearch(INDEX_NAME).setSize(0).get(), searchableDocs);
+
+        // 3. Delete metadata from remote translog
+        String indexUUID = client().admin()
+            .indices()
+            .prepareGetSettings(INDEX_NAME)
+            .get()
+            .getSetting(INDEX_NAME, IndexMetadata.SETTING_INDEX_UUID);
+
+        String shardPath = getShardLevelBlobPath(client(), INDEX_NAME, BlobPath.cleanPath(), "0", TRANSLOG, METADATA).buildAsString();
+        Path translogMetaDataPath = Path.of(translogRepoPath + "/" + shardPath);
+
+        try (Stream<Path> files = Files.list(translogMetaDataPath)) {
+            files.forEach(p -> {
+                try {
+                    Files.delete(p);
+                } catch (IOException e) {
+                    // Ignore
+                }
+            });
+        }
+
+        internalCluster().restartNode(dataNode);
+
+        ensureGreen(INDEX_NAME);
+
+        assertHitCount(client(dataNode).prepareSearch(INDEX_NAME).setSize(0).get(), searchableDocs);
+        indexBulk(INDEX_NAME, 15);
+        refresh(INDEX_NAME);
+        assertHitCount(client(dataNode).prepareSearch(INDEX_NAME).setSize(0).get(), searchableDocs + 15);
+    }
+
+}

server/src/internalClusterTest/java/org/opensearch/remotestore/translogmetadata/TranslogMetadataRemoteStoreRestoreIT.java

@@ -0,0 +1,45 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.remotestore.translogmetadata;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.remotestore.RemoteStoreRestoreIT;
+import org.opensearch.remotestore.translogmetadata.mocks.MockFsMetadataSupportedRepositoryPlugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.test.transport.MockTransportService;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class TranslogMetadataRemoteStoreRestoreIT extends RemoteStoreRestoreIT {
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(MockTransportService.TestPlugin.class, MockFsMetadataSupportedRepositoryPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(
+                remoteStoreClusterSettings(
+                    REPOSITORY_NAME,
+                    segmentRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD,
+                    REPOSITORY_2_NAME,
+                    translogRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD
+                )
+            )
+            .build();
+    }
+}

server/src/internalClusterTest/java/org/opensearch/remotestore/translogmetadata/TranslogMetadataRemoteStoreStatsIT.java

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.remotestore.translogmetadata;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.remotestore.RemoteStoreStatsIT;
+import org.opensearch.remotestore.translogmetadata.mocks.MockFsMetadataSupportedRepositoryPlugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.test.transport.MockTransportService;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class TranslogMetadataRemoteStoreStatsIT extends RemoteStoreStatsIT {
+
+    @Override
+    public Collection<Class<? extends Plugin>> nodePlugins() {
+        return Arrays.asList(MockTransportService.TestPlugin.class, MockFsMetadataSupportedRepositoryPlugin.class);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(
+                remoteStoreClusterSettings(
+                    REPOSITORY_NAME,
+                    segmentRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD,
+                    REPOSITORY_2_NAME,
+                    translogRepoPath,
+                    MockFsMetadataSupportedRepositoryPlugin.TYPE_MD
+                )
+            )
+            .build();
+    }
+
+}