Package org.nuxeo.elasticsearch

Class ElasticSearchComponent

java.lang.Object

org.nuxeo.runtime.model.DefaultComponent

org.nuxeo.elasticsearch.ElasticSearchComponent

All Implemented Interfaces:

ElasticSearchAdmin, ElasticSearchIndexing, ElasticSearchService, Adaptable, Component, Extensible, TimestampedService

public class ElasticSearchComponent
extends DefaultComponent
implements ElasticSearchAdmin, ElasticSearchIndexing, ElasticSearchService

Component used to configure and manage ElasticSearch integration

Field Summary

Fields

Modifier and Type	Field	Description
protected ElasticSearchClientConfig	clientConfig
protected ElasticSearchEmbeddedServerConfig	embeddedServerConfig
protected static String	EP_CLIENT_INIT
protected static String	EP_DOC_WRITER
protected static String	EP_EMBEDDED_SERVER
protected static String	EP_HINTS
protected static String	EP_INDEX
protected ElasticSearchAdminImpl	esa
protected ElasticSearchIndexingImpl	esi
protected ElasticSearchServiceImpl	ess
protected Map<String,ElasticSearchIndexConfig>	indexConfig
protected JsonESDocumentWriter	jsonESDocumentWriter
protected static Log	log
protected static long	REINDEX_TIMEOUT
protected AtomicInteger	runIndexingWorkerCount
protected List<IndexingCommand>	stackedCommands
protected com.google.common.util.concurrent.ListeningExecutorService	waiterExecutorService

Fields inherited from class org.nuxeo.runtime.model.DefaultComponent

lastModified, name

Constructor Summary

Constructors

Constructor	Description
ElasticSearchComponent()

Method Summary

