Add composite directory factory (opensearch-project#17988)

Signed-off-by: Shreyansh Ray <[email protected]>

Author: rayshrey

CHANGELOG.md

@@ -5,7 +5,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 2.x]
 ### Added
+<<<<<<< HEAD
 - Renaming the node role search to warm ([#17573](https://github.com/opensearch-project/OpenSearch/pull/17573))
+=======
+- [Rule based auto-tagging] Add get rule API ([#17336](https://github.com/opensearch-project/OpenSearch/pull/17336))
+- Add multi-threaded writer support in pull-based ingestion ([#17912](https://github.com/opensearch-project/OpenSearch/pull/17912))
+- Unset discovery nodes for every transport node actions request ([#17682](https://github.com/opensearch-project/OpenSearch/pull/17682))
+- Implement parallel shard refresh behind cluster settings ([#17782](https://github.com/opensearch-project/OpenSearch/pull/17782))
+- Bump OpenSearch Core main branch to 3.0.0 ([#18039](https://github.com/opensearch-project/OpenSearch/pull/18039))
+- Update API of Message in index to add the timestamp for lag calculation in ingestion polling ([#17977](https://github.com/opensearch-project/OpenSearch/pull/17977/))
+- Enabled default throttling for all tasks submitted to cluster manager ([#17711](https://github.com/opensearch-project/OpenSearch/pull/17711))
+- Add composite directory factory ([#17988](https://github.com/opensearch-project/OpenSearch/pull/17988))
+
+### Changed
+- Change the default max header size from 8KB to 16KB. ([#18024](https://github.com/opensearch-project/OpenSearch/pull/18024))
+- Avoid invalid retries in multiple replicas when querying [#17370](https://github.com/opensearch-project/OpenSearch/pull/17370)
+* Enable concurrent_segment_search auto mode by default[#17978](https://github.com/opensearch-project/OpenSearch/pull/17978)
+- Skip approximation when `track_total_hits` is set to `true` [#18017](https://github.com/opensearch-project/OpenSearch/pull/18017)
+
+>>>>>>> d744b40ef78 (Add composite directory factory (#17988))
 ### Dependencies
 - Bump `netty` from 4.1.118.Final to 4.1.121.Final ([#18192](https://github.com/opensearch-project/OpenSearch/pull/18192))
 

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -191,6 +191,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 MapperService.INDEX_MAPPING_FIELD_NAME_LENGTH_LIMIT_SETTING,
                 BitsetFilterCache.INDEX_LOAD_RANDOM_ACCESS_FILTERS_EAGERLY_SETTING,
                 IndexModule.INDEX_STORE_TYPE_SETTING,
+                IndexModule.INDEX_COMPOSITE_STORE_TYPE_SETTING,
                 IndexModule.INDEX_STORE_PRE_LOAD_SETTING,
                 IndexModule.INDEX_STORE_HYBRID_MMAP_EXTENSIONS,
                 IndexModule.INDEX_STORE_HYBRID_NIO_EXTENSIONS,

server/src/main/java/org/opensearch/index/IndexModule.java

@@ -77,6 +77,7 @@
 import org.opensearch.index.shard.IndexingOperationListener;
 import org.opensearch.index.shard.SearchOperationListener;
 import org.opensearch.index.similarity.SimilarityService;
+import org.opensearch.index.store.DefaultCompositeDirectoryFactory;
 import org.opensearch.index.store.FsDirectoryFactory;
 import org.opensearch.index.store.remote.directory.RemoteSnapshotDirectoryFactory;
 import org.opensearch.index.store.remote.filecache.FileCache;
@@ -134,6 +135,8 @@ public final class IndexModule {
     public static final Setting<Boolean> NODE_STORE_ALLOW_MMAP = Setting.boolSetting("node.store.allow_mmap", true, Property.NodeScope);
 
     private static final FsDirectoryFactory DEFAULT_DIRECTORY_FACTORY = new FsDirectoryFactory();
+    private static final IndexStorePlugin.CompositeDirectoryFactory DEFAULT_COMPOSITE_DIRECTORY_FACTORY =
+        new DefaultCompositeDirectoryFactory();
 
     private static final IndexStorePlugin.RecoveryStateFactory DEFAULT_RECOVERY_STATE_FACTORY = RecoveryState::new;
 
@@ -145,6 +148,14 @@ public final class IndexModule {
         Property.NodeScope
     );
 
+    public static final Setting<String> INDEX_COMPOSITE_STORE_TYPE_SETTING = new Setting<>(
+        "index.composite_store.type",
+        "default",
+        Function.identity(),
+        Property.IndexScope,
+        Property.NodeScope
+    );
+
     /**
      * Index setting which used to determine how the data is cached locally fully or partially.
      */
@@ -315,6 +326,7 @@ public Iterator<Setting<?>> settings() {
     private final Set<IndexEventListener> indexEventListeners = new HashSet<>();
     private final Map<String, TriFunction<Settings, Version, ScriptService, Similarity>> similarities = new HashMap<>();
     private final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories;
+    private final Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories;
     private final SetOnce<BiFunction<IndexSettings, IndicesQueryCache, QueryCache>> forceQueryCacheProvider = new SetOnce<>();
     private final List<SearchOperationListener> searchOperationListeners = new ArrayList<>();
     private final List<IndexingOperationListener> indexOperationListeners = new ArrayList<>();
@@ -340,6 +352,7 @@ public IndexModule(
         final EngineFactory engineFactory,
         final EngineConfigFactory engineConfigFactory,
         final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories,
+        final Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories,
         final BooleanSupplier allowExpensiveQueries,
         final IndexNameExpressionResolver expressionResolver,
         final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories,
@@ -353,6 +366,7 @@ public IndexModule(
         this.searchOperationListeners.add(new SearchSlowLog(indexSettings));
         this.indexOperationListeners.add(new IndexingSlowLog(indexSettings));
         this.directoryFactories = Collections.unmodifiableMap(directoryFactories);
+        this.compositeDirectoryFactories = Collections.unmodifiableMap(compositeDirectoryFactories);
         this.allowExpensiveQueries = allowExpensiveQueries;
         this.expressionResolver = expressionResolver;
         this.recoveryStateFactories = recoveryStateFactories;
@@ -401,6 +415,7 @@ public IndexModule(
             engineFactory,
             engineConfigFactory,
             directoryFactories,
+            Collections.emptyMap(),
             allowExpensiveQueries,
             expressionResolver,
             recoveryStateFactories,
@@ -788,6 +803,10 @@ public IndexService newIndexService(
             .get() == null ? (shard) -> null : indexReaderWrapper.get();
         eventListener.beforeIndexCreated(indexSettings.getIndex(), indexSettings.getSettings());
         final IndexStorePlugin.DirectoryFactory directoryFactory = getDirectoryFactory(indexSettings, directoryFactories);
+        final IndexStorePlugin.CompositeDirectoryFactory compositeDirectoryFactory = getCompositeDirectoryFactory(
+            indexSettings,
+            compositeDirectoryFactories
+        );
         final IndexStorePlugin.RecoveryStateFactory recoveryStateFactory = getRecoveryStateFactory(indexSettings, recoveryStateFactories);
         QueryCache queryCache = null;
         IndexAnalyzers indexAnalyzers = null;
@@ -824,6 +843,7 @@ public IndexService newIndexService(
                 client,
                 queryCache,
                 directoryFactory,
+                compositeDirectoryFactory,
                 remoteDirectoryFactory,
                 eventListener,
                 readerWrapperFactory,
@@ -885,6 +905,23 @@ private static IndexStorePlugin.DirectoryFactory getDirectoryFactory(
         return factory;
     }
 
+    private static IndexStorePlugin.CompositeDirectoryFactory getCompositeDirectoryFactory(
+        final IndexSettings indexSettings,
+        final Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories
+    ) {
+        final String compositeStoreType = indexSettings.getValue(INDEX_COMPOSITE_STORE_TYPE_SETTING);
+        final IndexStorePlugin.CompositeDirectoryFactory factory;
+        if (compositeStoreType.isEmpty() || compositeDirectoryFactories.isEmpty()) {
+            factory = DEFAULT_COMPOSITE_DIRECTORY_FACTORY;
+        } else {
+            factory = compositeDirectoryFactories.get(compositeStoreType);
+            if (factory == null) {
+                throw new IllegalArgumentException("Unknown composite store type [" + compositeStoreType + "]");
+            }
+        }
+        return factory;
+    }
+
     private static IndexStorePlugin.RecoveryStateFactory getRecoveryStateFactory(
         final IndexSettings indexSettings,
         final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories

server/src/main/java/org/opensearch/index/IndexService.java

@@ -93,7 +93,6 @@
 import org.opensearch.index.shard.ShardNotInPrimaryModeException;
 import org.opensearch.index.shard.ShardPath;
 import org.opensearch.index.similarity.SimilarityService;
-import org.opensearch.index.store.CompositeDirectory;
 import org.opensearch.index.store.RemoteSegmentStoreDirectoryFactory;
 import org.opensearch.index.store.Store;
 import org.opensearch.index.store.remote.filecache.FileCache;
@@ -153,6 +152,7 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final NodeEnvironment nodeEnv;
     private final ShardStoreDeleter shardStoreDeleter;
     private final IndexStorePlugin.DirectoryFactory directoryFactory;
+    private final IndexStorePlugin.CompositeDirectoryFactory compositeDirectoryFactory;
     private final IndexStorePlugin.DirectoryFactory remoteDirectoryFactory;
     private final IndexStorePlugin.RecoveryStateFactory recoveryStateFactory;
     private final CheckedFunction<DirectoryReader, DirectoryReader, IOException> readerWrapper;
@@ -216,6 +216,7 @@ public IndexService(
         Client client,
         QueryCache queryCache,
         IndexStorePlugin.DirectoryFactory directoryFactory,
+        IndexStorePlugin.CompositeDirectoryFactory compositeDirectoryFactory,
         IndexStorePlugin.DirectoryFactory remoteDirectoryFactory,
         IndexEventListener eventListener,
         Function<IndexService, CheckedFunction<DirectoryReader, DirectoryReader, IOException>> wrapperFactory,
@@ -296,6 +297,7 @@ public IndexService(
         this.eventListener = eventListener;
         this.nodeEnv = nodeEnv;
         this.directoryFactory = directoryFactory;
+        this.compositeDirectoryFactory = compositeDirectoryFactory;
         this.remoteDirectoryFactory = remoteDirectoryFactory;
         this.recoveryStateFactory = recoveryStateFactory;
         this.engineFactory = Objects.requireNonNull(engineFactory);
@@ -454,6 +456,7 @@ public IndexService(
             client,
             queryCache,
             directoryFactory,
+            null,
             remoteDirectoryFactory,
             eventListener,
             wrapperFactory,
@@ -726,8 +729,7 @@ protected void closeInternal() {
             if (FeatureFlags.isEnabled(FeatureFlags.WRITABLE_WARM_INDEX_SETTING) &&
             // TODO : Need to remove this check after support for hot indices is added in Composite Directory
                 this.indexSettings.isWarmIndex()) {
-                Directory localDirectory = directoryFactory.newDirectory(this.indexSettings, path);
-                directory = new CompositeDirectory(localDirectory, remoteDirectory, fileCache);
+                directory = compositeDirectoryFactory.newDirectory(this.indexSettings, path, directoryFactory, remoteDirectory, fileCache);
             } else {
                 directory = directoryFactory.newDirectory(this.indexSettings, path);
             }

server/src/main/java/org/opensearch/index/store/CompositeDirectory.java

@@ -53,10 +53,10 @@
 @ExperimentalApi
 public class CompositeDirectory extends FilterDirectory {
     private static final Logger logger = LogManager.getLogger(CompositeDirectory.class);
-    private final FSDirectory localDirectory;
-    private final RemoteSegmentStoreDirectory remoteDirectory;
-    private final FileCache fileCache;
-    private final TransferManager transferManager;
+    protected final FSDirectory localDirectory;
+    protected final RemoteSegmentStoreDirectory remoteDirectory;
+    protected final FileCache fileCache;
+    protected final TransferManager transferManager;
 
     /**
      * Constructor to initialise the composite directory
@@ -97,7 +97,7 @@ private String[] listLocalFiles() throws IOException {
      * @return A list of file names, including the original file (if present) and all its block files.
      * @throws IOException in case of I/O error while listing files.
      */
-    private List<String> listBlockFiles(String fileName) throws IOException {
+    protected List<String> listBlockFiles(String fileName) throws IOException {
         return Stream.of(listLocalFiles())
             .filter(file -> file.equals(fileName) || file.startsWith(fileName + FileTypeUtils.BLOCK_FILE_IDENTIFIER))
             .collect(Collectors.toList());
@@ -384,7 +384,7 @@ private String[] getRemoteFiles() throws IOException {
         return remoteFiles;
     }
 
-    private void cacheFile(String name) throws IOException {
+    protected void cacheFile(String name) throws IOException {
         Path filePath = getFilePath(name);
         // put will increase the refCount for the path, making sure it is not evicted, will decrease the ref after it is uploaded to Remote
         // so that it can be evicted after that

server/src/main/java/org/opensearch/index/store/DefaultCompositeDirectoryFactory.java

@@ -0,0 +1,40 @@
+/*
+ * 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.index.store;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.store.Directory;
+import org.opensearch.index.IndexSettings;
+import org.opensearch.index.shard.ShardPath;
+import org.opensearch.index.store.remote.filecache.FileCache;
+import org.opensearch.plugins.IndexStorePlugin;
+
+import java.io.IOException;
+
+/**
+ * Default composite directory factory
+ */
+public class DefaultCompositeDirectoryFactory implements IndexStorePlugin.CompositeDirectoryFactory {
+
+    private static final Logger logger = LogManager.getLogger(DefaultCompositeDirectoryFactory.class);
+
+    @Override
+    public Directory newDirectory(
+        IndexSettings indexSettings,
+        ShardPath shardPath,
+        IndexStorePlugin.DirectoryFactory localDirectoryFactory,
+        Directory remoteDirectory,
+        FileCache fileCache
+    ) throws IOException {
+        logger.trace("Creating composite directory from core - Default CompositeDirectoryFactory");
+        Directory localDirectory = localDirectoryFactory.newDirectory(indexSettings, shardPath);
+        return new CompositeDirectory(localDirectory, remoteDirectory, fileCache);
+    }
+}

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -344,6 +344,7 @@ public class IndicesService extends AbstractLifecycleComponent
     private final MetaStateService metaStateService;
     private final Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders;
     private final Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories;
+    private final Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories;
     private final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories;
     final AbstractRefCounted indicesRefCount; // pkg-private for testing
     private final CountDownLatch closeLatch = new CountDownLatch(1);
@@ -390,6 +391,7 @@ public IndicesService(
         MetaStateService metaStateService,
         Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders,
         Map<String, IndexStorePlugin.DirectoryFactory> directoryFactories,
+        Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories,
         ValuesSourceRegistry valuesSourceRegistry,
         Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories,
         IndexStorePlugin.DirectoryFactory remoteDirectoryFactory,
@@ -452,6 +454,7 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
         this.engineFactoryProviders = engineFactoryProviders;
 
         this.directoryFactories = directoryFactories;
+        this.compositeDirectoryFactories = compositeDirectoryFactories;
         this.recoveryStateFactories = recoveryStateFactories;
         // doClose() is called when shutting down a node, yet there might still be ongoing requests
         // that we need to wait for before closing some resources such as the caches. In order to
@@ -628,6 +631,7 @@ public IndicesService(
             metaStateService,
             engineFactoryProviders,
             directoryFactories,
+            Collections.emptyMap(),
             valuesSourceRegistry,
             recoveryStateFactories,
             remoteDirectoryFactory,
@@ -1022,6 +1026,7 @@ private synchronized IndexService createIndexService(
             getEngineFactory(idxSettings),
             getEngineConfigFactory(idxSettings),
             directoryFactories,
+            compositeDirectoryFactories,
             () -> allowExpensiveQueries,
             indexNameExpressionResolver,
             recoveryStateFactories,
@@ -1115,6 +1120,7 @@ public synchronized MapperService createIndexMapperService(IndexMetadata indexMe
             getEngineFactory(idxSettings),
             getEngineConfigFactory(idxSettings),
             directoryFactories,
+            compositeDirectoryFactories,
             () -> allowExpensiveQueries,
             indexNameExpressionResolver,
             recoveryStateFactories,

server/src/main/java/org/opensearch/node/Node.java

@@ -156,6 +156,7 @@
 import org.opensearch.index.recovery.RemoteStoreRestoreService;
 import org.opensearch.index.remote.RemoteIndexPathUploader;
 import org.opensearch.index.remote.RemoteStoreStatsTrackerFactory;
+import org.opensearch.index.store.DefaultCompositeDirectoryFactory;
 import org.opensearch.index.store.IndexStoreListener;
 import org.opensearch.index.store.RemoteSegmentStoreDirectoryFactory;
 import org.opensearch.index.store.remote.filecache.FileCache;
@@ -879,6 +880,22 @@ protected Node(
                 });
             directoryFactories.putAll(builtInDirectoryFactories);
 
+            final Map<String, IndexStorePlugin.CompositeDirectoryFactory> compositeDirectoryFactories = new HashMap<>();
+            pluginsService.filterPlugins(IndexStorePlugin.class)
+                .stream()
+                .map(IndexStorePlugin::getCompositeDirectoryFactories)
+                .flatMap(m -> m.entrySet().stream())
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
+                .forEach((k, v) -> {
+                    if (k.equals("default")) {
+                        throw new IllegalStateException(
+                            "registered composite index store type [" + k + "] conflicts with a built-in default type"
+                        );
+                    }
+                    compositeDirectoryFactories.put(k, v);
+                });
+            compositeDirectoryFactories.put("default", new DefaultCompositeDirectoryFactory());
+
             final Map<String, IndexStorePlugin.RecoveryStateFactory> recoveryStateFactories = pluginsService.filterPlugins(
                 IndexStorePlugin.class
             )
@@ -937,6 +954,7 @@ protected Node(
                 metaStateService,
                 engineFactoryProviders,
                 Map.copyOf(directoryFactories),
+                Map.copyOf(compositeDirectoryFactories),
                 searchModule.getValuesSourceRegistry(),
                 recoveryStateFactories,
                 remoteDirectoryFactory,