-
Notifications
You must be signed in to change notification settings - Fork 24.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Add CcrRestoreSourceService to track sessions #36578
Changes from 12 commits
b1ee4fd
2479f94
9fac474
7b79a5e
fe3ef84
e85bfa1
71db9aa
8d1a151
500350c
5f9c1f3
17a5f24
be2e6e8
e275823
6e4a59d
592bdc0
d0798ca
8a6ae8d
b781cb0
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,121 @@ | ||
/* | ||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
* or more contributor license agreements. Licensed under the Elastic License; | ||
* you may not use this file except in compliance with the Elastic License. | ||
*/ | ||
|
||
package org.elasticsearch.xpack.ccr.action.repositories; | ||
|
||
import org.elasticsearch.action.Action; | ||
import org.elasticsearch.action.FailedNodeException; | ||
import org.elasticsearch.action.support.ActionFilters; | ||
import org.elasticsearch.action.support.nodes.BaseNodeResponse; | ||
import org.elasticsearch.action.support.nodes.BaseNodesResponse; | ||
import org.elasticsearch.action.support.nodes.TransportNodesAction; | ||
import org.elasticsearch.cluster.ClusterName; | ||
import org.elasticsearch.cluster.node.DiscoveryNode; | ||
import org.elasticsearch.cluster.service.ClusterService; | ||
import org.elasticsearch.common.inject.Inject; | ||
import org.elasticsearch.common.io.stream.StreamInput; | ||
import org.elasticsearch.common.io.stream.StreamOutput; | ||
import org.elasticsearch.threadpool.ThreadPool; | ||
import org.elasticsearch.transport.TransportService; | ||
import org.elasticsearch.xpack.ccr.repository.CcrRestoreSourceService; | ||
|
||
import java.io.IOException; | ||
import java.util.List; | ||
|
||
public class ClearCcrRestoreSessionAction extends Action<ClearCcrRestoreSessionAction.ClearCcrRestoreSessionResponse> { | ||
|
||
public static final ClearCcrRestoreSessionAction INSTANCE = new ClearCcrRestoreSessionAction(); | ||
private static final String NAME = "internal:admin/ccr/restore/session/clear"; | ||
|
||
private ClearCcrRestoreSessionAction() { | ||
super(NAME); | ||
} | ||
|
||
@Override | ||
public ClearCcrRestoreSessionResponse newResponse() { | ||
return new ClearCcrRestoreSessionResponse(); | ||
} | ||
|
||
public static class TransportDeleteCcrRestoreSessionAction extends TransportNodesAction<ClearCcrRestoreSessionRequest, | ||
ClearCcrRestoreSessionResponse, ClearCcrRestoreSessionRequest.Request, Response> { | ||
|
||
private final CcrRestoreSourceService ccrRestoreService; | ||
|
||
@Inject | ||
public TransportDeleteCcrRestoreSessionAction(ThreadPool threadPool, ClusterService clusterService, ActionFilters actionFilters, | ||
TransportService transportService, CcrRestoreSourceService ccrRestoreService) { | ||
super(NAME, threadPool, clusterService, transportService, actionFilters, ClearCcrRestoreSessionRequest::new, | ||
ClearCcrRestoreSessionRequest.Request::new, ThreadPool.Names.GENERIC, Response.class); | ||
this.ccrRestoreService = ccrRestoreService; | ||
} | ||
|
||
@Override | ||
protected ClearCcrRestoreSessionResponse newResponse(ClearCcrRestoreSessionRequest request, List<Response> responses, | ||
List<FailedNodeException> failures) { | ||
return new ClearCcrRestoreSessionResponse(clusterService.getClusterName(), responses, failures); | ||
} | ||
|
||
@Override | ||
protected ClearCcrRestoreSessionRequest.Request newNodeRequest(String nodeId, ClearCcrRestoreSessionRequest request) { | ||
return request.getRequest(); | ||
} | ||
|
||
@Override | ||
protected Response newNodeResponse() { | ||
return new Response(); | ||
} | ||
|
||
@Override | ||
protected Response nodeOperation(ClearCcrRestoreSessionRequest.Request request) { | ||
ccrRestoreService.closeSession(request.getSessionUUID()); | ||
return new Response(clusterService.localNode()); | ||
} | ||
} | ||
|
||
public static class Response extends BaseNodeResponse { | ||
|
||
private Response() { | ||
} | ||
|
||
private Response(StreamInput streamInput) throws IOException { | ||
readFrom(streamInput); | ||
} | ||
|
||
private Response(DiscoveryNode node) { | ||
super(node); | ||
} | ||
|
||
@Override | ||
public void writeTo(StreamOutput out) throws IOException { | ||
super.writeTo(out); | ||
} | ||
|
||
@Override | ||
public void readFrom(StreamInput in) throws IOException { | ||
super.readFrom(in); | ||
} | ||
} | ||
|
||
public static class ClearCcrRestoreSessionResponse extends BaseNodesResponse<Response> { | ||
|
||
ClearCcrRestoreSessionResponse() { | ||
} | ||
|
||
ClearCcrRestoreSessionResponse(ClusterName clusterName, List<Response> chunkResponses, List<FailedNodeException> failures) { | ||
super(clusterName, chunkResponses, failures); | ||
} | ||
|
||
@Override | ||
protected List<Response> readNodesFrom(StreamInput in) throws IOException { | ||
return in.readList(Response::new); | ||
} | ||
|
||
@Override | ||
protected void writeNodesTo(StreamOutput out, List<Response> nodes) throws IOException { | ||
out.writeList(nodes); | ||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,73 @@ | ||
/* | ||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
* or more contributor license agreements. Licensed under the Elastic License; | ||
* you may not use this file except in compliance with the Elastic License. | ||
*/ | ||
|
||
package org.elasticsearch.xpack.ccr.action.repositories; | ||
|
||
import org.elasticsearch.action.support.nodes.BaseNodeRequest; | ||
import org.elasticsearch.action.support.nodes.BaseNodesRequest; | ||
import org.elasticsearch.common.io.stream.StreamInput; | ||
import org.elasticsearch.common.io.stream.StreamOutput; | ||
|
||
import java.io.IOException; | ||
|
||
public class ClearCcrRestoreSessionRequest extends BaseNodesRequest<ClearCcrRestoreSessionRequest> { | ||
|
||
private Request request; | ||
|
||
ClearCcrRestoreSessionRequest() { | ||
} | ||
|
||
public ClearCcrRestoreSessionRequest(String nodeId, Request request) { | ||
super(nodeId); | ||
this.request = request; | ||
} | ||
|
||
@Override | ||
public void readFrom(StreamInput streamInput) throws IOException { | ||
super.readFrom(streamInput); | ||
request = new Request(); | ||
request.readFrom(streamInput); | ||
} | ||
|
||
@Override | ||
public void writeTo(StreamOutput streamOutput) throws IOException { | ||
super.writeTo(streamOutput); | ||
request.writeTo(streamOutput); | ||
} | ||
|
||
public Request getRequest() { | ||
return request; | ||
} | ||
|
||
public static class Request extends BaseNodeRequest { | ||
|
||
private String sessionUUID; | ||
|
||
Request() { | ||
} | ||
|
||
public Request(String nodeId, String sessionUUID) { | ||
super(nodeId); | ||
this.sessionUUID = sessionUUID; | ||
} | ||
|
||
@Override | ||
public void readFrom(StreamInput streamInput) throws IOException { | ||
super.readFrom(streamInput); | ||
sessionUUID = streamInput.readString(); | ||
} | ||
|
||
@Override | ||
public void writeTo(StreamOutput streamOutput) throws IOException { | ||
super.writeTo(streamOutput); | ||
streamOutput.writeString(sessionUUID); | ||
} | ||
|
||
public String getSessionUUID() { | ||
return sessionUUID; | ||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,126 @@ | ||
/* | ||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
* or more contributor license agreements. Licensed under the Elastic License; | ||
* you may not use this file except in compliance with the Elastic License. | ||
*/ | ||
|
||
package org.elasticsearch.xpack.ccr.action.repositories; | ||
|
||
import org.elasticsearch.action.Action; | ||
import org.elasticsearch.action.ActionResponse; | ||
import org.elasticsearch.action.support.ActionFilters; | ||
import org.elasticsearch.action.support.single.shard.TransportSingleShardAction; | ||
import org.elasticsearch.cluster.ClusterState; | ||
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver; | ||
import org.elasticsearch.cluster.routing.ShardsIterator; | ||
import org.elasticsearch.cluster.service.ClusterService; | ||
import org.elasticsearch.common.inject.Inject; | ||
import org.elasticsearch.common.io.stream.StreamInput; | ||
import org.elasticsearch.common.io.stream.StreamOutput; | ||
import org.elasticsearch.common.io.stream.Writeable; | ||
import org.elasticsearch.index.shard.IndexShard; | ||
import org.elasticsearch.index.shard.ShardId; | ||
import org.elasticsearch.index.shard.ShardNotFoundException; | ||
import org.elasticsearch.indices.IndicesService; | ||
import org.elasticsearch.threadpool.ThreadPool; | ||
import org.elasticsearch.transport.TransportService; | ||
import org.elasticsearch.xpack.ccr.repository.CcrRestoreSourceService; | ||
|
||
import java.io.IOException; | ||
|
||
public class PutCcrRestoreSessionAction extends Action<PutCcrRestoreSessionAction.PutCcrRestoreSessionResponse> { | ||
|
||
public static final PutCcrRestoreSessionAction INSTANCE = new PutCcrRestoreSessionAction(); | ||
private static final String NAME = "internal:admin/ccr/restore/session/put"; | ||
|
||
private PutCcrRestoreSessionAction() { | ||
super(NAME); | ||
} | ||
|
||
@Override | ||
public PutCcrRestoreSessionResponse newResponse() { | ||
return new PutCcrRestoreSessionResponse(); | ||
} | ||
|
||
@Override | ||
public Writeable.Reader<PutCcrRestoreSessionAction.PutCcrRestoreSessionResponse> getResponseReader() { | ||
return PutCcrRestoreSessionAction.PutCcrRestoreSessionResponse::new; | ||
} | ||
|
||
public static class TransportPutCcrRestoreSessionAction | ||
extends TransportSingleShardAction<PutCcrRestoreSessionRequest, PutCcrRestoreSessionResponse> { | ||
|
||
private final IndicesService indicesService; | ||
private final CcrRestoreSourceService ccrRestoreService; | ||
|
||
@Inject | ||
public TransportPutCcrRestoreSessionAction(ThreadPool threadPool, ClusterService clusterService, ActionFilters actionFilters, | ||
IndexNameExpressionResolver resolver, TransportService transportService, | ||
IndicesService indicesService, CcrRestoreSourceService ccrRestoreService) { | ||
super(NAME, threadPool, clusterService, transportService, actionFilters, resolver, PutCcrRestoreSessionRequest::new, | ||
ThreadPool.Names.GENERIC); | ||
this.indicesService = indicesService; | ||
this.ccrRestoreService = ccrRestoreService; | ||
} | ||
|
||
@Override | ||
protected PutCcrRestoreSessionResponse shardOperation(PutCcrRestoreSessionRequest request, ShardId shardId) throws IOException { | ||
IndexShard indexShard = indicesService.getShardOrNull(shardId); | ||
if (indexShard == null) { | ||
throw new ShardNotFoundException(shardId); | ||
} | ||
ccrRestoreService.openSession(request.getSessionUUID(), indexShard); | ||
return new PutCcrRestoreSessionResponse(indexShard.routingEntry().currentNodeId()); | ||
} | ||
|
||
@Override | ||
protected PutCcrRestoreSessionResponse newResponse() { | ||
return new PutCcrRestoreSessionResponse(); | ||
} | ||
|
||
@Override | ||
protected boolean resolveIndex(PutCcrRestoreSessionRequest request) { | ||
return false; | ||
} | ||
|
||
@Override | ||
protected ShardsIterator shards(ClusterState state, InternalRequest request) { | ||
final ShardId shardId = request.request().getShardId(); | ||
return state.routingTable().shardRoutingTable(shardId).primaryShardIt(); | ||
} | ||
} | ||
|
||
|
||
public static class PutCcrRestoreSessionResponse extends ActionResponse { | ||
|
||
private String nodeId; | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. can this be made final? I see that you both implemented a constructor with StreamInput and the readFrom method? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I don't think so. Unfortunately you must implement this:
on |
||
|
||
PutCcrRestoreSessionResponse() { | ||
} | ||
|
||
PutCcrRestoreSessionResponse(String nodeId) { | ||
this.nodeId = nodeId; | ||
} | ||
|
||
PutCcrRestoreSessionResponse(StreamInput streamInput) throws IOException { | ||
super(streamInput); | ||
nodeId = streamInput.readString(); | ||
} | ||
|
||
@Override | ||
public void readFrom(StreamInput streamInput) throws IOException { | ||
super.readFrom(streamInput); | ||
nodeId = streamInput.readString(); | ||
} | ||
|
||
@Override | ||
public void writeTo(StreamOutput streamOutput) throws IOException { | ||
super.writeTo(streamOutput); | ||
streamOutput.writeString(nodeId); | ||
} | ||
|
||
public String getNodeId() { | ||
return nodeId; | ||
} | ||
} | ||
} |
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.
TransportNodesAction is only truly useful if you intend on sending something to multiple nodes. I think it might be simpler here to directly use HandledTransportAction?
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.
I think I would like to implement a specific node request for the file chunks and delete session in a follow-up? I added a meta task.