Modifier and Type	Method	Description
void	clearScroll(EsScrollResult scrollResult)	Clear scroll on ElasticSearch cluster for the given EsScrollResult.
protected void	dispatchWork(List<IndexingCommand> cmds)	Dispatch jobs between sync and async worker
void	dropAndInitIndex(String indexName)	Reinitialize an index.
void	dropAndInitRepositoryIndex(String repositoryName, boolean syncAlias)	Reinitialize the index of a repository.
void	flush()	Elasticsearch flush on all document indexes, triggers a lucene commit, empties the transaction log.
void	flushRepositoryIndex(String repositoryName)	Elasticsearch flush on document index for a specific repository, triggers a lucene commit, empties the transaction log.
int	getApplicationStartedOrder()	The component notification order for Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext).
ESClient	getClient()	Retrieves the Client that can be used to access Elasticsearch API
Optional<ESHintQueryBuilder>	getHintByOperator(String name)	Returns the hint by the Elasticsearch operator name.
String	getIndexNameForRepository(String repositoryName)	Get the search index name associated with the repository name.
String	getIndexNameForType(String type)	Get the first search index name with the given type.
List<String>	getIndexNamesForType(String type)	Get the index names with the given type.
long	getPendingWorkerCount()	Returns the number of indexing worker scheduled waiting to be executed.
String	getRepositoryForIndex(String indexName)	Gets the repository name associated with the index.
List<String>	getRepositoryNames()	List repository names that have Elasticsearch support.
long	getRunningWorkerCount()	Returns the number of indexing worker that are currently running.
String	getSecondaryWriteIndexName(String searchIndexName)	Returns the secondary write index used during re-indexing with aliases.
int	getTotalCommandProcessed()	Returns the total number of command processed by Elasticsearch for this Nuxeo instance.
String	getWriteIndexName(String searchIndexName)	Returns the index to use for any write operations.
void	indexNonRecursive(List<IndexingCommand> cmds)	Same as ElasticSearchIndexing.indexNonRecursive(org.nuxeo.elasticsearch.commands.IndexingCommand) but process the list command using a bulk request.
void	indexNonRecursive(IndexingCommand cmd)	Process the IndexingCommand.
void	initIndexes(boolean dropIfExists)	Initialize Elasticsearch indexes.
protected void	initListenerThreadPool()
void	initRepositoryIndexWithAliases(String repositoryName)	Creates a new index for the repository applying the mapping and settings.
protected boolean	isElasticsearchEnabled()
boolean	isEmbedded()	Returns true if the Elasticsearch is embedded with Nuxeo, sharing the same JVM.
boolean	isIndexingInProgress()	Returns true if there are indexing activities scheduled or running.
boolean	isReady()
void	optimize()	Elasticsearch run ElasticSearchAdmin.optimizeRepositoryIndex(java.lang.String) on all document indexes,
void	optimizeIndex(String indexName)	Elasticsearch optimize operation allows to reduce the number of segments to one, Note that this can potentially be a very heavy operation.
void	optimizeRepositoryIndex(String repositoryName)	Elasticsearch optimize operation allows to reduce the number of segments to one, Note that this can potentially be a very heavy operation.
com.google.common.util.concurrent.ListenableFuture<Boolean>	prepareWaitForIndexing()	A Future that accepts callback on completion when all the indexing worker are done.
DocumentModelList	query(CoreSession session, String nxql, int limit, int offset, SortInfo... sortInfos)	Deprecated.
DocumentModelList	query(CoreSession session, org.elasticsearch.index.query.QueryBuilder queryBuilder, int limit, int offset, SortInfo... sortInfos)	Deprecated.
DocumentModelList	query(NxQueryBuilder queryBuilder)	Returns a document list using an NxQueryBuilder.
EsResult	queryAndAggregate(NxQueryBuilder queryBuilder)	Returns documents and aggregates.
void	refresh()	Refresh all document indexes, immediately after the operation occurs, so that the updated document appears in search results immediately.
void	refreshRepositoryIndex(String repositoryName)	Refresh document index for the specific repository, immediately after the operation occurs, so that the updated document appears in search results immediately.
void	registerContribution(Object contribution, String extensionPoint, ComponentInstance contributor)
protected void	reindexOnStartup()
void	reindexRepository(String repositoryName)	Recreate an index and run an async reindexing worker.
protected void	runIndexingSyncWorker(Map<String,List<IndexingCommand>> syncCommands)
void	runIndexingWorker(List<IndexingCommand> cmds)	Run a worker to process the IndexingCommand.
void	runReindexingWorker(String repositoryName, String nxql, boolean syncAlias)	Reindex documents matching the NXQL query, This is done in an asynchronous job.
protected void	scheduleIndexingAsyncWorker(Map<String,List<IndexingCommand>> asyncCommands)
EsScrollResult	scroll(EsScrollResult scrollResult)	Retrieves the next batch of results of a scrollable search request for the given EsScrollResult.
EsScrollResult	scroll(NxQueryBuilder queryBuilder, long keepAlive)	Performs the initial search of a scrollable search request using an NxQueryBuilder.
protected void	shutdownListenerThreadPool()
org.elasticsearch.common.bytes.BytesReference	source(DocumentModel doc)	Returns the JSON Elasticsearch source representation of a document.
protected void	stackCommands(List<IndexingCommand> cmds)
void	start(ComponentContext context)	Start the component.
void	stop(ComponentContext context)	Stop the component.
void	syncSearchAndWriteAlias(String searchIndexName)	Make sure that the search alias point to the same index as the write alias.
boolean	useExternalVersion()	When true use an external version for Elasticsearch document, this enable an optimistic concurrency control ensuring that an older version of a document never overwrites a newer version.

Methods inherited from class org.nuxeo.runtime.model.DefaultComponent

activate, addRuntimeMessage, addRuntimeMessage, deactivate, getAdapter, getDescriptor, getDescriptors, getLastModified, getRegistry, register, registerExtension, setLastModified, setModifiedNow, setName, unregister, unregisterContribution, unregisterExtension

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.nuxeo.runtime.model.Component

applicationStarted

Methods inherited from interface org.nuxeo.elasticsearch.api.ElasticSearchAdmin

dropAndInitRepositoryIndex

