HotToWarmTieringService changes to tier shards

Signed-off-by: Neetika Singhal <neetiks@amazon.com>
neetikasinghal · Aug 7, 2024 · 7da1aa6 · 7da1aa6
1 parent c725431
commit 7da1aa6
Show file tree

Hide file tree

Showing 17 changed files with 1,099 additions and 58 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add ThreadContextPermission for stashAndMergeHeaders and stashWithOrigin ([#15039](https://github.com/opensearch-project/OpenSearch/pull/15039))
 - [Concurrent Segment Search] Support composite aggregations with scripting ([#15072](https://github.com/opensearch-project/OpenSearch/pull/15072))
 - Add `rangeQuery` and `regexpQuery` for `constant_keyword` field type ([#14711](https://github.com/opensearch-project/OpenSearch/pull/14711))
+- HotToWarmTieringService changes to tier shards ([#14891](https://github.com/opensearch-project/OpenSearch/pull/14891))
 
 ### Dependencies
 - Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))

diff --git a/server/src/internalClusterTest/java/org/opensearch/tiering/HotToWarmTieringServiceIT.java b/server/src/internalClusterTest/java/org/opensearch/tiering/HotToWarmTieringServiceIT.java
@@ -0,0 +1,114 @@
+/*
+ * 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.tiering;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.opensearch.action.admin.indices.get.GetIndexResponse;
+import org.opensearch.action.admin.indices.tiering.HotToWarmTieringAction;
+import org.opensearch.action.admin.indices.tiering.HotToWarmTieringResponse;
+import org.opensearch.action.admin.indices.tiering.TieringIndexRequest;
+import org.opensearch.action.search.SearchResponse;
+import org.opensearch.cluster.MockInternalClusterInfoService;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.core.common.unit.ByteSizeUnit;
+import org.opensearch.core.common.unit.ByteSizeValue;
+import org.opensearch.index.IndexModule;
+import org.opensearch.index.query.QueryBuilders;
+import org.opensearch.index.store.remote.file.CleanerDaemonThreadLeakFilter;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.junit.Before;
+
+import java.util.Map;
+
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertHitCount;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+
+@ThreadLeakFilters(filters = CleanerDaemonThreadLeakFilter.class)
+@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0, supportsDedicatedMasters = false)
+// Uncomment the below line to enable trace level logs for this test for better debugging
+// @TestLogging(reason = "Getting trace logs from tiering package", value =
+// "org.opensearch.tiering:TRACE,org.opensearch.cluster.routing.allocation.decider:TRACE")
+public class HotToWarmTieringServiceIT extends TieringBaseIntegTestCase {
+
+    protected static final String TEST_IDX_1 = "test-idx-1";
+    protected static final String TEST_IDX_2 = "test-idx-2";
+    protected static final int NUM_DOCS_IN_BULK = 10;
+    private static final long TOTAL_SPACE_BYTES = new ByteSizeValue(1000, ByteSizeUnit.KB).getBytes();
+
+    @Before
+    public void setup() {
+        internalCluster().startClusterManagerOnlyNode();
+    }
+
+    // waiting for the recovery pr to be merged in
+    @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/13647")
+    public void testTieringBasic() {
+        final int numReplicasIndex = 0;
+        internalCluster().ensureAtLeastNumDataNodes(1);
+        final Settings settings = Settings.builder()
+            .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+            .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, numReplicasIndex)
+            .put(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey(), IndexModule.DataLocalityType.FULL.name())
+            .build();
+
+        String[] indices = new String[] { TEST_IDX_1, TEST_IDX_2 };
+        for (String index : indices) {
+            assertAcked(client().admin().indices().prepareCreate(index).setSettings(settings).get());
+            ensureGreen(index);
+            // Ingesting some docs
+            indexBulk(index, NUM_DOCS_IN_BULK);
+            flushAndRefresh(index);
+            ensureGreen();
+            SearchResponse searchResponse = client().prepareSearch(index).setQuery(QueryBuilders.matchAllQuery()).get();
+            // Asserting that search returns same number of docs as ingested
+            assertHitCount(searchResponse, NUM_DOCS_IN_BULK);
+        }
+
+        // Spin up node having search role
+        internalCluster().ensureAtLeastNumSearchAndDataNodes(1);
+
+        final MockInternalClusterInfoService clusterInfoService = getMockInternalClusterInfoService();
+        clusterInfoService.setDiskUsageFunctionAndRefresh(
+            (discoveryNode, fsInfoPath) -> setDiskUsage(fsInfoPath, TOTAL_SPACE_BYTES, TOTAL_SPACE_BYTES)
+        );
+
+        TieringIndexRequest request = new TieringIndexRequest(TARGET_WARM_TIER, indices);
+        request.waitForCompletion(true);
+        HotToWarmTieringResponse response = client().admin().indices().execute(HotToWarmTieringAction.INSTANCE, request).actionGet();
+        assertAcked(response);
+        assertTrue(response.getFailedIndices().isEmpty());
+        assertTrue(response.isAcknowledged());
+        ensureGreen();
+        for (String index : indices) {
+            SearchResponse searchResponse = client().prepareSearch(index).setQuery(QueryBuilders.matchAllQuery()).get();
+            // Asserting that search returns same number of docs as ingested
+            assertHitCount(searchResponse, NUM_DOCS_IN_BULK);
+            GetIndexResponse getIndexResponse = client().admin().indices().prepareGetIndex().addIndices(index).get();
+            assertWarmSettings(getIndexResponse, index);
+            assertAcked(client().admin().indices().prepareDelete(index).get());
+        }
+    }
+
+    private void assertWarmSettings(GetIndexResponse response, String indexName) {
+        final Map<String, Settings> settings = response.settings();
+        assertThat(settings, notNullValue());
+        assertThat(settings.size(), equalTo(1));
+        Settings indexSettings = settings.get(indexName);
+        assertThat(indexSettings, notNullValue());
+        assertThat(
+            indexSettings.get(IndexModule.INDEX_STORE_LOCALITY_SETTING.getKey()),
+            equalTo(IndexModule.DataLocalityType.PARTIAL.name())
+        );
+        assertThat(indexSettings.get(IndexModule.INDEX_TIERING_STATE.getKey()), equalTo(IndexModule.TieringState.WARM.name()));
+    }
+}
diff --git a/server/src/internalClusterTest/java/org/opensearch/tiering/TieringBaseIntegTestCase.java b/server/src/internalClusterTest/java/org/opensearch/tiering/TieringBaseIntegTestCase.java
@@ -0,0 +1,102 @@
+/*
+ * 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.tiering;
+
+import org.opensearch.action.bulk.BulkRequest;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexRequest;
+import org.opensearch.cluster.ClusterInfoService;
+import org.opensearch.cluster.MockInternalClusterInfoService;
+import org.opensearch.common.UUIDs;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.FeatureFlags;
+import org.opensearch.monitor.fs.FsInfo;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+
+public class TieringBaseIntegTestCase extends OpenSearchIntegTestCase {
+
+    protected Path segmentRepoPath;
+    protected Path translogRepoPath;
+    Settings extraSettings = Settings.EMPTY;
+    private final List<String> documentKeys = List.of(
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5),
+        randomAlphaOfLength(5)
+    );
+
+    protected static final String REPOSITORY_NAME = "test-remote-store-repo";
+    protected static final String REPOSITORY_2_NAME = "test-remote-store-repo-2";
+    protected static final String TARGET_WARM_TIER = "warm";
+
+    /**
+     * Disable MockFSIndexStore plugin as it wraps the FSDirectory over a OpenSearchMockDirectoryWrapper which extends FilterDirectory (whereas FSDirectory extends BaseDirectory)
+     * As a result of this wrapping the local directory of Composite Directory does not satisfy the assertion that local directory must be of type FSDirectory
+     *
+     */
+    @Override
+    protected boolean addMockIndexStorePlugin() {
+        return false;
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Collections.singletonList(MockInternalClusterInfoService.TestPlugin.class);
+    }
+
+    @Override
+    protected Settings featureFlagSettings() {
+        Settings.Builder featureSettings = Settings.builder();
+        featureSettings.put(FeatureFlags.TIERED_REMOTE_INDEX, true);
+        return featureSettings.build();
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        if (segmentRepoPath == null || translogRepoPath == null) {
+            segmentRepoPath = randomRepoPath().toAbsolutePath();
+            translogRepoPath = randomRepoPath().toAbsolutePath();
+        }
+        return Settings.builder()
+            .put(super.nodeSettings(nodeOrdinal))
+            .put(extraSettings)
+            .put(remoteStoreClusterSettings(REPOSITORY_NAME, segmentRepoPath, REPOSITORY_2_NAME, translogRepoPath))
+            .put(REMOTE_CLUSTER_STATE_ENABLED_SETTING.getKey(), true)
+            .build();
+    }
+
+    protected BulkResponse indexBulk(String indexName, int numDocs) {
+        BulkRequest bulkRequest = new BulkRequest();
+        for (int i = 0; i < numDocs; i++) {
+            final IndexRequest request = client().prepareIndex(indexName)
+                .setId(UUIDs.randomBase64UUID())
+                .setSource(documentKeys.get(randomIntBetween(0, documentKeys.size() - 1)), randomAlphaOfLength(5))
+                .request();
+            bulkRequest.add(request);
+        }
+        return client().bulk(bulkRequest).actionGet();
+    }
+
+    protected MockInternalClusterInfoService getMockInternalClusterInfoService() {
+        return (MockInternalClusterInfoService) internalCluster().getCurrentClusterManagerNodeInstance(ClusterInfoService.class);
+    }
+
+    protected static FsInfo.Path setDiskUsage(FsInfo.Path original, long totalBytes, long freeBytes) {
+        return new FsInfo.Path(original.getPath(), original.getMount(), totalBytes, freeBytes, freeBytes);
+    }
+}