ElasticsearchIndex.java
/*
* Licensed to The Apereo Foundation under one or more contributor license
* agreements. See the NOTICE file distributed with this work for additional
* information regarding copyright ownership.
*
*
* The Apereo Foundation licenses this file to you under the Educational
* Community License, Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain a copy of the License
* at:
*
* http://opensource.org/licenses/ecl2.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*
*/
package org.opencastproject.elasticsearch.index;
import static org.opencastproject.util.data.functions.Misc.chuck;
import org.opencastproject.elasticsearch.api.SearchIndexException;
import org.opencastproject.elasticsearch.api.SearchMetadata;
import org.opencastproject.elasticsearch.api.SearchResult;
import org.opencastproject.elasticsearch.impl.AbstractElasticsearchIndex;
import org.opencastproject.elasticsearch.impl.ElasticsearchDocument;
import org.opencastproject.elasticsearch.impl.SearchMetadataCollection;
import org.opencastproject.elasticsearch.index.objects.event.Event;
import org.opencastproject.elasticsearch.index.objects.event.EventIndexUtils;
import org.opencastproject.elasticsearch.index.objects.event.EventQueryBuilder;
import org.opencastproject.elasticsearch.index.objects.event.EventSearchQuery;
import org.opencastproject.elasticsearch.index.objects.series.Series;
import org.opencastproject.elasticsearch.index.objects.series.SeriesIndexUtils;
import org.opencastproject.elasticsearch.index.objects.series.SeriesQueryBuilder;
import org.opencastproject.elasticsearch.index.objects.series.SeriesSearchQuery;
import org.opencastproject.elasticsearch.index.objects.theme.IndexTheme;
import org.opencastproject.elasticsearch.index.objects.theme.ThemeQueryBuilder;
import org.opencastproject.elasticsearch.index.objects.theme.ThemeSearchQuery;
import org.opencastproject.list.api.ListProvidersService;
import org.opencastproject.security.api.User;
import com.google.common.util.concurrent.Striped;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.osgi.framework.BundleContext;
import org.osgi.service.component.ComponentException;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Deactivate;
import org.osgi.service.component.annotations.Modified;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;
import org.osgi.service.component.annotations.ReferencePolicyOption;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import javax.xml.bind.Unmarshaller;
/**
* An index implementation based on ElasticSearch that serves the Admin UI API and the External API with data
* aggregated from multiple services.
*/
@Component(
property = {
"service.description=Elasticsearch Index"
},
service = { ElasticsearchIndex.class }
)
public class ElasticsearchIndex extends AbstractElasticsearchIndex {
/** Retry configuration */
private int maxRetryAttemptsGet;
private static final String MAX_RETRY_ATTEMPTS_GET_PROPERTY = "max.retry.attempts.get";
private static final int DEFAULT_MAX_RETRY_ATTEMPTS_GET = 0;
private int maxRetryAttemptsUpdate;
private static final String MAX_RETRY_ATTEMPTS_UPDATE_PROPERTY = "max.retry.attempts.update";
private static final int DEFAULT_MAX_RETRY_ATTEMPTS_UPDATE = 0;
private int retryWaitingPeriodGet;
private static final String RETRY_WAITING_PERIOD_GET_PROPERTY = "retry.waiting.period.get";
private static final int DEFAULT_RETRY_WAITING_PERIOD_GET = 1000;
private int retryWaitingPeriodUpdate;
private static final String RETRY_WAITING_PERIOD_UPDATE_PROPERTY = "retry.waiting.period.update";
private static final int DEFAULT_RETRY_WAITING_PERIOD_UPDATE = 1000;
/** The required index version */
private static final int INDEX_VERSION = 101;
/** The document types */
private static final String VERSION_DOCUMENT_TYPE = "version";
private static final String[] DOCUMENT_TYPES = new String[] {
Event.DOCUMENT_TYPE,
Series.DOCUMENT_TYPE,
IndexTheme.DOCUMENT_TYPE,
VERSION_DOCUMENT_TYPE
};
private static final Logger logger = LoggerFactory.getLogger(ElasticsearchIndex.class);
private final Striped<Lock> locks = Striped.lazyWeakLock(1024);
private ListProvidersService listProvidersService;
private static final String CONFIG_EPISODE_ID_ROLE = "org.opencastproject.episode.id.role.access";
private boolean episodeIdRole = false;
@Reference(
cardinality = ReferenceCardinality.OPTIONAL,
policy = ReferencePolicy.DYNAMIC,
policyOption = ReferencePolicyOption.GREEDY
)
public void setListProvidersService(ListProvidersService listProvidersService) {
this.listProvidersService = listProvidersService;
}
public void unsetListProvidersService(ListProvidersService listProvidersService) {
if (this.listProvidersService == listProvidersService) {
this.listProvidersService = null;
}
}
/**
* OSGi callback to activate this component instance.
*
* @param bundleContext
* The bundle context
* @param properties
* The configuration
* @throws ComponentException
* If the search index cannot be initialized
*/
@Activate
public void activate(BundleContext bundleContext, Map<String, Object> properties) throws ComponentException {
super.activate(properties, bundleContext);
modified(properties);
try {
init(INDEX_VERSION);
} catch (Throwable t) {
throw new ComponentException("Error initializing elastic search index", t);
}
episodeIdRole = BooleanUtils.toBoolean(Objects.toString(
bundleContext.getProperty(CONFIG_EPISODE_ID_ROLE), "false"));
logger.debug("Usage of episode ID roles is set to {}", episodeIdRole);
}
/**
* OSGi callback to deactivate this component.
*
* @throws IOException
* If closing the index fails
*/
@Deactivate
public void deactivate() throws IOException {
close();
}
/**
* OSGi callback for configuration changes.
*
* @param properties
* The configuration
*/
@Modified
public void modified(Map<String, Object> properties) {
super.modified(properties);
maxRetryAttemptsGet = NumberUtils.toInt((String) properties.get(MAX_RETRY_ATTEMPTS_GET_PROPERTY),
DEFAULT_MAX_RETRY_ATTEMPTS_GET);
retryWaitingPeriodGet = NumberUtils.toInt((String) properties.get(RETRY_WAITING_PERIOD_GET_PROPERTY),
DEFAULT_RETRY_WAITING_PERIOD_GET);
logger.info("Max retry attempts for get requests set to {}, timeout set to {} ms.", maxRetryAttemptsGet,
retryWaitingPeriodGet);
maxRetryAttemptsUpdate = NumberUtils.toInt((String) properties.get(MAX_RETRY_ATTEMPTS_UPDATE_PROPERTY),
DEFAULT_MAX_RETRY_ATTEMPTS_UPDATE);
retryWaitingPeriodUpdate = NumberUtils.toInt((String) properties.get(RETRY_WAITING_PERIOD_UPDATE_PROPERTY),
DEFAULT_RETRY_WAITING_PERIOD_UPDATE);
logger.info("Max retry attempts for update requests set to {}, timeout set to {} ms.", maxRetryAttemptsUpdate,
retryWaitingPeriodUpdate);
if (maxRetryAttemptsGet < 0 || maxRetryAttemptsUpdate < 0 || retryWaitingPeriodGet < 0
|| retryWaitingPeriodUpdate < 0) {
logger.warn("You have configured negative values for max attempts or retry periods. Is this intended? This is "
+ "equivalent to setting those values to 0.");
}
}
/**
* @see AbstractElasticsearchIndex#getDocumentTypes()
*/
@Override
public String[] getDocumentTypes() {
return DOCUMENT_TYPES;
}
/*
* Get index objects
*/
/**
* Loads the event from the search index if it exists.
*
* @param mediaPackageId
* The media package identifier
* @param organization
* The organization
* @param user
* The user
* @return the event (optional)
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple events with the same identifier are found
*/
public Optional<Event> getEvent(String mediaPackageId, String organization, User user) throws SearchIndexException {
return getEvent(mediaPackageId, organization, user, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* Loads the event from the search index if it exists.
*
* @param mediaPackageId
* The media package identifier
* @param organization
* The organization
* @param user
* The user
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
* @return the event (optional)
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple events with the same identifier are found
*/
private Optional<Event> getEvent(String mediaPackageId, String organization, User user, int maxRetryAttempts,
int retryWaitingPeriod) throws SearchIndexException {
EventSearchQuery query = new EventSearchQuery(organization, user).withoutActions().withIdentifier(mediaPackageId);
SearchResult<Event> searchResult = getByQuery(query, maxRetryAttempts, retryWaitingPeriod);
if (searchResult.getDocumentCount() == 0) {
return Optional.empty();
} else if (searchResult.getDocumentCount() == 1) {
return Optional.of(searchResult.getItems()[0].getSource());
} else {
throw new IllegalStateException(
"Multiple events with identifier " + mediaPackageId + " found in search index");
}
}
/**
* Loads the series from the search index if it exists.
*
* @param seriesId
* The series identifier
* @param organization
* The organization
* @param user
* The user
* @return the series (optional)
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple series with the same identifier are found
*/
public Optional<Series> getSeries(String seriesId, String organization, User user)
throws SearchIndexException {
return getSeries(seriesId, organization, user, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* Loads the series from the search index if it exists.
*
* @param seriesId
* The series identifier
* @param organization
* The organization
* @param user
* The user
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
* @return the series (optional)
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple series with the same identifier are found
*/
private Optional<Series> getSeries(String seriesId, String organization, User user, int maxRetryAttempts,
int retryWaitingPeriod) throws SearchIndexException {
SeriesSearchQuery query = new SeriesSearchQuery(organization, user).withoutActions().withIdentifier(seriesId);
SearchResult<Series> searchResult = getByQuery(query, maxRetryAttempts, retryWaitingPeriod);
if (searchResult.getDocumentCount() == 0) {
return Optional.empty();
} else if (searchResult.getDocumentCount() == 1) {
return Optional.of(searchResult.getItems()[0].getSource());
} else {
throw new IllegalStateException("Multiple series with identifier " + seriesId + " found in search index");
}
}
/**
* Loads the theme from the search index if it exists.
*
* @param themeId
* The theme identifier
* @param organization
* The organization
* @param user
* The user
* @return the theme wrapped in an optional
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple themes with the same identifier are found
*/
public Optional<IndexTheme> getTheme(long themeId, String organization, User user)
throws SearchIndexException {
return getTheme(themeId, organization, user, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* Loads the theme from the search index if it exists.
*
* @param themeId
* The theme identifier
* @param organization
* The organization
* @param user
* The user
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
* @return the theme wrapped in an optional
*
* @throws SearchIndexException
* If querying the search index fails
* @throws IllegalStateException
* If multiple themes with the same identifier are found
*/
private Optional<IndexTheme> getTheme(long themeId, String organization, User user, int maxRetryAttempts,
int retryWaitingPeriod)
throws SearchIndexException {
ThemeSearchQuery query = new ThemeSearchQuery(organization, user).withIdentifier(themeId);
SearchResult<IndexTheme> searchResult = getByQuery(query, maxRetryAttempts, retryWaitingPeriod);
if (searchResult.getDocumentCount() == 0) {
return Optional.empty();
} else if (searchResult.getDocumentCount() == 1) {
return Optional.of(searchResult.getItems()[0].getSource());
} else {
throw new IllegalStateException("Multiple themes with identifier " + themeId + " found in search index");
}
}
/*
* Add or update index objects
*/
/**
* Adds or updates the event in the search index. Uses a locking mechanism to avoid issues like Lost Update.
*
* @param id
* The id of the event to update
* @param updateFunction
* The function that does the actual updating
* @param orgId
* The organization the event belongs to
* @param user
* The user
*
* @throws SearchIndexException
* Thrown if unable to update the event.
*/
public Optional<Event> addOrUpdateEvent(String id, Function<Optional<Event>, Optional<Event>> updateFunction,
String orgId, User user) throws SearchIndexException {
final Lock lock = this.locks.get(id);
lock.lock();
logger.debug("Locked event '{}'", id);
try {
Optional<Event> eventOpt = getEvent(id, orgId, user, maxRetryAttemptsUpdate, retryWaitingPeriodUpdate);
Optional<Event> updatedEventOpt = updateFunction.apply(eventOpt);
if (updatedEventOpt.isPresent()) {
update(updatedEventOpt.get());
}
return updatedEventOpt;
} finally {
lock.unlock();
logger.debug("Released locked event '{}'", id);
}
}
/**
* Adds the recording event to the search index or updates it accordingly if it is there.
*
* @param event
* The event to update
*
* @throws SearchIndexException
* If the event cannot be added or updated
*/
private void update(Event event) throws SearchIndexException {
logger.debug("Adding event {} to search index", event.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = EventIndexUtils.toSearchMetadata(event, listProvidersService,
episodeIdRole);
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
ElasticsearchDocument doc = new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata);
try {
update(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, doc);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write event " + event + " to index", t);
}
}
/**
* Adds the recording events to the search index or updates it accordingly if it is there.
*
* @param eventList
* The events to update
*
* @throws SearchIndexException
* If the events cannot be added or updated
*/
public void bulkEventUpdate(List<Event> eventList) throws SearchIndexException {
List<ElasticsearchDocument> docs = new ArrayList<>();
for (Event event: eventList) {
logger.debug("Adding event {} to search index", event.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = EventIndexUtils.toSearchMetadata(event, listProvidersService,
episodeIdRole);
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
docs.add(new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata));
}
try {
bulkUpdate(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, docs);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write events " + eventList + " to index", t);
}
}
/**
* Adds or updates the series in the search index. Uses a locking mechanism to avoid issues like Lost Update.
*
* @param id
* The id of the series to add
* @param updateFunction
* The function that does the actual updating
* @param orgId
* The organization the series belongs to
* @param user
* The user
*
* @throws SearchIndexException
* Thrown if unable to add or update the series.
*/
public Optional<Series> addOrUpdateSeries(String id, Function<Optional<Series>, Optional<Series>> updateFunction,
String orgId, User user) throws SearchIndexException {
final Lock lock = this.locks.get(id);
lock.lock();
logger.debug("Locked series '{}'", id);
try {
Optional<Series> seriesOpt = getSeries(id, orgId, user, maxRetryAttemptsUpdate, retryWaitingPeriodUpdate);
Optional<Series> updatedSeriesOpt = updateFunction.apply(seriesOpt);
if (updatedSeriesOpt.isPresent()) {
update(updatedSeriesOpt.get());
}
return updatedSeriesOpt;
} finally {
lock.unlock();
logger.debug("Released locked series '{}'", id);
}
}
/**
* Add or update a series in the search index.
*
* @param series
* The series to update
*
* @throws SearchIndexException
* If the series cannot be added or updated
*/
private void update(Series series) throws SearchIndexException {
logger.debug("Adding series {} to search index", series.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = SeriesIndexUtils.toSearchMetadata(series);
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
ElasticsearchDocument doc = new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata);
try {
update(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, doc);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write series " + series + " to index", t);
}
}
/**
* Add or update a range of series in the search index.
*
* @param seriesList
* The series to update
*
* @throws SearchIndexException
* If the series cannot be added or updated
*/
public void bulkSeriesUpdate(List<Series> seriesList) throws SearchIndexException {
List<ElasticsearchDocument> docs = new ArrayList<>();
for (Series series: seriesList) {
logger.debug("Adding series {} to search index", series.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = SeriesIndexUtils.toSearchMetadata(series);
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
docs.add(new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata));
}
try {
bulkUpdate(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, docs);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write series " + seriesList + " to index", t);
}
}
/**
* Adds or updates the theme in the search index. Uses a locking mechanism to avoid issues like Lost Update.
*
* @param id
* The id of the theme to update
* @param updateFunction
* The function that does the actual updating
* @param orgId
* The organization the theme belongs to
* @param user
* The user
*
* @throws SearchIndexException
* Thrown if unable to update the theme.
*/
public Optional<IndexTheme> addOrUpdateTheme(long id, Function<Optional<IndexTheme>,
Optional<IndexTheme>> updateFunction, String orgId, User user) throws SearchIndexException {
final Lock lock = this.locks.get(id);
lock.lock();
logger.debug("Locked theme '{}'", id);
try {
Optional<IndexTheme> themeOpt = getTheme(id, orgId, user, maxRetryAttemptsUpdate, retryWaitingPeriodUpdate);
Optional<IndexTheme> updatedThemeOpt = updateFunction.apply(themeOpt);
if (updatedThemeOpt.isPresent()) {
update(updatedThemeOpt.get());
}
return updatedThemeOpt;
} finally {
lock.unlock();
logger.debug("Released locked theme '{}'", id);
}
}
/**
* Adds or updates the theme in the search index.
*
* @param theme
* The theme to update
*
* @throws SearchIndexException
* Thrown if unable to add or update the theme.
*/
private void update(IndexTheme theme) throws SearchIndexException {
logger.debug("Adding theme {} to search index", theme.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = theme.toSearchMetadata();
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
ElasticsearchDocument doc = new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata);
try {
update(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, doc);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write theme " + theme + " to index", t);
}
}
/**
* Adds or updates the themes in the search index.
*
* @param themeList
* The themes to update
*
* @throws SearchIndexException
* Thrown if unable to add or update the themes.
*/
public void bulkThemeUpdate(List<IndexTheme> themeList) throws SearchIndexException {
List<ElasticsearchDocument> docs = new ArrayList<>();
for (IndexTheme theme: themeList) {
logger.debug("Adding theme {} to search index", theme.getIdentifier());
// Add the resource to the index
SearchMetadataCollection inputDocument = theme.toSearchMetadata();
List<SearchMetadata<?>> resourceMetadata = inputDocument.getMetadata();
docs.add(new ElasticsearchDocument(inputDocument.getIdentifier(),
inputDocument.getDocumentType(), resourceMetadata));
}
try {
bulkUpdate(maxRetryAttemptsUpdate, retryWaitingPeriodUpdate, docs);
} catch (Throwable t) {
throw new SearchIndexException("Cannot write themes " + themeList + " to index", t);
}
}
/*
* Delete index objects
*/
/**
* Delete event from index.
*
* @param eventId
* The event identifier
* @param orgId
* The organization id
* @return
* true if it was deleted, false if it couldn't be found
* @throws SearchIndexException
* If there was an error during deletion
*/
public boolean deleteEvent(String eventId, String orgId) throws SearchIndexException {
return delete(Event.DOCUMENT_TYPE, eventId, orgId);
}
/**
* Delete series from index.
*
* @param seriesId
* The series identifier
* @param orgId
* The organization id
* @return
* true if it was deleted, false if it couldn't be found
* @throws SearchIndexException
* If there was an error during deletion
*/
public boolean deleteSeries(String seriesId, String orgId) throws SearchIndexException {
return delete(Series.DOCUMENT_TYPE, seriesId, orgId);
}
/**
* Delete theme from index.
*
* @param themeId
* The theme identifier
* @param orgId
* The organization id
* @return
* true if it was deleted, false if it couldn't be found
* @throws SearchIndexException
* If there was an error during deletion
*/
public boolean deleteTheme(String themeId, String orgId) throws SearchIndexException {
return delete(IndexTheme.DOCUMENT_TYPE, themeId, orgId);
}
/**
* Delete object from this index.
*
* @param type
* The type of object we want to delete
* @param id
* The identifier of this object
* @param orgId
* The organization id
* @return
* True if it was deleted, false if it couldn't be found
*
* @throws SearchIndexException
* If deleting from the index fails
*/
private boolean delete(String type, String id, String orgId) throws SearchIndexException {
final Lock lock = this.locks.get(id);
lock.lock();
logger.debug("Locked {} '{}'.", type, id);
try {
String idWithOrgId = id.concat(orgId);
logger.debug("Removing element with id '{}' from search index '{}'", idWithOrgId, getSubIndexIdentifier(type));
DeleteResponse deleteResponse = delete(type, idWithOrgId, maxRetryAttemptsUpdate, retryWaitingPeriodUpdate);
if (deleteResponse.getResult().equals(DocWriteResponse.Result.NOT_FOUND)) {
logger.trace("Document {} to delete was not found on index '{}'", idWithOrgId, getSubIndexIdentifier(type));
return false;
}
} catch (Throwable e) {
throw new SearchIndexException("Cannot remove " + type + " " + id + " from index", e);
} finally {
lock.unlock();
logger.debug("Released locked {} '{}'.", type, id);
}
return true;
}
/*
* Get index objects by query
*/
/**
* @param query
* The query to use to retrieve the events that match the query
* @return {@link SearchResult} collection of {@link Event} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
public SearchResult<Event> getByQuery(EventSearchQuery query) throws SearchIndexException {
return getByQuery(query, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* @param query
* The query to use to retrieve the events that match the query
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
* @return {@link SearchResult} collection of {@link Event} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
private SearchResult<Event> getByQuery(EventSearchQuery query, int maxRetryAttempts, int retryWaitingPeriod)
throws SearchIndexException {
logger.debug("Searching index using event query '{}'", query);
// Create the request
final SearchRequest searchRequest = getSearchRequest(query, new EventQueryBuilder(query));
try {
final Unmarshaller unmarshaller = Event.createUnmarshaller();
return executeQuery(query, searchRequest, metadata -> {
try {
return EventIndexUtils.toRecordingEvent(metadata, unmarshaller);
} catch (IOException e) {
return chuck(e);
}
}, maxRetryAttempts, retryWaitingPeriod);
} catch (Throwable t) {
throw new SearchIndexException("Error querying event index", t);
}
}
/**
* @param query
* The query to use to retrieve the series that match the query
* @return {@link SearchResult} collection of {@link Series} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
public SearchResult<Series> getByQuery(SeriesSearchQuery query) throws SearchIndexException {
return getByQuery(query, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* @param query
* The query to use to retrieve the series that match the query
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
*
* @return {@link SearchResult} collection of {@link Series} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
private SearchResult<Series> getByQuery(SeriesSearchQuery query, int maxRetryAttempts, int retryWaitingPeriod)
throws SearchIndexException {
logger.debug("Searching index using series query '{}'", query);
// Create the request
final SearchRequest searchRequest = getSearchRequest(query, new SeriesQueryBuilder(query));
try {
final Unmarshaller unmarshaller = Series.createUnmarshaller();
return executeQuery(query, searchRequest, metadata -> {
try {
return SeriesIndexUtils.toSeries(metadata, unmarshaller);
} catch (IOException e) {
return chuck(e);
}
}, maxRetryAttempts, retryWaitingPeriod);
} catch (Throwable t) {
throw new SearchIndexException("Error querying series index", t);
}
}
/**
* @param query
* The query to use to retrieve the themes that match the query
* @return {@link SearchResult} collection of {@link IndexTheme} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
public SearchResult<IndexTheme> getByQuery(ThemeSearchQuery query) throws SearchIndexException {
return getByQuery(query, maxRetryAttemptsGet, retryWaitingPeriodGet);
}
/**
* @param query
* The query to use to retrieve the themes that match the query
* @param maxRetryAttempts
* How often to retry query in case of ElasticsearchStatusException
* @param retryWaitingPeriod
* How long to wait (in ms) between retries
*
* @return {@link SearchResult} collection of {@link IndexTheme} from a query.
*
* @throws SearchIndexException
* Thrown if there is an error getting the results.
*/
private SearchResult<IndexTheme> getByQuery(ThemeSearchQuery query, int maxRetryAttempts, int retryWaitingPeriod)
throws SearchIndexException {
logger.debug("Searching index using theme query '{}'", query);
// Create the request
final SearchRequest searchRequest = getSearchRequest(query, new ThemeQueryBuilder(query));
try {
return executeQuery(query, searchRequest, metadata -> {
try {
return IndexTheme.fromSearchMetadata(metadata);
} catch (IOException e) {
return chuck(e);
}
}, maxRetryAttempts, retryWaitingPeriod);
} catch (Throwable t) {
throw new SearchIndexException("Error querying theme index", t);
}
}
}