Methods inherited from interface org.nuxeo.elasticsearch.api.ElasticSearchIndexing

runReindexingWorker

Field Detail

log

protected static final Log log

EP_EMBEDDED_SERVER

protected static final String EP_EMBEDDED_SERVER

See Also:

Constant Field Values

EP_CLIENT_INIT

protected static final String EP_CLIENT_INIT

See Also:

Constant Field Values

EP_INDEX

protected static final String EP_INDEX

See Also:

Constant Field Values

EP_DOC_WRITER

protected static final String EP_DOC_WRITER

See Also:

Constant Field Values

EP_HINTS

protected static final String EP_HINTS

Since:

11.1

See Also:

Constant Field Values

REINDEX_TIMEOUT

protected static final long REINDEX_TIMEOUT

See Also:

Constant Field Values

stackedCommands

protected final List<IndexingCommand> stackedCommands

indexConfig

protected final Map<String,ElasticSearchIndexConfig> indexConfig

runIndexingWorkerCount

protected final AtomicInteger runIndexingWorkerCount

embeddedServerConfig

protected ElasticSearchEmbeddedServerConfig embeddedServerConfig

clientConfig

protected ElasticSearchClientConfig clientConfig

esa

protected ElasticSearchAdminImpl esa

esi

protected ElasticSearchIndexingImpl esi

ess

protected ElasticSearchServiceImpl ess

jsonESDocumentWriter

protected JsonESDocumentWriter jsonESDocumentWriter

waiterExecutorService

protected com.google.common.util.concurrent.ListeningExecutorService waiterExecutorService

Constructor Detail

ElasticSearchComponent

public ElasticSearchComponent()

Method Detail

registerContribution

public void registerContribution(Object contribution,
                                 String extensionPoint,
                                 ComponentInstance contributor)

Overrides:

registerContribution in class DefaultComponent

start

public void start(ComponentContext context)

Description copied from interface: Component

Start the component. This method is called after all the components were resolved and activated

Specified by:

start in interface Component

Overrides:

start in class DefaultComponent

stop

public void stop(ComponentContext context)

Description copied from interface: Component

Stop the component.

Specified by:

stop in interface Component

Overrides:

stop in class DefaultComponent

reindexOnStartup

protected void reindexOnStartup()

isElasticsearchEnabled

protected boolean isElasticsearchEnabled()

getApplicationStartedOrder

public int getApplicationStartedOrder()

Description copied from interface: Component

The component notification order for Component.applicationStarted(org.nuxeo.runtime.model.ComponentContext).

Components are notified in increasing order. Order 1000 is the default order for components that don't care. Order 100 is the repository initialization.

Specified by:

getApplicationStartedOrder in interface Component

Returns:

the order, 1000 by default

getClient

public ESClient getClient()

Description copied from interface: ElasticSearchAdmin

Retrieves the Client that can be used to access Elasticsearch API

Specified by:

getClient in interface ElasticSearchAdmin

initIndexes

public void initIndexes(boolean dropIfExists)

Description copied from interface: ElasticSearchAdmin

Initialize Elasticsearch indexes. Setup the index settings and mapping for each index that has been registered.

Specified by:

initIndexes in interface ElasticSearchAdmin

Parameters:

dropIfExists - if {true} remove an existing index

dropAndInitIndex

public void dropAndInitIndex(String indexName)

Description copied from interface: ElasticSearchAdmin

Reinitialize an index. This will drop the existing index, recreate it with its settings and mapping, the index will be empty.

Specified by:

dropAndInitIndex in interface ElasticSearchAdmin

dropAndInitRepositoryIndex

public void dropAndInitRepositoryIndex(String repositoryName,
                                       boolean syncAlias)

Description copied from interface: ElasticSearchAdmin

Reinitialize the index of a repository. This will drop the existing index, recreate it with its settings and mapping, the index will be empty. When syncAlias is false then search alias is not updated with the new index, you need to explicitly call ElasticSearchAdmin.syncSearchAndWriteAlias(String)

Specified by:

