fabriziofortino commented on a change in pull request #263:
URL: https://github.com/apache/jackrabbit-oak/pull/263#discussion_r560303698
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/ElasticOutOfBandIndexer.java
##########
@@ -0,0 +1,65 @@
+package org.apache.jackrabbit.oak.index;
+
+
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.oak.plugins.index.CompositeIndexEditorProvider;
+import org.apache.jackrabbit.oak.plugins.index.IndexEditorProvider;
+import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticConnection;
+import
org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticIndexEditorProvider;
+import org.apache.jackrabbit.oak.plugins.index.search.ExtractedTextCache;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+import static java.util.Arrays.asList;
+
+public class ElasticOutOfBandIndexer extends OutOfBandIndexer {
+ private final Logger log = LoggerFactory.getLogger(this.getClass());
Review comment:
`log` is never used
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexerProvider.java
##########
@@ -0,0 +1,54 @@
+package org.apache.jackrabbit.oak.index.indexer.document;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.jackrabbit.oak.index.IndexHelper;
+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.index.ElasticIndexWriterFactory;
+import
org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter;
+import org.apache.jackrabbit.oak.plugins.index.search.ExtractedTextCache;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriter;
+import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+import org.apache.jackrabbit.oak.spi.state.NodeState;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME;
+
+public class ElasticIndexerProvider implements NodeStateIndexerProvider {
+ private final ExtractedTextCache textCache =
+ new ExtractedTextCache(FileUtils.ONE_MB * 5,
TimeUnit.HOURS.toSeconds(5));
+ private final IndexHelper indexHelper;
+ private final ElasticIndexWriterFactory indexWriterFactory;
+ private final ElasticConnection coordinate;
+
+ public ElasticIndexerProvider(IndexHelper indexHelper, ElasticConnection
coordinate) {
+ this.indexHelper = indexHelper;
+ this.indexWriterFactory = new ElasticIndexWriterFactory(coordinate);
+ this.coordinate = coordinate;
+ }
+
+
+ @Override
+ public @Nullable NodeStateIndexer getIndexer(@NotNull String type,
@NotNull String indexPath, @NotNull NodeBuilder definition, @NotNull NodeState
root, IndexingProgressReporter progressReporter) {
+ if
(!ElasticIndexDefinition.TYPE_ELASTICSEARCH.equals(definition.getString(TYPE_PROPERTY_NAME)))
{
+ return null;
+ }
+ ElasticIndexDefinition idxDefinition = (ElasticIndexDefinition) new
ElasticIndexDefinition.Builder(coordinate.getIndexPrefix()).
+
root(root).indexPath(indexPath).defn(definition.getNodeState()).reindex().build();
+
+ FulltextIndexWriter indexWriter =
indexWriterFactory.newInstance(idxDefinition, definition, CommitInfo.EMPTY,
true);
Review comment:
```suggestion
FulltextIndexWriter<ElasticDocument> indexWriter =
indexWriterFactory.newInstance(idxDefinition, definition, CommitInfo.EMPTY,
true);
```
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/ElasticIndexCommand.java
##########
@@ -0,0 +1,201 @@
+package org.apache.jackrabbit.oak.index;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Sets;
+import com.google.common.io.Closer;
+import joptsimple.OptionParser;
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
+import org.apache.jackrabbit.oak.plugins.index.importer.IndexDefinitionUpdater;
+import org.apache.jackrabbit.oak.run.cli.CommonOptions;
+import org.apache.jackrabbit.oak.run.cli.NodeStoreFixture;
+import org.apache.jackrabbit.oak.run.cli.NodeStoreFixtureProvider;
+import org.apache.jackrabbit.oak.run.cli.Options;
+import org.apache.jackrabbit.oak.run.commons.Command;
+import org.apache.jackrabbit.oak.run.commons.LoggingInitializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+public class ElasticIndexCommand implements Command {
+ private static final Logger log =
LoggerFactory.getLogger(ElasticIndexCommand.class);
+ private Options opts;
+ private ElasticIndexOptions indexOpts;
+ public static final String NAME = "index";
+
+ private final String summary = "Provides elastic index management related
operations";
+ private static boolean disableExitOnError;
+
+
+ @Override
+ public void execute(String... args) throws Exception {
+ OptionParser parser = new OptionParser();
+
+ opts = new Options();
+ opts.setCommandName(NAME);
+ opts.setSummary(summary);
+ opts.setConnectionString(CommonOptions.DEFAULT_CONNECTION_STRING);
+ opts.registerOptionsFactory(ElasticIndexOptions.FACTORY);
+ opts.parseAndConfigure(parser, args);
+
+ indexOpts = opts.getOptionBean(ElasticIndexOptions.class);
+
+ //Clean up before setting up NodeStore as the temp
+ //directory might be used by NodeStore for cache stuff like
persistentCache
+ //setupDirectories(indexOpts);
+ //setupLogging(indexOpts);
+
+ logCliArgs(args);
+
+ boolean success = false;
+ try {
+ try (Closer closer = Closer.create()) {
+ //configureCustomizer(opts, closer, true);
+ NodeStoreFixture fixture =
NodeStoreFixtureProvider.create(opts);
+ closer.register(fixture);
+ execute(fixture, indexOpts, closer);
+ //tellReportPaths();
+ }
+
+ success = true;
+ } catch (Throwable e) {
+ log.error("Error occurred while performing index tasks", e);
+ e.printStackTrace(System.err);
+ if (disableExitOnError) {
+ throw e;
+ }
+ } finally {
+ shutdownLogging();
+ }
+
+ if (!success) {
+ System.exit(1);
+ }
+ }
+
+ private void execute(NodeStoreFixture fixture, IndexOptions indexOpts,
Closer closer)
+ throws IOException, CommitFailedException {
+ IndexHelper indexHelper = createIndexHelper(fixture, indexOpts,
closer);
+
+ // TODO : See if we need to support dumpIndexStats and index defs for
elastic - not needed for now
+ //dumpIndexStats(indexOpts, indexHelper);
+ //dumpIndexDefinitions(indexOpts, indexHelper);
+ reindexOperation(indexOpts, indexHelper);
+ }
+
+ private IndexHelper createIndexHelper(NodeStoreFixture fixture,
+ IndexOptions indexOpts, Closer
closer) throws IOException {
+ IndexHelper indexHelper = new IndexHelper(fixture.getStore(),
fixture.getBlobStore(), fixture.getWhiteboard(),
+ indexOpts.getOutDir(), indexOpts.getWorkDir(),
computeIndexPaths(indexOpts));
+
+ // TODO : See if pre text extraction is needed for elastic
+ //configurePreExtractionSupport(indexOpts, indexHelper);
+
+ closer.register(indexHelper);
+ return indexHelper;
+ }
+
+ private List<String> computeIndexPaths(IndexOptions indexOpts) throws
IOException {
+ //Combine the indexPaths from json and cli args
+ Set<String> indexPaths = new
LinkedHashSet<>(indexOpts.getIndexPaths());
+ File definitions = indexOpts.getIndexDefinitionsFile();
+ if (definitions != null) {
+ IndexDefinitionUpdater updater = new
IndexDefinitionUpdater(definitions);
+ Set<String> indexPathsFromJson = updater.getIndexPaths();
+ Set<String> diff = Sets.difference(indexPathsFromJson, indexPaths);
+ if (!diff.isEmpty()) {
+ log.info("Augmenting the indexPaths with {} which are present
in {}", diff, definitions);
+ }
+ indexPaths.addAll(indexPathsFromJson);
+ }
+ return new ArrayList<>(indexPaths);
+ }
+
+ private void reindexOperation(IndexOptions indexOpts, IndexHelper
indexHelper) throws IOException, CommitFailedException {
+ if (!indexOpts.isReindex()) {
+ return;
+ }
+
+ String checkpoint = indexOpts.getCheckpoint();
+ reindex(indexOpts, indexHelper, checkpoint);
+ }
+
+ private void reindex(IndexOptions idxOpts, IndexHelper indexHelper, String
checkpoint) throws IOException, CommitFailedException {
+ Preconditions.checkNotNull(checkpoint, "Checkpoint value is required
for reindexing done in read only mode");
+
+ Stopwatch w = Stopwatch.createStarted();
Review comment:
this gets created but we never use it. Should we print the time the
reindex operation takes?
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/ElasticIndexCommand.java
##########
@@ -0,0 +1,201 @@
+package org.apache.jackrabbit.oak.index;
+
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Sets;
+import com.google.common.io.Closer;
+import joptsimple.OptionParser;
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.plugins.index.elastic.ElasticIndexDefinition;
+import org.apache.jackrabbit.oak.plugins.index.importer.IndexDefinitionUpdater;
+import org.apache.jackrabbit.oak.run.cli.CommonOptions;
+import org.apache.jackrabbit.oak.run.cli.NodeStoreFixture;
+import org.apache.jackrabbit.oak.run.cli.NodeStoreFixtureProvider;
+import org.apache.jackrabbit.oak.run.cli.Options;
+import org.apache.jackrabbit.oak.run.commons.Command;
+import org.apache.jackrabbit.oak.run.commons.LoggingInitializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+
+public class ElasticIndexCommand implements Command {
+ private static final Logger log =
LoggerFactory.getLogger(ElasticIndexCommand.class);
+ private Options opts;
+ private ElasticIndexOptions indexOpts;
+ public static final String NAME = "index";
+
+ private final String summary = "Provides elastic index management related
operations";
+ private static boolean disableExitOnError;
+
+
+ @Override
+ public void execute(String... args) throws Exception {
+ OptionParser parser = new OptionParser();
+
+ opts = new Options();
+ opts.setCommandName(NAME);
+ opts.setSummary(summary);
+ opts.setConnectionString(CommonOptions.DEFAULT_CONNECTION_STRING);
+ opts.registerOptionsFactory(ElasticIndexOptions.FACTORY);
+ opts.parseAndConfigure(parser, args);
+
+ indexOpts = opts.getOptionBean(ElasticIndexOptions.class);
+
+ //Clean up before setting up NodeStore as the temp
+ //directory might be used by NodeStore for cache stuff like
persistentCache
+ //setupDirectories(indexOpts);
+ //setupLogging(indexOpts);
+
+ logCliArgs(args);
+
+ boolean success = false;
+ try {
+ try (Closer closer = Closer.create()) {
+ //configureCustomizer(opts, closer, true);
+ NodeStoreFixture fixture =
NodeStoreFixtureProvider.create(opts);
+ closer.register(fixture);
+ execute(fixture, indexOpts, closer);
+ //tellReportPaths();
+ }
+
+ success = true;
+ } catch (Throwable e) {
+ log.error("Error occurred while performing index tasks", e);
+ e.printStackTrace(System.err);
+ if (disableExitOnError) {
+ throw e;
+ }
+ } finally {
+ shutdownLogging();
+ }
+
+ if (!success) {
+ System.exit(1);
+ }
+ }
+
+ private void execute(NodeStoreFixture fixture, IndexOptions indexOpts,
Closer closer)
+ throws IOException, CommitFailedException {
+ IndexHelper indexHelper = createIndexHelper(fixture, indexOpts,
closer);
+
+ // TODO : See if we need to support dumpIndexStats and index defs for
elastic - not needed for now
+ //dumpIndexStats(indexOpts, indexHelper);
+ //dumpIndexDefinitions(indexOpts, indexHelper);
+ reindexOperation(indexOpts, indexHelper);
+ }
+
+ private IndexHelper createIndexHelper(NodeStoreFixture fixture,
+ IndexOptions indexOpts, Closer
closer) throws IOException {
+ IndexHelper indexHelper = new IndexHelper(fixture.getStore(),
fixture.getBlobStore(), fixture.getWhiteboard(),
+ indexOpts.getOutDir(), indexOpts.getWorkDir(),
computeIndexPaths(indexOpts));
+
+ // TODO : See if pre text extraction is needed for elastic
+ //configurePreExtractionSupport(indexOpts, indexHelper);
+
+ closer.register(indexHelper);
+ return indexHelper;
+ }
+
+ private List<String> computeIndexPaths(IndexOptions indexOpts) throws
IOException {
+ //Combine the indexPaths from json and cli args
+ Set<String> indexPaths = new
LinkedHashSet<>(indexOpts.getIndexPaths());
+ File definitions = indexOpts.getIndexDefinitionsFile();
+ if (definitions != null) {
+ IndexDefinitionUpdater updater = new
IndexDefinitionUpdater(definitions);
+ Set<String> indexPathsFromJson = updater.getIndexPaths();
+ Set<String> diff = Sets.difference(indexPathsFromJson, indexPaths);
+ if (!diff.isEmpty()) {
+ log.info("Augmenting the indexPaths with {} which are present
in {}", diff, definitions);
+ }
+ indexPaths.addAll(indexPathsFromJson);
+ }
+ return new ArrayList<>(indexPaths);
+ }
+
+ private void reindexOperation(IndexOptions indexOpts, IndexHelper
indexHelper) throws IOException, CommitFailedException {
+ if (!indexOpts.isReindex()) {
+ return;
+ }
+
+ String checkpoint = indexOpts.getCheckpoint();
+ reindex(indexOpts, indexHelper, checkpoint);
+ }
+
+ private void reindex(IndexOptions idxOpts, IndexHelper indexHelper, String
checkpoint) throws IOException, CommitFailedException {
+ Preconditions.checkNotNull(checkpoint, "Checkpoint value is required
for reindexing done in read only mode");
+
+ Stopwatch w = Stopwatch.createStarted();
+ IndexerSupport indexerSupport = createIndexerSupport(indexHelper,
checkpoint);
+ log.info("Proceeding to index {} upto checkpoint {} {}",
indexHelper.getIndexPaths(), checkpoint,
+ indexerSupport.getCheckpointInfo());
+
+ if (opts.getCommonOpts().isMongo() && idxOpts.isDocTraversalMode()) {
+ log.info("Using Document order traversal to perform reindexing");
+ try (ElasticDocumentStoreIndexer indexer = new
ElasticDocumentStoreIndexer(indexHelper, indexerSupport,
indexOpts.getIndexPrefix(),
+ indexOpts.getElasticScheme(), indexOpts.getElasticHost(),
+ indexOpts.getElasticPort(), indexOpts.getApiKeyId(),
indexOpts.getApiKeySecret())) {
+ indexer.reindex();
+ // Wait for default flush interval before exiting the try block
+ // to make sure the client is not closed before the last flush
+ // TODO : See if this can be handled in a better manner
+
Thread.sleep(ElasticIndexDefinition.BULK_FLUSH_INTERVAL_MS_DEFAULT*2);
+ } catch (InterruptedException e) {
+ //
+ }
+ } else {
+ try(ElasticOutOfBandIndexer indexer = new
ElasticOutOfBandIndexer(indexHelper, indexerSupport, indexOpts.getIndexPrefix(),
+ indexOpts.getElasticScheme(), indexOpts.getElasticHost(),
+ indexOpts.getElasticPort(), indexOpts.getApiKeyId(),
indexOpts.getApiKeySecret())) {
+
+ indexer.reindex();
+ // Wait for default flush interval before exiting the try block
+ // to make sure the client is not closed before the last flush
+
Thread.sleep(ElasticIndexDefinition.BULK_FLUSH_INTERVAL_MS_DEFAULT*2);
+ } catch (InterruptedException e) {
+ //
+ }
+ }
+ indexerSupport.writeMetaInfo(checkpoint);
+ }
+
+ private IndexerSupport createIndexerSupport(IndexHelper indexHelper,
String checkpoint) {
+ IndexerSupport indexerSupport = new IndexerSupport(indexHelper,
checkpoint);
+
+ File definitions = indexOpts.getIndexDefinitionsFile();
+ if (definitions != null) {
+ Preconditions.checkArgument(definitions.exists(), "Index
definitions file [%s] not found", getPath(definitions));
+ indexerSupport.setIndexDefinitions(definitions);
+ }
+ return indexerSupport;
+ }
+
+ static Path getPath(File file) {
+ return file.toPath().normalize().toAbsolutePath();
+ }
+
+ private void shutdownLogging() {
+ LoggingInitializer.shutdownLogging();
+ }
+
+ private static void logCliArgs(String[] args) {
+ log.info("Command line arguments used for indexing [{}]", Joiner.on('
').join(args));
+ List<String> inputArgs =
ManagementFactory.getRuntimeMXBean().getInputArguments();
+ if (!inputArgs.isEmpty()) {
+ log.info("System properties and vm options passed {}", inputArgs);
+ }
+ }
+
+ public static void setDisableExitOnError(boolean disableExitOnError) {
Review comment:
this is never called
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java
##########
@@ -0,0 +1,99 @@
+package org.apache.jackrabbit.oak.index.indexer.document;
+
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.plugins.document.NodeDocument;
+import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocument;
+import
org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocumentMaker;
+import
org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter;
+import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriter;
+import org.apache.jackrabbit.oak.spi.filter.PathFilter;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+
+import java.io.IOException;
+import java.util.Set;
+
+public class ElasticIndexer implements NodeStateIndexer {
+
+ private final IndexDefinition definition;
+ private final FulltextBinaryTextExtractor binaryTextExtractor;
+ private final NodeBuilder definitionBuilder;
+ private final IndexingProgressReporter progressReporter;
+ //private final ElasticIndexEditorProvider elasticIndexEditorProvider;
+ private final FulltextIndexWriter indexWriter;
+
+ public ElasticIndexer(IndexDefinition definition,
FulltextBinaryTextExtractor binaryTextExtractor,
+ NodeBuilder definitionBuilder,
IndexingProgressReporter progressReporter,
+ FulltextIndexWriter indexWriter) {
Review comment:
```suggestion
FulltextIndexWriter<ElasticDocument> indexWriter) {
```
##########
File path:
oak-run-elastic/src/main/java/org/apache/jackrabbit/oak/index/indexer/document/ElasticIndexer.java
##########
@@ -0,0 +1,99 @@
+package org.apache.jackrabbit.oak.index.indexer.document;
+
+import org.apache.jackrabbit.oak.api.CommitFailedException;
+import org.apache.jackrabbit.oak.plugins.document.NodeDocument;
+import org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocument;
+import
org.apache.jackrabbit.oak.plugins.index.elastic.index.ElasticDocumentMaker;
+import
org.apache.jackrabbit.oak.plugins.index.progress.IndexingProgressReporter;
+import org.apache.jackrabbit.oak.plugins.index.search.IndexDefinition;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.binary.FulltextBinaryTextExtractor;
+import
org.apache.jackrabbit.oak.plugins.index.search.spi.editor.FulltextIndexWriter;
+import org.apache.jackrabbit.oak.spi.filter.PathFilter;
+import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
+
+import java.io.IOException;
+import java.util.Set;
+
+public class ElasticIndexer implements NodeStateIndexer {
+
+ private final IndexDefinition definition;
+ private final FulltextBinaryTextExtractor binaryTextExtractor;
+ private final NodeBuilder definitionBuilder;
+ private final IndexingProgressReporter progressReporter;
+ //private final ElasticIndexEditorProvider elasticIndexEditorProvider;
+ private final FulltextIndexWriter indexWriter;
Review comment:
```suggestion
private final FulltextIndexWriter<ElasticDocument> indexWriter;
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
