Scheduler for regular search evaluation runs #220
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
| package org.opensearch.searchrelevance.dao; | ||
|
|
||
| import static org.opensearch.searchrelevance.indices.SearchRelevanceIndices.SCHEDULED_JOBS; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.opensearch.action.StepListener; | ||
| import org.opensearch.action.delete.DeleteResponse; | ||
| import org.opensearch.action.search.SearchResponse; | ||
| import org.opensearch.common.xcontent.XContentFactory; | ||
| import org.opensearch.core.action.ActionListener; | ||
| import org.opensearch.core.rest.RestStatus; | ||
| import org.opensearch.core.xcontent.ToXContent; | ||
| import org.opensearch.index.query.QueryBuilders; | ||
| import org.opensearch.search.builder.SearchSourceBuilder; | ||
| import org.opensearch.searchrelevance.exception.SearchRelevanceException; | ||
| import org.opensearch.searchrelevance.indices.SearchRelevanceIndicesManager; | ||
| import org.opensearch.searchrelevance.model.ScheduledJob; | ||
|
|
||
| public class ScheduledJobsDao { | ||
| private static final Logger LOGGER = LogManager.getLogger(ScheduledJobsDao.class); | ||
| private final SearchRelevanceIndicesManager searchRelevanceIndicesManager; | ||
|
|
||
| public ScheduledJobsDao(SearchRelevanceIndicesManager searchRelevanceIndicesManager) { | ||
| this.searchRelevanceIndicesManager = searchRelevanceIndicesManager; | ||
| } | ||
|
|
||
| /** | ||
| * Create scheduled jobs index if not exists | ||
| * @param stepListener - step lister for async operation | ||
| */ | ||
| public void createIndexIfAbsent(final StepListener<Void> stepListener) { | ||
epugh marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| searchRelevanceIndicesManager.createIndexIfAbsent(SCHEDULED_JOBS, stepListener); | ||
| } | ||
|
|
||
| /** | ||
| * Stores scheduled job to in the system index | ||
| * @param scheduledJob - Scheduled job content to be stored | ||
| * @param listener - action lister for async operation | ||
| */ | ||
| public void putScheduledJob(final ScheduledJob scheduledJob, final ActionListener listener) { | ||
| if (scheduledJob == null) { | ||
| listener.onFailure(new SearchRelevanceException("Scheduled job cannot be null", RestStatus.BAD_REQUEST)); | ||
| return; | ||
| } | ||
| try { | ||
| searchRelevanceIndicesManager.putDoc( | ||
| scheduledJob.getId(), | ||
| scheduledJob.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS), | ||
| SCHEDULED_JOBS, | ||
| listener | ||
| ); | ||
| } catch (IOException e) { | ||
| throw new SearchRelevanceException("Failed to store scheduled job", e, RestStatus.INTERNAL_SERVER_ERROR); | ||
| } | ||
| } | ||
|
|
||
| public void updateScheduledJob(final ScheduledJob scheduledJob, final ActionListener listener) { | ||
| if (scheduledJob == null) { | ||
| listener.onFailure(new SearchRelevanceException("Scheduled job cannot be null", RestStatus.BAD_REQUEST)); | ||
| return; | ||
| } | ||
| try { | ||
| searchRelevanceIndicesManager.updateDoc( | ||
| scheduledJob.getId(), | ||
| scheduledJob.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS), | ||
| SCHEDULED_JOBS, | ||
| listener | ||
| ); | ||
| } catch (IOException e) { | ||
| throw new SearchRelevanceException("Failed to store scheduled job", e, RestStatus.INTERNAL_SERVER_ERROR); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Delete scheduled job by scheduledJobId | ||
| * @param scheduledJobId - id to be deleted | ||
| * @param listener - action lister for async operation | ||
| */ | ||
| public void deleteScheduledJob(final String scheduledJobId, final ActionListener<DeleteResponse> listener) { | ||
| searchRelevanceIndicesManager.deleteDocByDocId(scheduledJobId, SCHEDULED_JOBS, listener); | ||
| } | ||
|
|
||
| /** | ||
| * Get scheduled job by scheduledJobId | ||
| * @param scheduledJobId - id to be deleted | ||
| * @param listener - action lister for async operation | ||
| */ | ||
| public SearchResponse getScheduledJob(String scheduledJobId, ActionListener<SearchResponse> listener) { | ||
| if (scheduledJobId == null || scheduledJobId.isEmpty()) { | ||
| listener.onFailure(new SearchRelevanceException("jobId must not be null or empty", RestStatus.BAD_REQUEST)); | ||
| return null; | ||
| } | ||
| return searchRelevanceIndicesManager.getDocByDocId(scheduledJobId, SCHEDULED_JOBS, listener); | ||
| } | ||
|
|
||
| /** | ||
| * List scheduled jobs by source builder | ||
| * @param sourceBuilder - source builder to be searched | ||
| * @param listener - action lister for async operation | ||
ajleong623 marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| */ | ||
| public SearchResponse listScheduledJob(SearchSourceBuilder sourceBuilder, ActionListener<SearchResponse> listener) { | ||
| // Apply default values if not set | ||
| if (sourceBuilder == null) { | ||
| sourceBuilder = new SearchSourceBuilder(); | ||
| } | ||
|
|
||
| // Ensure we have a query | ||
| if (sourceBuilder.query() == null) { | ||
| sourceBuilder.query(QueryBuilders.matchAllQuery()); | ||
| } | ||
|
|
||
| return searchRelevanceIndicesManager.listDocsBySearchRequest(sourceBuilder, SCHEDULED_JOBS, listener); | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| */ | ||
| package org.opensearch.searchrelevance.model; | ||
|
|
||
| import java.io.IOException; | ||
| import java.time.Instant; | ||
|
|
||
| import org.opensearch.core.xcontent.ToXContentObject; | ||
| import org.opensearch.core.xcontent.XContentBuilder; | ||
| import org.opensearch.jobscheduler.spi.schedule.Schedule; | ||
|
|
||
| import lombok.AllArgsConstructor; | ||
| import lombok.Getter; | ||
|
|
||
| @Getter | ||
| @AllArgsConstructor | ||
| public class ScheduledJob implements ToXContentObject { | ||
| public static final String ID = "id"; | ||
| public static final String NAME_FIELD = "name"; | ||
| public static final String ENABLED_FILED = "enabled"; | ||
| public static final String LAST_UPDATE_TIME_FIELD = "last_update_time"; | ||
| public static final String LAST_UPDATE_TIME_FIELD_READABLE = "last_update_time_field"; | ||
| public static final String SCHEDULE_FIELD = "schedule"; | ||
| public static final String ENABLED_TIME_FILED = "enabled_time"; | ||
| public static final String ENABLED_TIME_FILED_READABLE = "enabled_time_field"; | ||
| public static final String INDEX_NAME_FIELD = "index_name_to_watch"; | ||
| public static final String LOCK_DURATION_SECONDS = "lock_duration_seconds"; | ||
| public static final String JITTER = "jitter"; | ||
| public static final String EXPERIMENT_ID = "experiment_id"; | ||
|
|
||
| private final String id; | ||
| private final String jobName; | ||
| private final Instant lastUpdateTime; | ||
| private final Instant enabledTime; | ||
| private final boolean isEnabled; | ||
| private final Schedule schedule; | ||
| private final String indexToWatch; | ||
| private final Long lockDurationSeconds; | ||
| private final Double jitter; | ||
| private final String experimentId; | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder.startObject(); | ||
| builder.field(ID, this.id) | ||
| .field(NAME_FIELD, this.jobName) | ||
| .field(ENABLED_FILED, this.isEnabled) | ||
| .field(SCHEDULE_FIELD, this.schedule) | ||
| .field(INDEX_NAME_FIELD, this.indexToWatch); | ||
| if (this.enabledTime != null) { | ||
| builder.timeField(ENABLED_TIME_FILED, ENABLED_TIME_FILED_READABLE, this.enabledTime.toEpochMilli()); | ||
| } | ||
| if (this.lastUpdateTime != null) { | ||
| builder.timeField(LAST_UPDATE_TIME_FIELD, LAST_UPDATE_TIME_FIELD_READABLE, this.lastUpdateTime.toEpochMilli()); | ||
| } | ||
| if (this.lockDurationSeconds != null) { | ||
| builder.field(LOCK_DURATION_SECONDS, this.lockDurationSeconds); | ||
| } | ||
| if (this.jitter != null) { | ||
| builder.field(JITTER, this.jitter); | ||
| } | ||
| if (this.experimentId != null) { | ||
| builder.field(EXPERIMENT_ID, this.experimentId); | ||
| } | ||
| builder.endObject(); | ||
| return builder; | ||
| } | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What's the dependency on the non-SPI jar for?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Removed.