dropAndInitRepositoryIndex in interface ElasticSearchAdmin

initRepositoryIndexWithAliases

public void initRepositoryIndexWithAliases(String repositoryName)

Description copied from interface: ElasticSearchAdmin

Creates a new index for the repository applying the mapping and settings. Update the write alias to point to this new index. The search alias is updated to the new index only for the initial creation. Otherwise, there are two write indexes until an explicit call to ElasticSearchAdmin.syncSearchAndWriteAlias(String).

Specified by:

initRepositoryIndexWithAliases in interface ElasticSearchAdmin

getRepositoryNames

public List<String> getRepositoryNames()

Description copied from interface: ElasticSearchAdmin

List repository names that have Elasticsearch support.

Specified by:

getRepositoryNames in interface ElasticSearchAdmin

getIndexNameForRepository

public String getIndexNameForRepository(String repositoryName)

Description copied from interface: ElasticSearchAdmin

Get the search index name associated with the repository name.

Specified by:

getIndexNameForRepository in interface ElasticSearchAdmin

getRepositoryForIndex

public String getRepositoryForIndex(String indexName)

Description copied from interface: ElasticSearchAdmin

Gets the repository name associated with the index.

Specified by:

getRepositoryForIndex in interface ElasticSearchAdmin

getIndexNamesForType

public List<String> getIndexNamesForType(String type)

Description copied from interface: ElasticSearchAdmin

Get the index names with the given type.

Specified by:

getIndexNamesForType in interface ElasticSearchAdmin

getIndexNameForType

public String getIndexNameForType(String type)

Description copied from interface: ElasticSearchAdmin

Get the first search index name with the given type.

Specified by:

getIndexNameForType in interface ElasticSearchAdmin

getWriteIndexName

public String getWriteIndexName(String searchIndexName)

Description copied from interface: ElasticSearchAdmin

Returns the index to use for any write operations.

Specified by:

getWriteIndexName in interface ElasticSearchAdmin

getSecondaryWriteIndexName

public String getSecondaryWriteIndexName(String searchIndexName)

Description copied from interface: ElasticSearchAdmin

Returns the secondary write index used during re-indexing with aliases.

Specified by:

getSecondaryWriteIndexName in interface ElasticSearchAdmin

Returns:

an index name or null if there is no re-indexing in progress.

syncSearchAndWriteAlias

public void syncSearchAndWriteAlias(String searchIndexName)

Description copied from interface: ElasticSearchAdmin

Make sure that the search alias point to the same index as the write alias.

Specified by:

syncSearchAndWriteAlias in interface ElasticSearchAdmin

getPendingWorkerCount

public long getPendingWorkerCount()

Description copied from interface: ElasticSearchAdmin

Returns the number of indexing worker scheduled waiting to be executed.

Specified by:

getPendingWorkerCount in interface ElasticSearchAdmin

getRunningWorkerCount

public long getRunningWorkerCount()

Description copied from interface: ElasticSearchAdmin

Returns the number of indexing worker that are currently running.

Specified by:

getRunningWorkerCount in interface ElasticSearchAdmin

getTotalCommandProcessed

public int getTotalCommandProcessed()

Description copied from interface: ElasticSearchAdmin

Returns the total number of command processed by Elasticsearch for this Nuxeo instance. Useful for test assertion.

Specified by:

getTotalCommandProcessed in interface ElasticSearchAdmin

isEmbedded

public boolean isEmbedded()

Description copied from interface: ElasticSearchAdmin

Returns true if the Elasticsearch is embedded with Nuxeo, sharing the same JVM.

Specified by:

isEmbedded in interface ElasticSearchAdmin

useExternalVersion

public boolean useExternalVersion()

Description copied from interface: ElasticSearchAdmin

When true use an external version for Elasticsearch document, this enable an optimistic concurrency control ensuring that an older version of a document never overwrites a newer version.

Specified by:

useExternalVersion in interface ElasticSearchAdmin

isIndexingInProgress

public boolean isIndexingInProgress()

