[Backport 2.x] [Point in time] Backport point in time changes #4406
+8,097
−31
Next
Next commit
Adding create pit service layer changes
Signed-off-by: Bharathwaj G <bharath78910@gmail.com>
commit ff7d47f4d799e750591eb269fdba5d4ef43f315f
There are no files selected for viewing
This file contains 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 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 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 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,23 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.action.ActionType; | ||
|
|
||
| /** | ||
| * Action type for creating PIT reader context | ||
| */ | ||
| public class CreatePitAction extends ActionType<CreatePitResponse> { | ||
| public static final CreatePitAction INSTANCE = new CreatePitAction(); | ||
| public static final String NAME = "indices:data/read/point_in_time/create"; | ||
|
|
||
| private CreatePitAction() { | ||
| super(NAME, CreatePitResponse::new); | ||
| } | ||
| } |
This file contains 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,255 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.opensearch.OpenSearchException; | ||
| import org.opensearch.action.ActionListener; | ||
| import org.opensearch.action.StepListener; | ||
| import org.opensearch.action.support.GroupedActionListener; | ||
| import org.opensearch.cluster.ClusterState; | ||
| import org.opensearch.cluster.node.DiscoveryNode; | ||
| import org.opensearch.cluster.service.ClusterService; | ||
| import org.opensearch.common.Strings; | ||
| import org.opensearch.common.inject.Inject; | ||
| import org.opensearch.common.io.stream.NamedWriteableRegistry; | ||
| import org.opensearch.common.settings.Setting; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.index.shard.ShardId; | ||
| import org.opensearch.search.SearchPhaseResult; | ||
| import org.opensearch.search.SearchShardTarget; | ||
| import org.opensearch.tasks.Task; | ||
| import org.opensearch.transport.Transport; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.Collection; | ||
| import java.util.Collections; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.function.BiFunction; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import static org.opensearch.common.unit.TimeValue.timeValueSeconds; | ||
|
|
||
| /** | ||
| * Controller for creating PIT reader context | ||
| * Phase 1 of create PIT request : Create PIT reader contexts in the associated shards with a temporary keep alive | ||
| * Phase 2 of create PIT : Update PIT reader context with PIT ID and keep alive from request and | ||
| * fail user request if any of the updates in this phase are failed - we clean up PITs in case of such failures. | ||
| * This two phase approach is used to save PIT ID as part of context which is later used for other use cases like list PIT etc. | ||
| */ | ||
| public class CreatePitController { | ||
| private final SearchTransportService searchTransportService; | ||
| private final ClusterService clusterService; | ||
| private final TransportSearchAction transportSearchAction; | ||
| private final NamedWriteableRegistry namedWriteableRegistry; | ||
| private static final Logger logger = LogManager.getLogger(CreatePitController.class); | ||
| public static final Setting<TimeValue> PIT_INIT_KEEP_ALIVE = Setting.positiveTimeSetting( | ||
| "point_in_time.init.keep_alive", | ||
| timeValueSeconds(30), | ||
| Setting.Property.NodeScope | ||
| ); | ||
|
|
||
| @Inject | ||
| public CreatePitController( | ||
| SearchTransportService searchTransportService, | ||
| ClusterService clusterService, | ||
| TransportSearchAction transportSearchAction, | ||
| NamedWriteableRegistry namedWriteableRegistry | ||
| ) { | ||
| this.searchTransportService = searchTransportService; | ||
| this.clusterService = clusterService; | ||
| this.transportSearchAction = transportSearchAction; | ||
| this.namedWriteableRegistry = namedWriteableRegistry; | ||
| } | ||
|
|
||
| /** | ||
| * This method creates PIT reader context | ||
| */ | ||
| public void executeCreatePit( | ||
| CreatePitRequest request, | ||
| Task task, | ||
| StepListener<SearchResponse> createPitListener, | ||
| ActionListener<CreatePitResponse> updatePitIdListener | ||
| ) { | ||
| SearchRequest searchRequest = new SearchRequest(request.getIndices()); | ||
| searchRequest.preference(request.getPreference()); | ||
| searchRequest.routing(request.getRouting()); | ||
| searchRequest.indicesOptions(request.getIndicesOptions()); | ||
| searchRequest.allowPartialSearchResults(request.shouldAllowPartialPitCreation()); | ||
| SearchTask searchTask = searchRequest.createTask( | ||
| task.getId(), | ||
| task.getType(), | ||
| task.getAction(), | ||
| task.getParentTaskId(), | ||
| Collections.emptyMap() | ||
| ); | ||
| /** | ||
| * Phase 1 of create PIT | ||
| */ | ||
| executeCreatePit(searchTask, searchRequest, createPitListener); | ||
|
|
||
| /** | ||
| * Phase 2 of create PIT where we update pit id in pit contexts | ||
| */ | ||
| createPitListener.whenComplete( | ||
| searchResponse -> { executeUpdatePitId(request, searchRequest, searchResponse, updatePitIdListener); }, | ||
| updatePitIdListener::onFailure | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * Creates PIT reader context with temporary keep alive | ||
| */ | ||
| void executeCreatePit(Task task, SearchRequest searchRequest, StepListener<SearchResponse> createPitListener) { | ||
| logger.debug( | ||
| () -> new ParameterizedMessage("Executing creation of PIT context for indices [{}]", Arrays.toString(searchRequest.indices())) | ||
| ); | ||
| transportSearchAction.executeRequest( | ||
| task, | ||
| searchRequest, | ||
| TransportCreatePitAction.CREATE_PIT_ACTION, | ||
| true, | ||
| new TransportSearchAction.SinglePhaseSearchAction() { | ||
| @Override | ||
| public void executeOnShardTarget( | ||
| SearchTask searchTask, | ||
| SearchShardTarget target, | ||
| Transport.Connection connection, | ||
| ActionListener<SearchPhaseResult> searchPhaseResultActionListener | ||
| ) { | ||
| searchTransportService.createPitContext( | ||
| connection, | ||
| new TransportCreatePitAction.CreateReaderContextRequest( | ||
| target.getShardId(), | ||
| PIT_INIT_KEEP_ALIVE.get(clusterService.getSettings()) | ||
| ), | ||
| searchTask, | ||
| ActionListener.wrap(r -> searchPhaseResultActionListener.onResponse(r), searchPhaseResultActionListener::onFailure) | ||
| ); | ||
| } | ||
| }, | ||
| createPitListener | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * Updates PIT ID, keep alive and createdTime of PIT reader context | ||
| */ | ||
| void executeUpdatePitId( | ||
| CreatePitRequest request, | ||
| SearchRequest searchRequest, | ||
| SearchResponse searchResponse, | ||
| ActionListener<CreatePitResponse> updatePitIdListener | ||
| ) { | ||
| logger.debug( | ||
| () -> new ParameterizedMessage( | ||
| "Updating PIT context with PIT ID [{}], creation time and keep alive", | ||
| searchResponse.pointInTimeId() | ||
| ) | ||
| ); | ||
| /** | ||
| * store the create time ( same create time for all PIT contexts across shards ) to be used | ||
| * for list PIT api | ||
| */ | ||
| final long relativeStartNanos = System.nanoTime(); | ||
| final TransportSearchAction.SearchTimeProvider timeProvider = new TransportSearchAction.SearchTimeProvider( | ||
| searchRequest.getOrCreateAbsoluteStartMillis(), | ||
| relativeStartNanos, | ||
| System::nanoTime | ||
| ); | ||
| final long creationTime = timeProvider.getAbsoluteStartMillis(); | ||
| CreatePitResponse createPITResponse = new CreatePitResponse( | ||
| searchResponse.pointInTimeId(), | ||
| creationTime, | ||
| searchResponse.getTotalShards(), | ||
| searchResponse.getSuccessfulShards(), | ||
| searchResponse.getSkippedShards(), | ||
| searchResponse.getFailedShards(), | ||
| searchResponse.getShardFailures() | ||
| ); | ||
| SearchContextId contextId = SearchContextId.decode(namedWriteableRegistry, createPITResponse.getId()); | ||
| final StepListener<BiFunction<String, String, DiscoveryNode>> lookupListener = getConnectionLookupListener(contextId); | ||
| lookupListener.whenComplete(nodelookup -> { | ||
| final ActionListener<UpdatePitContextResponse> groupedActionListener = getGroupedListener( | ||
| updatePitIdListener, | ||
| createPITResponse, | ||
| contextId.shards().size(), | ||
| contextId.shards().values() | ||
| ); | ||
| for (Map.Entry<ShardId, SearchContextIdForNode> entry : contextId.shards().entrySet()) { | ||
| DiscoveryNode node = nodelookup.apply(entry.getValue().getClusterAlias(), entry.getValue().getNode()); | ||
| try { | ||
| final Transport.Connection connection = searchTransportService.getConnection(entry.getValue().getClusterAlias(), node); | ||
| searchTransportService.updatePitContext( | ||
| connection, | ||
| new UpdatePitContextRequest( | ||
| entry.getValue().getSearchContextId(), | ||
| createPITResponse.getId(), | ||
| request.getKeepAlive().millis(), | ||
| creationTime | ||
| ), | ||
| groupedActionListener | ||
| ); | ||
| } catch (Exception e) { | ||
| logger.error( | ||
| () -> new ParameterizedMessage( | ||
| "Create pit update phase failed for PIT ID [{}] on node [{}]", | ||
| searchResponse.pointInTimeId(), | ||
| node | ||
| ), | ||
| e | ||
| ); | ||
| groupedActionListener.onFailure( | ||
| new OpenSearchException( | ||
| "Create pit update phase for PIT ID [" + searchResponse.pointInTimeId() + "] failed on node[" + node + "]", | ||
| e | ||
| ) | ||
| ); | ||
| } | ||
| } | ||
| }, updatePitIdListener::onFailure); | ||
| } | ||
|
|
||
| private StepListener<BiFunction<String, String, DiscoveryNode>> getConnectionLookupListener(SearchContextId contextId) { | ||
| ClusterState state = clusterService.state(); | ||
| final Set<String> clusters = contextId.shards() | ||
| .values() | ||
| .stream() | ||
| .filter(ctx -> Strings.isEmpty(ctx.getClusterAlias()) == false) | ||
| .map(SearchContextIdForNode::getClusterAlias) | ||
| .collect(Collectors.toSet()); | ||
| return (StepListener<BiFunction<String, String, DiscoveryNode>>) SearchUtils.getConnectionLookupListener( | ||
| searchTransportService.getRemoteClusterService(), | ||
| state, | ||
| clusters | ||
| ); | ||
| } | ||
|
|
||
| private ActionListener<UpdatePitContextResponse> getGroupedListener( | ||
| ActionListener<CreatePitResponse> updatePitIdListener, | ||
| CreatePitResponse createPITResponse, | ||
| int size, | ||
| Collection<SearchContextIdForNode> contexts | ||
| ) { | ||
| return new GroupedActionListener<>(new ActionListener<>() { | ||
| @Override | ||
| public void onResponse(final Collection<UpdatePitContextResponse> responses) { | ||
| updatePitIdListener.onResponse(createPITResponse); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(final Exception e) { | ||
| updatePitIdListener.onFailure(e); | ||
| } | ||
| }, size); | ||
| } | ||
| } |
This file contains 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,195 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.action.ActionRequest; | ||
| import org.opensearch.action.ActionRequestValidationException; | ||
| import org.opensearch.action.IndicesRequest; | ||
| import org.opensearch.action.support.IndicesOptions; | ||
| import org.opensearch.common.Nullable; | ||
| import org.opensearch.common.Strings; | ||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.common.xcontent.ToXContent; | ||
| import org.opensearch.common.xcontent.XContentBuilder; | ||
| import org.opensearch.tasks.Task; | ||
| import org.opensearch.tasks.TaskId; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Map; | ||
| import java.util.Objects; | ||
|
|
||
| import static org.opensearch.action.ValidateActions.addValidationError; | ||
|
|
||
| /** | ||
| * A request to make create point in time against one or more indices. | ||
| */ | ||
| public class CreatePitRequest extends ActionRequest implements IndicesRequest.Replaceable, ToXContent { | ||
|
|
||
| // keep alive for pit reader context | ||
| private TimeValue keepAlive; | ||
|
|
||
| // this describes whether PIT can be created with partial failures | ||
| private Boolean allowPartialPitCreation; | ||
| @Nullable | ||
| private String routing = null; | ||
| @Nullable | ||
| private String preference = null; | ||
| private String[] indices = Strings.EMPTY_ARRAY; | ||
| private IndicesOptions indicesOptions = SearchRequest.DEFAULT_INDICES_OPTIONS; | ||
|
|
||
| public CreatePitRequest(TimeValue keepAlive, Boolean allowPartialPitCreation, String... indices) { | ||
| this.keepAlive = keepAlive; | ||
| this.allowPartialPitCreation = allowPartialPitCreation; | ||
| this.indices = indices; | ||
| } | ||
|
|
||
| public CreatePitRequest(StreamInput in) throws IOException { | ||
| super(in); | ||
| indices = in.readStringArray(); | ||
| indicesOptions = IndicesOptions.readIndicesOptions(in); | ||
| routing = in.readOptionalString(); | ||
| preference = in.readOptionalString(); | ||
| keepAlive = in.readTimeValue(); | ||
| allowPartialPitCreation = in.readOptionalBoolean(); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| out.writeStringArray(indices); | ||
| indicesOptions.writeIndicesOptions(out); | ||
| out.writeOptionalString(routing); | ||
| out.writeOptionalString(preference); | ||
| out.writeTimeValue(keepAlive); | ||
| out.writeOptionalBoolean(allowPartialPitCreation); | ||
| } | ||
|
|
||
| public String getRouting() { | ||
| return routing; | ||
| } | ||
|
|
||
| public String getPreference() { | ||
| return preference; | ||
| } | ||
|
|
||
| public String[] getIndices() { | ||
| return indices; | ||
| } | ||
|
|
||
| public IndicesOptions getIndicesOptions() { | ||
| return indicesOptions; | ||
| } | ||
|
|
||
| public TimeValue getKeepAlive() { | ||
| return keepAlive; | ||
| } | ||
|
|
||
| /** | ||
| * Sets if this request should allow partial results. | ||
| */ | ||
| public void allowPartialPitCreation(Boolean allowPartialPitCreation) { | ||
| this.allowPartialPitCreation = allowPartialPitCreation; | ||
| } | ||
|
|
||
| public boolean shouldAllowPartialPitCreation() { | ||
| return allowPartialPitCreation; | ||
| } | ||
|
|
||
| public void setRouting(String routing) { | ||
| this.routing = routing; | ||
| } | ||
|
|
||
| public void setPreference(String preference) { | ||
| this.preference = preference; | ||
| } | ||
|
|
||
| public void setIndices(String[] indices) { | ||
| this.indices = indices; | ||
| } | ||
|
|
||
| public void setIndicesOptions(IndicesOptions indicesOptions) { | ||
| this.indicesOptions = Objects.requireNonNull(indicesOptions, "indicesOptions must not be null"); | ||
| } | ||
|
|
||
| @Override | ||
| public ActionRequestValidationException validate() { | ||
| ActionRequestValidationException validationException = null; | ||
| if (keepAlive == null) { | ||
| validationException = addValidationError("keep alive not specified", validationException); | ||
| } | ||
| return validationException; | ||
| } | ||
|
|
||
| @Override | ||
| public String[] indices() { | ||
| return indices; | ||
| } | ||
|
|
||
| @Override | ||
| public IndicesOptions indicesOptions() { | ||
| return indicesOptions; | ||
| } | ||
|
|
||
| public CreatePitRequest indicesOptions(IndicesOptions indicesOptions) { | ||
| this.indicesOptions = Objects.requireNonNull(indicesOptions, "indicesOptions must not be null"); | ||
| return this; | ||
| } | ||
|
|
||
| public void setKeepAlive(TimeValue keepAlive) { | ||
| this.keepAlive = keepAlive; | ||
| } | ||
|
|
||
| public final String buildDescription() { | ||
| StringBuilder sb = new StringBuilder(); | ||
| sb.append("indices["); | ||
| Strings.arrayToDelimitedString(indices, ",", sb); | ||
| sb.append("], "); | ||
| sb.append("pointintime[").append(keepAlive).append("], "); | ||
| sb.append("allowPartialPitCreation[").append(allowPartialPitCreation).append("], "); | ||
| return sb.toString(); | ||
| } | ||
|
|
||
| @Override | ||
| public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) { | ||
| return new Task(id, type, action, this.buildDescription(), parentTaskId, headers); | ||
| } | ||
|
|
||
| private void validateIndices(String... indices) { | ||
| Objects.requireNonNull(indices, "indices must not be null"); | ||
| for (String index : indices) { | ||
| Objects.requireNonNull(index, "index must not be null"); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public CreatePitRequest indices(String... indices) { | ||
| validateIndices(indices); | ||
| this.indices = indices; | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder.field("keep_alive", keepAlive); | ||
| builder.field("allow_partial_pit_creation", allowPartialPitCreation); | ||
| if (indices != null) { | ||
| builder.startArray("indices"); | ||
| for (String index : indices) { | ||
| builder.value(index); | ||
| } | ||
| builder.endArray(); | ||
| } | ||
| if (indicesOptions != null) { | ||
| indicesOptions.toXContent(builder, params); | ||
| } | ||
| return builder; | ||
| } | ||
| } |
This file contains 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,232 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.action.ActionResponse; | ||
| import org.opensearch.common.ParseField; | ||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.common.xcontent.StatusToXContentObject; | ||
| import org.opensearch.common.xcontent.XContentBuilder; | ||
| import org.opensearch.common.xcontent.XContentParser; | ||
| import org.opensearch.rest.RestStatus; | ||
| import org.opensearch.rest.action.RestActions; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| import static org.opensearch.common.xcontent.XContentParserUtils.ensureExpectedToken; | ||
|
|
||
| /** | ||
| * Create point in time response with point in time id and shard success / failures | ||
| */ | ||
| public class CreatePitResponse extends ActionResponse implements StatusToXContentObject { | ||
| private static final ParseField ID = new ParseField("id"); | ||
| private static final ParseField CREATION_TIME = new ParseField("creation_time"); | ||
|
|
||
| // point in time id | ||
| private final String id; | ||
| private final int totalShards; | ||
| private final int successfulShards; | ||
| private final int failedShards; | ||
| private final int skippedShards; | ||
| private final ShardSearchFailure[] shardFailures; | ||
| private final long creationTime; | ||
|
|
||
| public CreatePitResponse(StreamInput in) throws IOException { | ||
| super(in); | ||
| id = in.readString(); | ||
| totalShards = in.readVInt(); | ||
| successfulShards = in.readVInt(); | ||
| failedShards = in.readVInt(); | ||
| skippedShards = in.readVInt(); | ||
| creationTime = in.readLong(); | ||
| int size = in.readVInt(); | ||
| if (size == 0) { | ||
| shardFailures = ShardSearchFailure.EMPTY_ARRAY; | ||
| } else { | ||
| shardFailures = new ShardSearchFailure[size]; | ||
| for (int i = 0; i < shardFailures.length; i++) { | ||
| shardFailures[i] = ShardSearchFailure.readShardSearchFailure(in); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public CreatePitResponse( | ||
| String id, | ||
| long creationTime, | ||
| int totalShards, | ||
| int successfulShards, | ||
| int skippedShards, | ||
| int failedShards, | ||
| ShardSearchFailure[] shardFailures | ||
| ) { | ||
| this.id = id; | ||
| this.creationTime = creationTime; | ||
| this.totalShards = totalShards; | ||
| this.successfulShards = successfulShards; | ||
| this.skippedShards = skippedShards; | ||
| this.failedShards = failedShards; | ||
| this.shardFailures = shardFailures; | ||
| } | ||
|
|
||
| @Override | ||
| public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException { | ||
| builder.startObject(); | ||
| builder.field(ID.getPreferredName(), id); | ||
| RestActions.buildBroadcastShardsHeader( | ||
| builder, | ||
| params, | ||
| getTotalShards(), | ||
| getSuccessfulShards(), | ||
| getSkippedShards(), | ||
| getFailedShards(), | ||
| getShardFailures() | ||
| ); | ||
| builder.field(CREATION_TIME.getPreferredName(), creationTime); | ||
| builder.endObject(); | ||
| return builder; | ||
| } | ||
|
|
||
| /** | ||
| * Parse the create PIT response body into a new {@link CreatePitResponse} object | ||
| */ | ||
| public static CreatePitResponse fromXContent(XContentParser parser) throws IOException { | ||
| ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser); | ||
| parser.nextToken(); | ||
| return innerFromXContent(parser); | ||
| } | ||
|
|
||
| public static CreatePitResponse innerFromXContent(XContentParser parser) throws IOException { | ||
| ensureExpectedToken(XContentParser.Token.FIELD_NAME, parser.currentToken(), parser); | ||
| String currentFieldName = parser.currentName(); | ||
| int successfulShards = -1; | ||
| int totalShards = -1; | ||
| int skippedShards = 0; | ||
| int failedShards = 0; | ||
| String id = null; | ||
| long creationTime = 0; | ||
| List<ShardSearchFailure> failures = new ArrayList<>(); | ||
| for (XContentParser.Token token = parser.nextToken(); token != XContentParser.Token.END_OBJECT; token = parser.nextToken()) { | ||
| if (token == XContentParser.Token.FIELD_NAME) { | ||
| currentFieldName = parser.currentName(); | ||
| } else if (token.isValue()) { | ||
| if (CREATION_TIME.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| creationTime = parser.longValue(); | ||
| } else if (ID.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| id = parser.text(); | ||
| } else { | ||
| parser.skipChildren(); | ||
| } | ||
| } else if (token == XContentParser.Token.START_OBJECT) { | ||
| if (RestActions._SHARDS_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) { | ||
| if (token == XContentParser.Token.FIELD_NAME) { | ||
| currentFieldName = parser.currentName(); | ||
| } else if (token.isValue()) { | ||
| if (RestActions.FAILED_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| failedShards = parser.intValue(); // we don't need it but need to consume it | ||
| } else if (RestActions.SUCCESSFUL_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| successfulShards = parser.intValue(); | ||
| } else if (RestActions.TOTAL_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| totalShards = parser.intValue(); | ||
| } else if (RestActions.SKIPPED_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| skippedShards = parser.intValue(); | ||
| } else { | ||
| parser.skipChildren(); | ||
| } | ||
| } else if (token == XContentParser.Token.START_ARRAY) { | ||
| if (RestActions.FAILURES_FIELD.match(currentFieldName, parser.getDeprecationHandler())) { | ||
| while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) { | ||
| failures.add(ShardSearchFailure.fromXContent(parser)); | ||
| } | ||
| } else { | ||
| parser.skipChildren(); | ||
| } | ||
| } else { | ||
| parser.skipChildren(); | ||
| } | ||
| } | ||
| } else { | ||
| parser.skipChildren(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return new CreatePitResponse( | ||
| id, | ||
| creationTime, | ||
| totalShards, | ||
| successfulShards, | ||
| skippedShards, | ||
| failedShards, | ||
| failures.toArray(ShardSearchFailure.EMPTY_ARRAY) | ||
| ); | ||
| } | ||
|
|
||
| public long getCreationTime() { | ||
| return creationTime; | ||
| } | ||
|
|
||
| /** | ||
| * The failed number of shards the search was executed on. | ||
| */ | ||
| public int getFailedShards() { | ||
| return shardFailures.length; | ||
| } | ||
|
|
||
| /** | ||
| * The failures that occurred during the search. | ||
| */ | ||
| public ShardSearchFailure[] getShardFailures() { | ||
| return this.shardFailures; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| out.writeString(id); | ||
| out.writeVInt(totalShards); | ||
| out.writeVInt(successfulShards); | ||
| out.writeVInt(failedShards); | ||
| out.writeVInt(skippedShards); | ||
| out.writeLong(creationTime); | ||
| out.writeVInt(shardFailures.length); | ||
| for (ShardSearchFailure shardSearchFailure : shardFailures) { | ||
| shardSearchFailure.writeTo(out); | ||
| } | ||
| } | ||
|
|
||
| public String getId() { | ||
| return id; | ||
| } | ||
|
|
||
| /** | ||
| * The total number of shards the create pit operation was executed on. | ||
| */ | ||
| public int getTotalShards() { | ||
| return totalShards; | ||
| } | ||
|
|
||
| /** | ||
| * The successful number of shards the create pit operation was executed on. | ||
| */ | ||
| public int getSuccessfulShards() { | ||
| return successfulShards; | ||
| } | ||
|
|
||
| public int getSkippedShards() { | ||
| return skippedShards; | ||
| } | ||
|
|
||
| @Override | ||
| public RestStatus status() { | ||
| return RestStatus.status(successfulShards, totalShards, shardFailures); | ||
| } | ||
| } |
This file contains 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,50 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.common.io.stream.Writeable; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| /** | ||
| * Pit ID along with Id for a search context per node. | ||
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public class PitSearchContextIdForNode implements Writeable { | ||
|
|
||
| private final String pitId; | ||
| private final SearchContextIdForNode searchContextIdForNode; | ||
|
|
||
| public PitSearchContextIdForNode(String pitId, SearchContextIdForNode searchContextIdForNode) { | ||
| this.pitId = pitId; | ||
| this.searchContextIdForNode = searchContextIdForNode; | ||
| } | ||
|
|
||
| PitSearchContextIdForNode(StreamInput in) throws IOException { | ||
| this.pitId = in.readString(); | ||
| this.searchContextIdForNode = new SearchContextIdForNode(in); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| out.writeString(pitId); | ||
| searchContextIdForNode.writeTo(out); | ||
| } | ||
|
|
||
| public String getPitId() { | ||
| return pitId; | ||
| } | ||
|
|
||
| public SearchContextIdForNode getSearchContextIdForNode() { | ||
| return searchContextIdForNode; | ||
| } | ||
| } |
This file contains 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 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 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 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,44 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.action.ActionListener; | ||
| import org.opensearch.action.StepListener; | ||
| import org.opensearch.cluster.ClusterState; | ||
| import org.opensearch.cluster.node.DiscoveryNode; | ||
| import org.opensearch.transport.RemoteClusterService; | ||
|
|
||
| import java.util.Set; | ||
| import java.util.function.BiFunction; | ||
|
|
||
| /** | ||
| * Helper class for common search functions | ||
| */ | ||
| public class SearchUtils { | ||
|
|
||
| public SearchUtils() {} | ||
|
|
||
| /** | ||
| * Get connection lookup listener for list of clusters passed | ||
| */ | ||
| public static ActionListener<BiFunction<String, String, DiscoveryNode>> getConnectionLookupListener( | ||
| RemoteClusterService remoteClusterService, | ||
| ClusterState state, | ||
| Set<String> clusters | ||
| ) { | ||
| final StepListener<BiFunction<String, String, DiscoveryNode>> lookupListener = new StepListener<>(); | ||
|
|
||
| if (clusters.isEmpty()) { | ||
| lookupListener.onResponse((cluster, nodeId) -> state.getNodes().get(nodeId)); | ||
| } else { | ||
| remoteClusterService.collectNodes(clusters, lookupListener); | ||
| } | ||
| return lookupListener; | ||
| } | ||
| } |
This file contains 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,133 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.opensearch.action.ActionListener; | ||
| import org.opensearch.action.StepListener; | ||
| import org.opensearch.action.support.ActionFilters; | ||
| import org.opensearch.action.support.HandledTransportAction; | ||
| import org.opensearch.cluster.service.ClusterService; | ||
| import org.opensearch.common.inject.Inject; | ||
| import org.opensearch.common.io.stream.NamedWriteableRegistry; | ||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.index.shard.ShardId; | ||
| import org.opensearch.search.SearchPhaseResult; | ||
| import org.opensearch.search.internal.ShardSearchContextId; | ||
| import org.opensearch.tasks.Task; | ||
| import org.opensearch.transport.TransportRequest; | ||
| import org.opensearch.transport.TransportService; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Arrays; | ||
|
|
||
| /** | ||
| * Transport action for creating PIT reader context | ||
| */ | ||
| public class TransportCreatePitAction extends HandledTransportAction<CreatePitRequest, CreatePitResponse> { | ||
|
|
||
| public static final String CREATE_PIT_ACTION = "create_pit"; | ||
| private final TransportService transportService; | ||
| private final SearchTransportService searchTransportService; | ||
| private final ClusterService clusterService; | ||
| private final TransportSearchAction transportSearchAction; | ||
| private final NamedWriteableRegistry namedWriteableRegistry; | ||
| private final CreatePitController createPitController; | ||
|
|
||
| @Inject | ||
| public TransportCreatePitAction( | ||
| TransportService transportService, | ||
| ActionFilters actionFilters, | ||
| SearchTransportService searchTransportService, | ||
| ClusterService clusterService, | ||
| TransportSearchAction transportSearchAction, | ||
| NamedWriteableRegistry namedWriteableRegistry, | ||
| CreatePitController createPitController | ||
| ) { | ||
| super(CreatePitAction.NAME, transportService, actionFilters, in -> new CreatePitRequest(in)); | ||
| this.transportService = transportService; | ||
| this.searchTransportService = searchTransportService; | ||
| this.clusterService = clusterService; | ||
| this.transportSearchAction = transportSearchAction; | ||
| this.namedWriteableRegistry = namedWriteableRegistry; | ||
| this.createPitController = createPitController; | ||
| } | ||
|
|
||
| @Override | ||
| protected void doExecute(Task task, CreatePitRequest request, ActionListener<CreatePitResponse> listener) { | ||
| final StepListener<SearchResponse> createPitListener = new StepListener<>(); | ||
| final ActionListener<CreatePitResponse> updatePitIdListener = ActionListener.wrap(r -> listener.onResponse(r), e -> { | ||
| logger.error( | ||
| () -> new ParameterizedMessage( | ||
| "PIT creation failed while updating PIT ID for indices [{}]", | ||
| Arrays.toString(request.indices()) | ||
| ) | ||
| ); | ||
| listener.onFailure(e); | ||
| }); | ||
| createPitController.executeCreatePit(request, task, createPitListener, updatePitIdListener); | ||
| } | ||
|
|
||
| /** | ||
| * Request to create pit reader context with keep alive | ||
| */ | ||
| public static class CreateReaderContextRequest extends TransportRequest { | ||
| private final ShardId shardId; | ||
| private final TimeValue keepAlive; | ||
|
|
||
| public CreateReaderContextRequest(ShardId shardId, TimeValue keepAlive) { | ||
| this.shardId = shardId; | ||
| this.keepAlive = keepAlive; | ||
| } | ||
|
|
||
| public ShardId getShardId() { | ||
| return shardId; | ||
| } | ||
|
|
||
| public TimeValue getKeepAlive() { | ||
| return keepAlive; | ||
| } | ||
|
|
||
| public CreateReaderContextRequest(StreamInput in) throws IOException { | ||
| super(in); | ||
| this.shardId = new ShardId(in); | ||
| this.keepAlive = in.readTimeValue(); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| shardId.writeTo(out); | ||
| out.writeTimeValue(keepAlive); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Create pit reader context response which holds the contextId | ||
| */ | ||
| public static class CreateReaderContextResponse extends SearchPhaseResult { | ||
| public CreateReaderContextResponse(ShardSearchContextId shardSearchContextId) { | ||
| this.contextId = shardSearchContextId; | ||
| } | ||
|
|
||
| public CreateReaderContextResponse(StreamInput in) throws IOException { | ||
| super(in); | ||
| contextId = new ShardSearchContextId(in); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| contextId.writeTo(out); | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains 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 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,67 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.search.internal.ShardSearchContextId; | ||
| import org.opensearch.transport.TransportRequest; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| /** | ||
| * Request used to update PIT reader contexts with pitId, keepAlive and creationTime | ||
| */ | ||
| public class UpdatePitContextRequest extends TransportRequest { | ||
| private final String pitId; | ||
| private final long keepAlive; | ||
|
|
||
| private final long creationTime; | ||
| private final ShardSearchContextId searchContextId; | ||
|
|
||
| public UpdatePitContextRequest(ShardSearchContextId searchContextId, String pitId, long keepAlive, long creationTime) { | ||
| this.pitId = pitId; | ||
| this.searchContextId = searchContextId; | ||
| this.keepAlive = keepAlive; | ||
| this.creationTime = creationTime; | ||
| } | ||
|
|
||
| UpdatePitContextRequest(StreamInput in) throws IOException { | ||
| super(in); | ||
| pitId = in.readString(); | ||
| keepAlive = in.readLong(); | ||
| creationTime = in.readLong(); | ||
| searchContextId = new ShardSearchContextId(in); | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| super.writeTo(out); | ||
| out.writeString(pitId); | ||
| out.writeLong(keepAlive); | ||
| out.writeLong(creationTime); | ||
| searchContextId.writeTo(out); | ||
| } | ||
|
|
||
| public ShardSearchContextId getSearchContextId() { | ||
| return searchContextId; | ||
| } | ||
|
|
||
| public String getPitId() { | ||
| return pitId; | ||
| } | ||
|
|
||
| public long getCreationTime() { | ||
| return creationTime; | ||
| } | ||
|
|
||
| public long getKeepAlive() { | ||
| return keepAlive; | ||
| } | ||
| } |
This file contains 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,58 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.common.io.stream.StreamInput; | ||
| import org.opensearch.common.io.stream.StreamOutput; | ||
| import org.opensearch.transport.TransportResponse; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| /** | ||
| * Update PIT context response with creation time, keep alive etc. | ||
| */ | ||
| public class UpdatePitContextResponse extends TransportResponse { | ||
| private final String pitId; | ||
|
|
||
| private final long creationTime; | ||
|
|
||
| private final long keepAlive; | ||
|
|
||
| UpdatePitContextResponse(StreamInput in) throws IOException { | ||
| super(in); | ||
| pitId = in.readString(); | ||
| creationTime = in.readLong(); | ||
| keepAlive = in.readLong(); | ||
| } | ||
|
|
||
| public UpdatePitContextResponse(String pitId, long creationTime, long keepAlive) { | ||
| this.pitId = pitId; | ||
| this.keepAlive = keepAlive; | ||
| this.creationTime = creationTime; | ||
| } | ||
|
|
||
| @Override | ||
| public void writeTo(StreamOutput out) throws IOException { | ||
| out.writeString(pitId); | ||
| out.writeLong(creationTime); | ||
| out.writeLong(keepAlive); | ||
| } | ||
|
|
||
| public String getPitId() { | ||
| return pitId; | ||
| } | ||
|
|
||
| public long getKeepAlive() { | ||
| return keepAlive; | ||
| } | ||
|
|
||
| public long getCreationTime() { | ||
| return creationTime; | ||
| } | ||
| } |
This file contains 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 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 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 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 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 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 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 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 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,70 @@ | ||
| /* | ||
| * 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.search.internal; | ||
|
|
||
| import org.apache.lucene.util.SetOnce; | ||
| import org.opensearch.common.lease.Releasable; | ||
| import org.opensearch.common.lease.Releasables; | ||
| import org.opensearch.index.IndexService; | ||
| import org.opensearch.index.engine.Engine; | ||
| import org.opensearch.index.shard.IndexShard; | ||
|
|
||
| /** | ||
| * PIT reader context containing PIT specific information such as pit id, create time etc. | ||
| */ | ||
| public class PitReaderContext extends ReaderContext { | ||
|
|
||
| // Storing the encoded PIT ID as part of PIT reader context for use cases such as list pit API | ||
| private final SetOnce<String> pitId = new SetOnce<>(); | ||
| // Creation time of PIT contexts which helps users to differentiate between multiple PIT reader contexts | ||
| private final SetOnce<Long> creationTime = new SetOnce<>(); | ||
|
|
||
| public PitReaderContext( | ||
| ShardSearchContextId id, | ||
| IndexService indexService, | ||
| IndexShard indexShard, | ||
| Engine.SearcherSupplier searcherSupplier, | ||
| long keepAliveInMillis, | ||
| boolean singleSession | ||
| ) { | ||
| super(id, indexService, indexShard, searcherSupplier, keepAliveInMillis, singleSession); | ||
| } | ||
|
|
||
| public String getPitId() { | ||
| return this.pitId.get(); | ||
| } | ||
|
|
||
| public void setPitId(final String pitId) { | ||
| this.pitId.set(pitId); | ||
| } | ||
|
|
||
| /** | ||
| * Returns a releasable to indicate that the caller has stopped using this reader. | ||
| * The pit id can be updated and time to live of the reader usage can be extended using the provided | ||
| * <code>keepAliveInMillis</code>. | ||
| */ | ||
| public Releasable updatePitIdAndKeepAlive(long keepAliveInMillis, String pitId, long createTime) { | ||
| getRefCounted().incRef(); | ||
| tryUpdateKeepAlive(keepAliveInMillis); | ||
| setPitId(pitId); | ||
| setCreationTime(createTime); | ||
| return Releasables.releaseOnce(() -> { | ||
| updateLastAccessTime(); | ||
| getRefCounted().decRef(); | ||
| }); | ||
| } | ||
|
|
||
| public long getCreationTime() { | ||
| return this.creationTime.get(); | ||
| } | ||
|
|
||
| public void setCreationTime(final long creationTime) { | ||
| this.creationTime.set(creationTime); | ||
| } | ||
| } |
This file contains 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 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,84 @@ | ||
| /* | ||
| * 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.action.search; | ||
|
|
||
| import org.opensearch.Version; | ||
| import org.opensearch.common.util.concurrent.AtomicArray; | ||
| import org.opensearch.index.query.IdsQueryBuilder; | ||
| import org.opensearch.index.query.MatchAllQueryBuilder; | ||
| import org.opensearch.index.query.QueryBuilder; | ||
| import org.opensearch.index.query.TermQueryBuilder; | ||
| import org.opensearch.index.shard.ShardId; | ||
| import org.opensearch.search.SearchPhaseResult; | ||
| import org.opensearch.search.SearchShardTarget; | ||
| import org.opensearch.search.internal.AliasFilter; | ||
| import org.opensearch.search.internal.ShardSearchContextId; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
|
|
||
| import static org.opensearch.test.OpenSearchTestCase.between; | ||
| import static org.opensearch.test.OpenSearchTestCase.randomAlphaOfLength; | ||
| import static org.opensearch.test.OpenSearchTestCase.randomBoolean; | ||
|
|
||
| /** | ||
| * Helper class for common pit tests functions | ||
| */ | ||
| public class PitTestsUtil { | ||
| private PitTestsUtil() {} | ||
|
|
||
| public static QueryBuilder randomQueryBuilder() { | ||
| if (randomBoolean()) { | ||
| return new TermQueryBuilder(randomAlphaOfLength(10), randomAlphaOfLength(10)); | ||
| } else if (randomBoolean()) { | ||
| return new MatchAllQueryBuilder(); | ||
| } else { | ||
| return new IdsQueryBuilder().addIds(randomAlphaOfLength(10)); | ||
| } | ||
| } | ||
|
|
||
| public static String getPitId() { | ||
| AtomicArray<SearchPhaseResult> array = new AtomicArray<>(3); | ||
| SearchAsyncActionTests.TestSearchPhaseResult testSearchPhaseResult1 = new SearchAsyncActionTests.TestSearchPhaseResult( | ||
| new ShardSearchContextId("a", 1), | ||
| null | ||
| ); | ||
| testSearchPhaseResult1.setSearchShardTarget(new SearchShardTarget("node_1", new ShardId("idx", "uuid1", 2), null, null)); | ||
| SearchAsyncActionTests.TestSearchPhaseResult testSearchPhaseResult2 = new SearchAsyncActionTests.TestSearchPhaseResult( | ||
| new ShardSearchContextId("b", 12), | ||
| null | ||
| ); | ||
| testSearchPhaseResult2.setSearchShardTarget(new SearchShardTarget("node_2", new ShardId("idy", "uuid2", 42), null, null)); | ||
| SearchAsyncActionTests.TestSearchPhaseResult testSearchPhaseResult3 = new SearchAsyncActionTests.TestSearchPhaseResult( | ||
| new ShardSearchContextId("c", 42), | ||
| null | ||
| ); | ||
| testSearchPhaseResult3.setSearchShardTarget(new SearchShardTarget("node_3", new ShardId("idy", "uuid2", 43), null, null)); | ||
| array.setOnce(0, testSearchPhaseResult1); | ||
| array.setOnce(1, testSearchPhaseResult2); | ||
| array.setOnce(2, testSearchPhaseResult3); | ||
|
|
||
| final Version version = Version.CURRENT; | ||
| final Map<String, AliasFilter> aliasFilters = new HashMap<>(); | ||
| for (SearchPhaseResult result : array.asList()) { | ||
| final AliasFilter aliasFilter; | ||
| if (randomBoolean()) { | ||
| aliasFilter = new AliasFilter(randomQueryBuilder()); | ||
| } else if (randomBoolean()) { | ||
| aliasFilter = new AliasFilter(randomQueryBuilder(), "alias-" + between(1, 10)); | ||
| } else { | ||
| aliasFilter = AliasFilter.EMPTY; | ||
| } | ||
| if (randomBoolean()) { | ||
| aliasFilters.put(result.getSearchShardTarget().getShardId().getIndex().getUUID(), aliasFilter); | ||
| } | ||
| } | ||
| return SearchContextId.encode(array.asList(), aliasFilters, version); | ||
| } | ||
| } |
This file contains 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,202 @@ | ||
| /* | ||
| * 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.search; | ||
|
|
||
| import org.junit.After; | ||
| import org.junit.Before; | ||
| import org.opensearch.action.ActionFuture; | ||
| import org.opensearch.action.search.CreatePitAction; | ||
| import org.opensearch.action.search.CreatePitRequest; | ||
| import org.opensearch.action.search.CreatePitResponse; | ||
| import org.opensearch.action.search.SearchResponse; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.search.builder.PointInTimeBuilder; | ||
| import org.opensearch.test.InternalTestCluster; | ||
| import org.opensearch.test.OpenSearchIntegTestCase; | ||
|
|
||
| import java.util.concurrent.ExecutionException; | ||
|
|
||
| import static org.hamcrest.Matchers.containsString; | ||
| import static org.opensearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE; | ||
| import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; | ||
|
|
||
| /** | ||
| * Multi node integration tests for PIT creation and search operation with PIT ID. | ||
| */ | ||
| @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.SUITE, numDataNodes = 2) | ||
| public class CreatePitMultiNodeTests extends OpenSearchIntegTestCase { | ||
|
|
||
| @Before | ||
| public void setupIndex() throws ExecutionException, InterruptedException { | ||
| createIndex("index", Settings.builder().put("index.number_of_shards", 2).put("index.number_of_replicas", 0).build()); | ||
| client().prepareIndex("index").setId("1").setSource("field", "value").setRefreshPolicy(IMMEDIATE).execute().get(); | ||
| ensureGreen(); | ||
| } | ||
|
|
||
| @After | ||
| public void clearIndex() { | ||
| client().admin().indices().prepareDelete("index").get(); | ||
| } | ||
|
|
||
| public void testPit() throws Exception { | ||
| CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true); | ||
| request.setIndices(new String[] { "index" }); | ||
| ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request); | ||
| CreatePitResponse pitResponse = execute.get(); | ||
| SearchResponse searchResponse = client().prepareSearch("index") | ||
| .setSize(2) | ||
| .setPointInTime(new PointInTimeBuilder(pitResponse.getId()).setKeepAlive(TimeValue.timeValueDays(1))) | ||
| .get(); | ||
| assertEquals(2, searchResponse.getSuccessfulShards()); | ||
| assertEquals(2, searchResponse.getTotalShards()); | ||
| } | ||
|
|
||
| public void testCreatePitWhileNodeDropWithAllowPartialCreationFalse() throws Exception { | ||
| CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), false); | ||
| request.setIndices(new String[] { "index" }); | ||
| internalCluster().restartRandomDataNode(new InternalTestCluster.RestartCallback() { | ||
| @Override | ||
| public Settings onNodeStopped(String nodeName) throws Exception { | ||
| ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request); | ||
| ExecutionException ex = expectThrows(ExecutionException.class, execute::get); | ||
| assertTrue(ex.getMessage().contains("Failed to execute phase [create_pit]")); | ||
| assertTrue(ex.getMessage().contains("Partial shards failure")); | ||
| return super.onNodeStopped(nodeName); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public void testCreatePitWhileNodeDropWithAllowPartialCreationTrue() throws Exception { | ||
| CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true); | ||
| request.setIndices(new String[] { "index" }); | ||
| internalCluster().restartRandomDataNode(new InternalTestCluster.RestartCallback() { | ||
| @Override | ||
| public Settings onNodeStopped(String nodeName) throws Exception { | ||
| ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request); | ||
| CreatePitResponse pitResponse = execute.get(); | ||
| assertEquals(1, pitResponse.getSuccessfulShards()); | ||
| assertEquals(2, pitResponse.getTotalShards()); | ||
| SearchResponse searchResponse = client().prepareSearch("index") | ||
| .setSize(2) | ||
| .setPointInTime(new PointInTimeBuilder(pitResponse.getId()).setKeepAlive(TimeValue.timeValueDays(1))) | ||
| .get(); | ||
| assertEquals(1, searchResponse.getSuccessfulShards()); | ||
| assertEquals(1, searchResponse.getTotalShards()); | ||
| return super.onNodeStopped(nodeName); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public void testPitSearchWithNodeDrop() throws Exception { | ||
| CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true); | ||
| request.setIndices(new String[] { "index" }); | ||
| ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request); | ||
| CreatePitResponse pitResponse = execute.get(); | ||
| internalCluster().restartRandomDataNode(new InternalTestCluster.RestartCallback() { | ||
| @Override | ||
| public Settings onNodeStopped(String nodeName) throws Exception { | ||
| SearchResponse searchResponse = client().prepareSearch() | ||
| .setSize(2) | ||
| .setPointInTime(new PointInTimeBuilder(pitResponse.getId()).setKeepAlive(TimeValue.timeValueDays(1))) | ||
| .get(); | ||
| assertEquals(1, searchResponse.getSuccessfulShards()); | ||
| assertEquals(1, searchResponse.getFailedShards()); | ||
| assertEquals(0, searchResponse.getSkippedShards()); | ||
| assertEquals(2, searchResponse.getTotalShards()); | ||
| return super.onNodeStopped(nodeName); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public void testPitSearchWithNodeDropWithPartialSearchResultsFalse() throws Exception { | ||
| CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true); | ||
| request.setIndices(new String[] { "index" }); | ||
| ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request); | ||
| CreatePitResponse pitResponse = execute.get(); | ||
| internalCluster().restartRandomDataNode(new InternalTestCluster.RestartCallback() { | ||
| @Override | ||
| public Settings onNodeStopped(String nodeName) throws Exception { | ||
| ActionFuture<SearchResponse> execute = client().prepareSearch() | ||
| .setSize(2) | ||
| .setPointInTime(new PointInTimeBuilder(pitResponse.getId()).setKeepAlive(TimeValue.timeValueDays(1))) | ||
| .setAllowPartialSearchResults(false) | ||
| .execute(); | ||
| ExecutionException ex = expectThrows(ExecutionException.class, execute::get); | ||
| assertTrue(ex.getMessage().contains("Partial shards failure")); | ||
| return super.onNodeStopped(nodeName); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public void testPitInvalidDefaultKeepAlive() { | ||
| IllegalArgumentException exc = expectThrows( | ||
| IllegalArgumentException.class, | ||
| () -> client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("point_in_time.max_keep_alive", "1m").put("search.default_keep_alive", "2m")) | ||
| .get() | ||
| ); | ||
| assertThat(exc.getMessage(), containsString("was (2m > 1m)")); | ||
| assertAcked( | ||
| client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("search.default_keep_alive", "5m").put("point_in_time.max_keep_alive", "5m")) | ||
| .get() | ||
| ); | ||
| assertAcked( | ||
| client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("search.default_keep_alive", "2m")) | ||
| .get() | ||
| ); | ||
| assertAcked( | ||
| client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("point_in_time.max_keep_alive", "2m")) | ||
| .get() | ||
| ); | ||
| exc = expectThrows( | ||
| IllegalArgumentException.class, | ||
| () -> client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("search.default_keep_alive", "3m")) | ||
| .get() | ||
| ); | ||
| assertThat(exc.getMessage(), containsString("was (3m > 2m)")); | ||
| assertAcked( | ||
| client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("search.default_keep_alive", "1m")) | ||
| .get() | ||
| ); | ||
| exc = expectThrows( | ||
| IllegalArgumentException.class, | ||
| () -> client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().put("point_in_time.max_keep_alive", "30s")) | ||
| .get() | ||
| ); | ||
| assertThat(exc.getMessage(), containsString("was (1m > 30s)")); | ||
| assertAcked( | ||
| client().admin() | ||
| .cluster() | ||
| .prepareUpdateSettings() | ||
| .setPersistentSettings(Settings.builder().putNull("*")) | ||
| .setTransientSettings(Settings.builder().putNull("*")) | ||
| ); | ||
| } | ||
| } |
This file contains 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 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
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.
@bharath-techie from the API perspective, this is beaking change. Since we backport to 2.x (no breaking changes), we probably should mark this methods as
default.