Skip to content

Commit ad8915f

Browse files
jasontedorjasontedor
committed
Enable engine factory to be pluggable (#31183)
This commit enables the engine factory to be pluggable based on index settings used when creating the index service for an index.
1 parent 270b8ee commit ad8915f

File tree

16 files changed

+349
-74
lines changed

16 files changed

+349
-74
lines changed

‎server/src/main/java/org/elasticsearch/index/IndexModule.java‎

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,8 @@
1919

2020
package org.elasticsearch.index;
2121

22-
import org.apache.lucene.search.similarities.Similarity;
2322
import org.apache.lucene.search.similarities.BM25Similarity;
23+
import org.apache.lucene.search.similarities.Similarity;
2424
import org.apache.lucene.util.SetOnce;
2525
import org.elasticsearch.Version;
2626
import org.elasticsearch.client.Client;
@@ -60,6 +60,7 @@
6060
import java.util.List;
6161
import java.util.Locale;
6262
import java.util.Map;
63+
import java.util.Objects;
6364
import java.util.Set;
6465
import java.util.concurrent.atomic.AtomicBoolean;
6566
import java.util.function.BiFunction;
@@ -104,8 +105,7 @@ public final class IndexModule {
104105

105106
private final IndexSettings indexSettings;
106107
private final AnalysisRegistry analysisRegistry;
107-
// pkg private so tests can mock
108-
final SetOnce<EngineFactory> engineFactory = new SetOnce<>();
108+
private final EngineFactory engineFactory;
109109
private SetOnce<IndexSearcherWrapperFactory> indexSearcherWrapper = new SetOnce<>();
110110
private final Set<IndexEventListener> indexEventListeners = new HashSet<>();
111111
private final Map<String, TriFunction<Settings, Version, ScriptService, Similarity>> similarities = new HashMap<>();
@@ -115,9 +115,18 @@ public final class IndexModule {
115115
private final List<IndexingOperationListener> indexOperationListeners = new ArrayList<>();
116116
private final AtomicBoolean frozen = new AtomicBoolean(false);
117117

118-
public IndexModule(IndexSettings indexSettings, AnalysisRegistry analysisRegistry) {
118+
/**
119+
* Construct the index module for the index with the specified index settings. The index module contains extension points for plugins
120+
* via {@link org.elasticsearch.plugins.PluginsService#onIndexModule(IndexModule)}.
121+
*
122+
* @param indexSettings the index settings
123+
* @param analysisRegistry the analysis registry
124+
* @param engineFactory the engine factory
125+
*/
126+
public IndexModule(final IndexSettings indexSettings, final AnalysisRegistry analysisRegistry, final EngineFactory engineFactory) {
119127
this.indexSettings = indexSettings;
120128
this.analysisRegistry = analysisRegistry;
129+
this.engineFactory = Objects.requireNonNull(engineFactory);
121130
this.searchOperationListeners.add(new SearchSlowLog(indexSettings));
122131
this.indexOperationListeners.add(new IndexingSlowLog(indexSettings));
123132
}
@@ -158,6 +167,15 @@ public Index getIndex() {
158167
return indexSettings.getIndex();
159168
}
160169

170+
/**
171+
* The engine factory provided during construction of this index module.
172+
*
173+
* @return the engine factory
174+
*/
175+
EngineFactory getEngineFactory() {
176+
return engineFactory;
177+
}
178+
161179
/**
162180
* Adds an {@link IndexEventListener} for this index. All listeners added here
163181
* are maintained for the entire index lifecycle on this node. Once an index is closed or deleted these
@@ -364,7 +382,7 @@ public IndexService newIndexService(
364382
}
365383
return new IndexService(indexSettings, environment, xContentRegistry,
366384
new SimilarityService(indexSettings, scriptService, similarities),
367-
shardStoreDeleter, analysisRegistry, engineFactory.get(), circuitBreakerService, bigArrays, threadPool, scriptService,
385+
shardStoreDeleter, analysisRegistry, engineFactory, circuitBreakerService, bigArrays, threadPool, scriptService,
368386
client, queryCache, store, eventListener, searcherWrapperFactory, mapperRegistry,
369387
indicesFieldDataCache, searchOperationListeners, indexOperationListeners, namedWriteableRegistry);
370388
}

‎server/src/main/java/org/elasticsearch/index/IndexService.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ public IndexService(
138138
SimilarityService similarityService,
139139
ShardStoreDeleter shardStoreDeleter,
140140
AnalysisRegistry registry,
141-
@Nullable EngineFactory engineFactory,
141+
EngineFactory engineFactory,
142142
CircuitBreakerService circuitBreakerService,
143143
BigArrays bigArrays,
144144
ThreadPool threadPool,
@@ -187,7 +187,7 @@ public IndexService(
187187
this.warmer = new IndexWarmer(indexSettings.getSettings(), threadPool, indexFieldData,
188188
bitsetFilterCache.createListener(threadPool));
189189
this.indexCache = new IndexCache(indexSettings, queryCache, bitsetFilterCache);
190-
this.engineFactory = engineFactory;
190+
this.engineFactory = Objects.requireNonNull(engineFactory);
191191
// initialize this last -- otherwise if the wrapper requires any other member to be non-null we fail with an NPE
192192
this.searcherWrapper = wrapperFactory.newWrapper(this);
193193
this.searchOperationListeners = Collections.unmodifiableList(searchOperationListeners);
@@ -659,9 +659,9 @@ public interface ShardStoreDeleter {
659659
void addPendingDelete(ShardId shardId, IndexSettings indexSettings);
660660
}
661661

662-
final EngineFactory getEngineFactory() {
662+
public final EngineFactory getEngineFactory() {
663663
return engineFactory;
664-
} // pkg private for testing
664+
}
665665

666666
final IndexSearcherWrapper getSearcherWrapper() {
667667
return searcherWrapper;

‎server/src/main/java/org/elasticsearch/index/shard/IndexShard.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,6 @@
8484
import org.elasticsearch.index.engine.EngineException;
8585
import org.elasticsearch.index.engine.EngineFactory;
8686
import org.elasticsearch.index.engine.InternalEngine;
87-
import org.elasticsearch.index.engine.InternalEngineFactory;
8887
import org.elasticsearch.index.engine.RefreshFailedEngineException;
8988
import org.elasticsearch.index.engine.Segment;
9089
import org.elasticsearch.index.engine.SegmentsStats;
@@ -196,7 +195,7 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl
196195
protected volatile IndexShardState state;
197196
protected volatile long primaryTerm;
198197
protected final AtomicReference<Engine> currentEngineReference = new AtomicReference<>();
199-
protected final EngineFactory engineFactory;
198+
final EngineFactory engineFactory;
200199

201200
private final IndexingOperationListener indexingOperationListeners;
202201
private final Runnable globalCheckpointSyncer;
@@ -266,7 +265,7 @@ public IndexShard(
266265
this.warmer = warmer;
267266
this.similarityService = similarityService;
268267
Objects.requireNonNull(store, "Store must be provided to the index shard");
269-
this.engineFactory = engineFactory == null ? new InternalEngineFactory() : engineFactory;
268+
this.engineFactory = Objects.requireNonNull(engineFactory);
270269
this.store = store;
271270
this.indexSortSupplier = indexSortSupplier;
272271
this.indexEventListener = indexEventListener;

‎server/src/main/java/org/elasticsearch/indices/IndicesModule.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import org.elasticsearch.common.geo.ShapesAvailability;
2828
import org.elasticsearch.common.inject.AbstractModule;
2929
import org.elasticsearch.common.io.stream.NamedWriteableRegistry.Entry;
30+
import org.elasticsearch.index.IndexSettings;
31+
import org.elasticsearch.index.engine.EngineFactory;
3032
import org.elasticsearch.index.mapper.AllFieldMapper;
3133
import org.elasticsearch.index.mapper.BinaryFieldMapper;
3234
import org.elasticsearch.index.mapper.BooleanFieldMapper;
@@ -63,10 +65,12 @@
6365
import org.elasticsearch.plugins.MapperPlugin;
6466

6567
import java.util.ArrayList;
68+
import java.util.Collection;
6669
import java.util.Collections;
6770
import java.util.LinkedHashMap;
6871
import java.util.List;
6972
import java.util.Map;
73+
import java.util.Optional;
7074
import java.util.Set;
7175
import java.util.function.Function;
7276
import java.util.function.Predicate;
@@ -239,4 +243,9 @@ protected void configure() {
239243
public MapperRegistry getMapperRegistry() {
240244
return mapperRegistry;
241245
}
246+
247+
public Collection<Function<IndexSettings, Optional<EngineFactory>>> getEngineFactories() {
248+
return Collections.emptyList();
249+
}
250+
242251
}

‎server/src/main/java/org/elasticsearch/indices/IndicesService.java‎

Lines changed: 42 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import org.apache.lucene.store.AlreadyClosedException;
2626
import org.apache.lucene.store.LockObtainFailedException;
2727
import org.apache.lucene.util.CollectionUtil;
28-
import org.elasticsearch.core.internal.io.IOUtils;
2928
import org.apache.lucene.util.RamUsageEstimator;
3029
import org.elasticsearch.ElasticsearchException;
3130
import org.elasticsearch.ResourceAlreadyExistsException;
@@ -45,6 +44,7 @@
4544
import org.elasticsearch.common.Nullable;
4645
import org.elasticsearch.common.breaker.CircuitBreaker;
4746
import org.elasticsearch.common.bytes.BytesReference;
47+
import org.elasticsearch.common.collect.Tuple;
4848
import org.elasticsearch.common.component.AbstractLifecycleComponent;
4949
import org.elasticsearch.common.io.FileSystemUtils;
5050
import org.elasticsearch.common.io.stream.BytesStreamOutput;
@@ -67,6 +67,7 @@
6767
import org.elasticsearch.common.xcontent.XContentFactory;
6868
import org.elasticsearch.common.xcontent.XContentParser;
6969
import org.elasticsearch.common.xcontent.XContentType;
70+
import org.elasticsearch.core.internal.io.IOUtils;
7071
import org.elasticsearch.env.NodeEnvironment;
7172
import org.elasticsearch.env.ShardLock;
7273
import org.elasticsearch.env.ShardLockObtainFailedException;
@@ -79,6 +80,8 @@
7980
import org.elasticsearch.index.IndexSettings;
8081
import org.elasticsearch.index.analysis.AnalysisRegistry;
8182
import org.elasticsearch.index.cache.request.ShardRequestCache;
83+
import org.elasticsearch.index.engine.EngineFactory;
84+
import org.elasticsearch.index.engine.InternalEngineFactory;
8285
import org.elasticsearch.index.fielddata.IndexFieldDataCache;
8386
import org.elasticsearch.index.flush.FlushStats;
8487
import org.elasticsearch.index.get.GetStats;
@@ -116,10 +119,14 @@
116119
import java.io.IOException;
117120
import java.nio.file.Files;
118121
import java.util.ArrayList;
122+
import java.util.Collection;
119123
import java.util.HashMap;
120124
import java.util.Iterator;
121125
import java.util.List;
126+
import java.util.Locale;
122127
import java.util.Map;
128+
import java.util.Objects;
129+
import java.util.Optional;
123130
import java.util.Set;
124131
import java.util.concurrent.CountDownLatch;
125132
import java.util.concurrent.ExecutorService;
@@ -172,6 +179,7 @@ public class IndicesService extends AbstractLifecycleComponent
172179
private final IndicesRequestCache indicesRequestCache;
173180
private final IndicesQueryCache indicesQueryCache;
174181
private final MetaStateService metaStateService;
182+
private final Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders;
175183

176184
@Override
177185
protected void doStart() {
@@ -183,7 +191,8 @@ public IndicesService(Settings settings, PluginsService pluginsService, NodeEnvi
183191
AnalysisRegistry analysisRegistry, IndexNameExpressionResolver indexNameExpressionResolver,
184192
MapperRegistry mapperRegistry, NamedWriteableRegistry namedWriteableRegistry, ThreadPool threadPool,
185193
IndexScopedSettings indexScopedSettings, CircuitBreakerService circuitBreakerService, BigArrays bigArrays,
186-
ScriptService scriptService, Client client, MetaStateService metaStateService) {
194+
ScriptService scriptService, Client client, MetaStateService metaStateService,
195+
Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders) {
187196
super(settings);
188197
this.threadPool = threadPool;
189198
this.pluginsService = pluginsService;
@@ -214,6 +223,7 @@ public void onRemoval(ShardId shardId, String fieldName, boolean wasEvicted, lon
214223
this.cleanInterval = INDICES_CACHE_CLEAN_INTERVAL_SETTING.get(settings);
215224
this.cacheCleaner = new CacheCleaner(indicesFieldDataCache, indicesRequestCache, logger, threadPool, this.cleanInterval);
216225
this.metaStateService = metaStateService;
226+
this.engineFactoryProviders = engineFactoryProviders;
217227
}
218228

219229
@Override
@@ -442,7 +452,7 @@ private synchronized IndexService createIndexService(final String reason,
442452
idxSettings.getNumberOfReplicas(),
443453
reason);
444454

445-
final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry);
455+
final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry, getEngineFactory(idxSettings));
446456
for (IndexingOperationListener operationListener : indexingOperationListeners) {
447457
indexModule.addIndexOperationListener(operationListener);
448458
}
@@ -466,6 +476,34 @@ private synchronized IndexService createIndexService(final String reason,
466476
);
467477
}
468478

479+
private EngineFactory getEngineFactory(final IndexSettings idxSettings) {
480+
final List<Optional<EngineFactory>> engineFactories =
481+
engineFactoryProviders
482+
.stream()
483+
.map(engineFactoryProvider -> engineFactoryProvider.apply(idxSettings))
484+
.filter(maybe -> Objects.requireNonNull(maybe).isPresent())
485+
.collect(Collectors.toList());
486+
if (engineFactories.isEmpty()) {
487+
return new InternalEngineFactory();
488+
} else if (engineFactories.size() == 1) {
489+
assert engineFactories.get(0).isPresent();
490+
return engineFactories.get(0).get();
491+
} else {
492+
final String message = String.format(
493+
Locale.ROOT,
494+
"multiple engine factories provided for %s: %s",
495+
idxSettings.getIndex(),
496+
engineFactories
497+
.stream()
498+
.map(t -> {
499+
assert t.isPresent();
500+
return "[" + t.get().getClass().getName() + "]";
501+
})
502+
.collect(Collectors.joining(",")));
503+
throw new IllegalStateException(message);
504+
}
505+
}
506+
469507
/**
470508
* creates a new mapper service for the given index, in order to do administrative work like mapping updates.
471509
* This *should not* be used for document parsing. Doing so will result in an exception.
@@ -474,7 +512,7 @@ private synchronized IndexService createIndexService(final String reason,
474512
*/
475513
public synchronized MapperService createIndexMapperService(IndexMetaData indexMetaData) throws IOException {
476514
final IndexSettings idxSettings = new IndexSettings(indexMetaData, this.settings, indexScopedSettings);
477-
final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry);
515+
final IndexModule indexModule = new IndexModule(idxSettings, analysisRegistry, getEngineFactory(idxSettings));
478516
pluginsService.onIndexModule(indexModule);
479517
return indexModule.newIndexMapperService(xContentRegistry, mapperRegistry, scriptService);
480518
}

‎server/src/main/java/org/elasticsearch/node/Node.java‎

Lines changed: 24 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121

2222
import org.apache.logging.log4j.Logger;
2323
import org.apache.lucene.util.Constants;
24-
import org.elasticsearch.core.internal.io.IOUtils;
2524
import org.apache.lucene.util.SetOnce;
2625
import org.elasticsearch.Build;
2726
import org.elasticsearch.ElasticsearchException;
@@ -82,6 +81,7 @@
8281
import org.elasticsearch.common.util.BigArrays;
8382
import org.elasticsearch.common.util.PageCacheRecycler;
8483
import org.elasticsearch.common.xcontent.NamedXContentRegistry;
84+
import org.elasticsearch.core.internal.io.IOUtils;
8585
import org.elasticsearch.discovery.Discovery;
8686
import org.elasticsearch.discovery.DiscoveryModule;
8787
import org.elasticsearch.discovery.DiscoverySettings;
@@ -93,8 +93,9 @@
9393
import org.elasticsearch.gateway.GatewayService;
9494
import org.elasticsearch.gateway.MetaStateService;
9595
import org.elasticsearch.http.HttpServerTransport;
96-
import org.elasticsearch.http.HttpTransportSettings;
96+
import org.elasticsearch.index.IndexSettings;
9797
import org.elasticsearch.index.analysis.AnalysisRegistry;
98+
import org.elasticsearch.index.engine.EngineFactory;
9899
import org.elasticsearch.indices.IndicesModule;
99100
import org.elasticsearch.indices.IndicesService;
100101
import org.elasticsearch.indices.analysis.AnalysisModule;
@@ -109,10 +110,15 @@
109110
import org.elasticsearch.ingest.IngestService;
110111
import org.elasticsearch.monitor.MonitorService;
111112
import org.elasticsearch.monitor.jvm.JvmInfo;
113+
import org.elasticsearch.persistent.PersistentTasksClusterService;
114+
import org.elasticsearch.persistent.PersistentTasksExecutor;
115+
import org.elasticsearch.persistent.PersistentTasksExecutorRegistry;
116+
import org.elasticsearch.persistent.PersistentTasksService;
112117
import org.elasticsearch.plugins.ActionPlugin;
113118
import org.elasticsearch.plugins.AnalysisPlugin;
114119
import org.elasticsearch.plugins.ClusterPlugin;
115120
import org.elasticsearch.plugins.DiscoveryPlugin;
121+
import org.elasticsearch.plugins.EnginePlugin;
116122
import org.elasticsearch.plugins.IngestPlugin;
117123
import org.elasticsearch.plugins.MapperPlugin;
118124
import org.elasticsearch.plugins.MetaDataUpgrader;
@@ -140,10 +146,6 @@
140146
import org.elasticsearch.transport.TransportService;
141147
import org.elasticsearch.usage.UsageService;
142148
import org.elasticsearch.watcher.ResourceWatcherService;
143-
import org.elasticsearch.persistent.PersistentTasksClusterService;
144-
import org.elasticsearch.persistent.PersistentTasksExecutor;
145-
import org.elasticsearch.persistent.PersistentTasksExecutorRegistry;
146-
import org.elasticsearch.persistent.PersistentTasksService;
147149

148150
import java.io.BufferedWriter;
149151
import java.io.Closeable;
@@ -161,6 +163,7 @@
161163
import java.util.Collections;
162164
import java.util.List;
163165
import java.util.Map;
166+
import java.util.Optional;
164167
import java.util.Set;
165168
import java.util.concurrent.CountDownLatch;
166169
import java.util.concurrent.TimeUnit;
@@ -395,11 +398,21 @@ protected Node(final Environment environment, Collection<Class<? extends Plugin>
395398
.flatMap(Function.identity()).collect(toList()));
396399
modules.add(new RepositoriesModule(this.environment, pluginsService.filterPlugins(RepositoryPlugin.class), xContentRegistry));
397400
final MetaStateService metaStateService = new MetaStateService(settings, nodeEnvironment, xContentRegistry);
398-
final IndicesService indicesService = new IndicesService(settings, pluginsService, nodeEnvironment, xContentRegistry,
399-
analysisModule.getAnalysisRegistry(),
400-
clusterModule.getIndexNameExpressionResolver(), indicesModule.getMapperRegistry(), namedWriteableRegistry,
401-
threadPool, settingsModule.getIndexScopedSettings(), circuitBreakerService, bigArrays, scriptModule.getScriptService(),
402-
client, metaStateService);
401+
402+
// collect engine factory providers from server and from plugins
403+
final Collection<EnginePlugin> enginePlugins = pluginsService.filterPlugins(EnginePlugin.class);
404+
final Collection<Function<IndexSettings, Optional<EngineFactory>>> engineFactoryProviders =
405+
Stream.concat(
406+
indicesModule.getEngineFactories().stream(),
407+
enginePlugins.stream().map(plugin -> plugin::getEngineFactory))
408+
.collect(Collectors.toList());
409+
410+
final IndicesService indicesService =
411+
new IndicesService(settings, pluginsService, nodeEnvironment, xContentRegistry, analysisModule.getAnalysisRegistry(),
412+
clusterModule.getIndexNameExpressionResolver(), indicesModule.getMapperRegistry(), namedWriteableRegistry,
413+
threadPool, settingsModule.getIndexScopedSettings(), circuitBreakerService, bigArrays,
414+
scriptModule.getScriptService(), client, metaStateService, engineFactoryProviders);
415+
403416

404417
Collection<Object> pluginComponents = pluginsService.filterPlugins(Plugin.class).stream()
405418
.flatMap(p -> p.createComponents(client, clusterService, threadPool, resourceWatcherService,

0 commit comments

Comments
 (0)