Description copied from interface: ElasticSearchAdmin

Returns true if there are indexing activities scheduled or running.

Specified by:

isIndexingInProgress in interface ElasticSearchAdmin

prepareWaitForIndexing

public com.google.common.util.concurrent.ListenableFuture<Boolean> prepareWaitForIndexing()

Description copied from interface: ElasticSearchAdmin

A Future that accepts callback on completion when all the indexing worker are done.

Specified by:

prepareWaitForIndexing in interface ElasticSearchAdmin

initListenerThreadPool

protected void initListenerThreadPool()

shutdownListenerThreadPool

protected void shutdownListenerThreadPool()

refresh

public void refresh()

Description copied from interface: ElasticSearchAdmin

Refresh all document indexes, immediately after the operation occurs, so that the updated document appears in search results immediately. There is no fsync thus doesn't guarantee durability.

Specified by:

refresh in interface ElasticSearchAdmin

refreshRepositoryIndex

public void refreshRepositoryIndex(String repositoryName)

Description copied from interface: ElasticSearchAdmin

Refresh document index for the specific repository, immediately after the operation occurs, so that the updated document appears in search results immediately. There is no fsync thus doesn't guarantee durability.

Specified by:

refreshRepositoryIndex in interface ElasticSearchAdmin

flush

public void flush()

Description copied from interface: ElasticSearchAdmin

Elasticsearch flush on all document indexes, triggers a lucene commit, empties the transaction log. Data is flushed to disk.

Specified by:

flush in interface ElasticSearchAdmin

flushRepositoryIndex

public void flushRepositoryIndex(String repositoryName)

Description copied from interface: ElasticSearchAdmin

Elasticsearch flush on document index for a specific repository, triggers a lucene commit, empties the transaction log. Data is flushed to disk.

Specified by:

flushRepositoryIndex in interface ElasticSearchAdmin

optimize

public void optimize()

Description copied from interface: ElasticSearchAdmin

Elasticsearch run ElasticSearchAdmin.optimizeRepositoryIndex(java.lang.String) on all document indexes,

Specified by:

optimize in interface ElasticSearchAdmin

optimizeRepositoryIndex

public void optimizeRepositoryIndex(String repositoryName)

Description copied from interface: ElasticSearchAdmin

Elasticsearch optimize operation allows to reduce the number of segments to one, Note that this can potentially be a very heavy operation.

Specified by:

optimizeRepositoryIndex in interface ElasticSearchAdmin

optimizeIndex

public void optimizeIndex(String indexName)

Description copied from interface: ElasticSearchAdmin

Elasticsearch optimize operation allows to reduce the number of segments to one, Note that this can potentially be a very heavy operation.

Specified by:

optimizeIndex in interface ElasticSearchAdmin

indexNonRecursive

public void indexNonRecursive(IndexingCommand cmd)

Description copied from interface: ElasticSearchIndexing

Process the IndexingCommand.

Send indexing command to Elasticsearch, if the command is synchronous the index is refreshed so the document is searchable immediately. Recursive indexing is not taken in account except for deletion. This is not a transactional operation, a rollback will not discard the executed commands.

Specified by:

indexNonRecursive in interface ElasticSearchIndexing

indexNonRecursive

public void indexNonRecursive(List<IndexingCommand> cmds)

Description copied from interface: ElasticSearchIndexing

Same as ElasticSearchIndexing.indexNonRecursive(org.nuxeo.elasticsearch.commands.IndexingCommand) but process the list command using a bulk request.

Specified by:

indexNonRecursive in interface ElasticSearchIndexing

stackCommands

protected void stackCommands(List<IndexingCommand> cmds)

runIndexingWorker

public void runIndexingWorker(List<IndexingCommand> cmds)

Description copied from interface: ElasticSearchIndexing

Run a worker to process the IndexingCommand.

Asynchronous command schedules an indexing job and return.

Synchronous command execute an indexing job using a new Tx then refresh the index so the document is searchable immediately. if the command is also recursive the children are processed asynchronously.

