Skip to content

OAK-12249: lazy ES index provisioning — skip creation for empty reindex #2955

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bhabegger wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

OAK-12249: lazy ES index provisioning — skip creation for empty reindex #2955

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions ...src/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexDefinition.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,13 @@ public class ElasticIndexDefinition extends IndexDefinition {
*/
public static final String PROP_INDEX_NAME_SEED = ":nameSeed";

/**
* Hidden property written when a lazy reindex (OAK-12249) produced zero documents and left no
* ES index or alias. Signals the next incremental-write cycle to provision the index on demand.
* Cleared once provisioning completes.
*/
public static final String PROP_REQUIRES_PROVISIONING = ":requiresProvisioning";

/**
* Hidden property to store similarity tags
*/
Expand Down Expand Up @@ -268,6 +275,10 @@ public String getIndexAlias() {
return indexAlias;
}

public boolean requiresProvisioning() {
return getOptionalValue(getDefinitionNodeState(), PROP_REQUIRES_PROVISIONING, false);
}

public Map<String, List<PropertyDefinition>> getPropertiesByName() {
return propertiesByName;
}
Expand Down
3 changes: 3 additions & 0 deletions ...ain/java/org/apache/jackrabbit/oak/plugins/index/elastic/ElasticIndexProviderService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,9 @@ private void activate(BundleContext bundleContext, Config config) {
oakRegs.add(whiteboard.register(FeatureToggle.class,
new FeatureToggle(ElasticIndexStatistics.FT_OAK_12248, ElasticIndexStatistics.FT_OAK_12248_ENABLE),
emptyMap()));
oakRegs.add(whiteboard.register(FeatureToggle.class,
new FeatureToggle(ElasticIndexEditorProvider.FT_OAK_12249, ElasticIndexEditorProvider.FT_OAK_12249_ENABLE),
emptyMap()));
if (System.getProperty(QueryEngineSettings.OAK_INFERENCE_ENABLED) != null) {
this.isInferenceEnabled = Boolean.parseBoolean(System.getProperty(QueryEngineSettings.OAK_INFERENCE_ENABLED));
} else {
Expand Down
5 changes: 3 additions & 2 deletions ...java/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticIndexEditorContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.DocumentMaker;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexEditorContext;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriter;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.jetbrains.annotations.Nullable;
Expand Down Expand Up @@ -50,8 +51,8 @@ public DocumentMaker<ElasticDocument> newDocumentMaker(IndexDefinition.IndexingR
}

@Override
public ElasticIndexWriter getWriter() {
return (ElasticIndexWriter) super.getWriter();
public FulltextIndexWriter<ElasticDocument> getWriter() {
return super.getWriter();
}

@Override
Expand Down
31 changes: 31 additions & 0 deletions ...ava/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticIndexEditorProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.jackrabbit.oak.plugins.index.IndexingContext;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticConnection;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexStatistics;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexTracker;
import org.apache.jackrabbit.oak.plugins.index.search.ExtractedTextCache;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexEditor;
Expand All @@ -32,6 +33,8 @@
import org.apache.jackrabbit.oak.spi.state.NodeState;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;
Expand All @@ -41,6 +44,8 @@

public class ElasticIndexEditorProvider implements IndexEditorProvider {

private static final Logger LOG = LoggerFactory.getLogger(ElasticIndexEditorProvider.class);

private final ElasticIndexTracker indexTracker;
private final ElasticConnection elasticConnection;
private final ExtractedTextCache extractedTextCache;
Expand All @@ -57,6 +62,32 @@ public class ElasticIndexEditorProvider implements IndexEditorProvider {
*/
public static final AtomicBoolean FT_OAK_12206_DISABLE = new AtomicBoolean(false);

public static final String FT_OAK_12249 = "FT_OAK-12249";
/**
* When {@code true} AND {@link ElasticIndexStatistics#FT_OAK_12248_ENABLE} is also {@code true},
* Elasticsearch index provisioning is deferred to the first {@code updateDocument()} or
* {@code deleteDocuments()} call. A reindex that produces no documents never creates an ES
* index or alias.
*
* <p>Requires {@code FT_OAK-12248} (graceful 404 handling) to be enabled first. Enabling
* lazy provisioning without graceful 404 handling would cause unhandled ES 404 errors on every
* query against an empty-reindexed index. {@link #isLazyProvisioningActive()} enforces this
* dependency at runtime.
*
* <p>Disabled by default.
*/
public static final AtomicBoolean FT_OAK_12249_ENABLE = new AtomicBoolean(false);

/**
* Returns {@code true} when lazy provisioning is active. Requires both this toggle and
* {@link ElasticIndexStatistics#FT_OAK_12248_ENABLE} to be {@code true}. The combined check
* enforces the deployment order: graceful 404 handling (OAK-12248) must be on before lazy
* provisioning (OAK-12249) can take effect.
*/
public static boolean isLazyProvisioningActive() {
return FT_OAK_12249_ENABLE.get() && ElasticIndexStatistics.FT_OAK_12248_ENABLE.get();
}

private final boolean OAK_INDEX_ELASTIC_WRITER_DISABLE = Boolean.getBoolean(OAK_INDEX_ELASTIC_WRITER_DISABLE_KEY);

public ElasticIndexEditorProvider(@NotNull ElasticIndexTracker indexTracker,
Expand Down
64 changes: 34 additions & 30 deletions ...c/main/java/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticIndexWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,46 +62,46 @@ class ElasticIndexWriter implements FulltextIndexWriter<ElasticDocument> {
private final ElasticConnection elasticConnection;
private final ElasticIndexDefinition indexDefinition;
private final ElasticBulkProcessorHandler bulkProcessorHandler;
private final boolean reindex;
private final boolean requiresProvisioning;
private final String indexName;
private final ElasticRetryPolicy retryPolicy;
private final NodeBuilder definitionBuilder;

ElasticIndexWriter(@NotNull ElasticIndexTracker indexTracker,
@NotNull ElasticConnection elasticConnection,
@NotNull ElasticIndexDefinition indexDefinition,
@NotNull NodeBuilder definitionBuilder,
boolean reindex, CommitInfo commitInfo,
boolean requiresProvisioning, CommitInfo commitInfo,
ElasticBulkProcessorHandler bulkProcessorHandler,
ElasticRetryPolicy retryPolicy) {
this.indexTracker = indexTracker;
this.elasticConnection = elasticConnection;
this.indexDefinition = indexDefinition;
this.reindex = reindex;
this.requiresProvisioning = requiresProvisioning;
this.bulkProcessorHandler = bulkProcessorHandler;
this.retryPolicy = retryPolicy;
this.definitionBuilder = definitionBuilder;

// We don't use stored index definitions with elastic. Every time a new writer gets created we
// use the actual index name (based on the current seed) while reindexing, or the alias (pointing to the
// old index until the new one gets enabled) during incremental reindexing
if (this.reindex) {
if (requiresProvisioning) {
// Full provisioning: generate a seed-based backing index, create it in ES, and prepare
// for alias flip on close(). Applies to both a standard reindex and an incremental write
// arriving after a lazy reindex that produced zero documents (OAK-12249).
try {
//TODO we should observe changes under inference config path.
InferenceConfig.reInitialize();
// refresh inference config on any index reindex.
long seed = indexDefinition.indexNameSeed == 0L ? UUID.randomUUID().getMostSignificantBits() : indexDefinition.indexNameSeed;
// merge gets called on node store later in the indexing flow
definitionBuilder.setProperty(ElasticIndexDefinition.PROP_INDEX_NAME_SEED, seed);
// let's store the current mapping version in the index definition
definitionBuilder.setProperty(ElasticIndexDefinition.PROP_INDEX_MAPPING_VERSION, ElasticIndexDefinition.MAPPING_VERSION.toString());

definitionBuilder.removeProperty(ElasticIndexDefinition.PROP_REQUIRES_PROVISIONING);
indexName = ElasticIndexNameHelper.
getRemoteIndexName(elasticConnection.getIndexPrefix(), indexDefinition.getIndexPath(), seed);

provisionIndex();
} catch (IOException e) {
throw new IllegalStateException("Unable to provision index", e);
}
} else indexName = indexDefinition.getIndexAlias();
} else {
indexName = indexDefinition.getIndexAlias();
}
boolean waitForESAcknowledgement = true;
PropertyState async = indexDefinition.getDefinitionNodeState().getProperty("async");
if (async != null) {
Expand Down Expand Up @@ -132,14 +132,15 @@ class ElasticIndexWriter implements FulltextIndexWriter<ElasticDocument> {
@NotNull ElasticIndexDefinition indexDefinition,
@NotNull ElasticBulkProcessorHandler bulkProcessorHandler,
@NotNull ElasticRetryPolicy retryPolicy,
boolean reindex) {
boolean requiresProvisioning) {
this.indexTracker = indexTracker;
this.elasticConnection = elasticConnection;
this.indexDefinition = indexDefinition;
this.bulkProcessorHandler = bulkProcessorHandler;
this.indexName = indexDefinition.getIndexAlias();
this.retryPolicy = retryPolicy;
this.reindex = reindex;
this.requiresProvisioning = requiresProvisioning;
this.definitionBuilder = null;
}

@Override
Expand All @@ -155,7 +156,7 @@ public void updateDocument(String path, ElasticDocument doc) throws IOException
AND InferenceIndexConfig is NOOP
)
*/
if (reindex
if (requiresProvisioning
|| (!indexDefinition.isExternallyModifiable()
&& !InferenceConfig.getInstance().isInferenceEnabled()
&& (InferenceIndexConfig.NOOP.equals(InferenceConfig.getInstance().getInferenceIndexConfig(jcrIndexName))))) {
Expand Down Expand Up @@ -190,8 +191,7 @@ public void deleteDocuments(String path) throws IOException {
@Override
public boolean close(long timestamp) throws IOException {
boolean updateStatus = bulkProcessorHandler.flushIndex(indexName);
if (reindex) {
// if we are closing a writer in reindex mode, it means we need to open the new index for queries
if (requiresProvisioning) {
this.enableIndex();
}
if (updateStatus) {
Expand Down Expand Up @@ -220,43 +220,47 @@ private void saveMetrics() {

private void provisionIndex() throws IOException {
final ElasticsearchIndicesClient esClient = elasticConnection.getClient().indices();
// check if index already exists
if (esClient.exists(i -> i.index(indexName)).value()) {
LOG.info("Index {} already exists. Skip index provision", indexName);
return;
}
createIndex(indexName);
}

/**
* Builds a {@link CreateIndexRequest} for {@code backingIndexName} and submits it to
* Elasticsearch, with debug logging and idempotent handling of concurrent-creation races.
*/
private void createIndex(String backingIndexName) throws IOException {
final ElasticsearchIndicesClient esClient = elasticConnection.getClient().indices();
CreateIndexRequest request;
try {
request = ElasticIndexHelper.createIndexRequest(indexName, indexDefinition);
request = ElasticIndexHelper.createIndexRequest(backingIndexName, indexDefinition);
} catch (Exception e) {
LOG.error("Failed to create index {}: {}", indexName, e.toString());
LOG.error("Failed to create index {}: {}", backingIndexName, e.toString());
throw e;
}
if (LOG.isDebugEnabled()) {
int old = JsonpUtils.maxToStringLength();
try {
// temporarily increase the length, to avoid truncation
JsonpUtils.maxToStringLength(1_000_000);
LOG.debug("Creating Index with request {}", request);
} finally {
JsonpUtils.maxToStringLength(old);
}
}
// create the new index
try {
final CreateIndexResponse response = esClient.create(request);
LOG.info("Created index {}. Response acknowledged: {}", indexName, response.acknowledged());
checkResponseAcknowledgement(response, "Create index call not acknowledged for index " + indexName);
LOG.info("Created index {}. Response acknowledged: {}", backingIndexName, response.acknowledged());
checkResponseAcknowledgement(response, "Create index call not acknowledged for index " + backingIndexName);
} catch (ElasticsearchException ese) {
// We already check index existence as first thing in this method, if we get here it means we have got into
// a conflict (eg: multiple cluster nodes provision concurrently).
// Elasticsearch does not have a CREATE IF NOT EXIST, need to inspect exception
// We already check index existence as first thing in provisionIndex(); if we get here it
// means a concurrent cluster node raced us. Elasticsearch has no CREATE IF NOT EXISTS:
// https://github.com/elastic/elasticsearch/issues/19862
if (ese.status() == 400 && ese.getMessage().contains("resource_already_exists_exception")) {
LOG.warn("Index {} already exists. Ignoring error", indexName);
LOG.warn("Index {} already exists. Ignoring error", backingIndexName);
} else {
LOG.warn("Failed to create index {}", indexName, ese);
LOG.warn("Failed to create index {}", backingIndexName, ese);
StringBuilder sb = new StringBuilder();
int old = JsonpUtils.maxToStringLength();
try {
Expand Down
22 changes: 19 additions & 3 deletions ...java/org/apache/jackrabbit/oak/plugins/index/elastic/index/ElasticIndexWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexTracker;
import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriter;
import org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriterFactory;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
Expand All @@ -43,15 +44,30 @@ public ElasticIndexWriterFactory(@NotNull ElasticConnection elasticConnection, @
}

@Override
public ElasticIndexWriter newInstance(IndexDefinition definition, NodeBuilder definitionBuilder,
CommitInfo commitInfo, boolean reindex) {
public FulltextIndexWriter<ElasticDocument> newInstance(IndexDefinition definition, NodeBuilder definitionBuilder,
CommitInfo commitInfo, boolean reindex) {
if (!(definition instanceof ElasticIndexDefinition)) {
throw new IllegalArgumentException("IndexDefinition must be of type ElasticsearchIndexDefinition " +
"instead of " + definition.getClass().getName());
}

ElasticIndexDefinition esDefinition = (ElasticIndexDefinition) definition;

return new ElasticIndexWriter(indexTracker, elasticConnection, esDefinition, definitionBuilder, reindex, commitInfo, bulkProcessorHandler, retryPolicy);
// requiresProvisioning=true for a standard reindex, or when a prior lazy reindex produced
// zero documents and set PROP_REQUIRES_PROVISIONING in the node store.
boolean requiresProvisioning = reindex || esDefinition.requiresProvisioning();

if (requiresProvisioning && ElasticIndexEditorProvider.isLazyProvisioningActive()) {
// OAK-12249: defer provisioning to the first write, whether this is a reindex or an
// incremental cycle after an empty lazy reindex. If no documents arrive the supplier is
// never called, PROP_REQUIRES_PROVISIONING is re-written, and the next cycle retries.
return new LazyElasticIndexWriter(
() -> new ElasticIndexWriter(indexTracker, elasticConnection, esDefinition,
definitionBuilder, true, commitInfo, bulkProcessorHandler, retryPolicy),
definitionBuilder);
}

return new ElasticIndexWriter(indexTracker, elasticConnection, esDefinition,
definitionBuilder, requiresProvisioning, commitInfo, bulkProcessorHandler, retryPolicy);
}
}
Loading