If there is more than one cmd the elasticsearch request is done in bulk mode.

Specified by:

runIndexingWorker in interface ElasticSearchIndexing

dispatchWork

protected void dispatchWork(List<IndexingCommand> cmds)

Dispatch jobs between sync and async worker

scheduleIndexingAsyncWorker

protected void scheduleIndexingAsyncWorker(Map<String,List<IndexingCommand>> asyncCommands)

runIndexingSyncWorker

protected void runIndexingSyncWorker(Map<String,List<IndexingCommand>> syncCommands)

runReindexingWorker

public void runReindexingWorker(String repositoryName,
                                String nxql,
                                boolean syncAlias)

Description copied from interface: ElasticSearchIndexing

Reindex documents matching the NXQL query, This is done in an asynchronous job. When syncAlias is true a call is made to sync the search alias with write alias once indexing is done.

Specified by:

runReindexingWorker in interface ElasticSearchIndexing

reindexRepository

public void reindexRepository(String repositoryName)

Description copied from interface: ElasticSearchIndexing

Recreate an index and run an async reindexing worker.

Specified by:

reindexRepository in interface ElasticSearchIndexing

source

public org.elasticsearch.common.bytes.BytesReference source(DocumentModel doc)
                                                     throws IOException

Description copied from interface: ElasticSearchIndexing

Returns the JSON Elasticsearch source representation of a document.

Specified by:

source in interface ElasticSearchIndexing

Throws:

IOException

query

public DocumentModelList query(NxQueryBuilder queryBuilder)

Description copied from interface: ElasticSearchService

Returns a document list using an NxQueryBuilder.

Specified by:

query in interface ElasticSearchService

queryAndAggregate

public EsResult queryAndAggregate(NxQueryBuilder queryBuilder)

Description copied from interface: ElasticSearchService

Returns documents and aggregates.

Specified by:

queryAndAggregate in interface ElasticSearchService

scroll

public EsScrollResult scroll(NxQueryBuilder queryBuilder,
                             long keepAlive)

Description copied from interface: ElasticSearchService

Performs the initial search of a scrollable search request using an NxQueryBuilder.

Specified by:

scroll in interface ElasticSearchService

keepAlive - the search context lifetime

Returns:

an EsScrollResult including the search results and a scroll id, to be passed to the subsequent calls to ElasticSearchService.scroll(EsScrollResult)

scroll

public EsScrollResult scroll(EsScrollResult scrollResult)

Description copied from interface: ElasticSearchService

Retrieves the next batch of results of a scrollable search request for the given EsScrollResult.

Specified by:

scroll in interface ElasticSearchService

Returns:

an EsScrollResult including the search results and a scroll id, to be passed to the subsequent calls to {code scroll}.

clearScroll

public void clearScroll(EsScrollResult scrollResult)

Description copied from interface: ElasticSearchService

Clear scroll on ElasticSearch cluster for the given EsScrollResult.

Specified by:

clearScroll in interface ElasticSearchService

query

@Deprecated
public DocumentModelList query(CoreSession session,
                               String nxql,
                               int limit,
                               int offset,
                               SortInfo... sortInfos)

Deprecated.

Description copied from interface: ElasticSearchService

Returns a document list using an NXQL query. Fetch documents from the VCS repository.

Specified by:

query in interface ElasticSearchService

query

@Deprecated
public DocumentModelList query(CoreSession session,
                               org.elasticsearch.index.query.QueryBuilder queryBuilder,
                               int limit,
                               int offset,
                               SortInfo... sortInfos)

Deprecated.

Description copied from interface: ElasticSearchService

Returns a document list using an ElasticSearch QueryBuilder. Fetch documents from the VCS repository.

Specified by:

query in interface ElasticSearchService

isReady

public boolean isReady()

getHintByOperator

public Optional<ESHintQueryBuilder> getHintByOperator(String name)

Description copied from interface: ElasticSearchAdmin

Returns the hint by the Elasticsearch operator name.

Specified by:

getHintByOperator in interface ElasticSearchAdmin