changes as per new security model

Signed-off-by: Bharathwaj G <[email protected]>

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -238,16 +238,14 @@
 import org.opensearch.action.search.ClearScrollAction;
 import org.opensearch.action.search.CreatePitAction;
 import org.opensearch.action.search.DeletePitAction;
-import org.opensearch.action.search.GetAllPitsAction;
 import org.opensearch.action.search.MultiSearchAction;
-import org.opensearch.action.search.NodesGetAllPitsAction;
+import org.opensearch.action.search.GetAllPitsAction;
 import org.opensearch.action.search.SearchAction;
 import org.opensearch.action.search.SearchScrollAction;
 import org.opensearch.action.search.TransportClearScrollAction;
 import org.opensearch.action.search.TransportCreatePitAction;
 import org.opensearch.action.search.TransportDeletePitAction;
 import org.opensearch.action.search.TransportGetAllPitsAction;
-import org.opensearch.action.search.TransportNodesGetAllPitsAction;
 import org.opensearch.action.search.TransportMultiSearchAction;
 import org.opensearch.action.search.TransportSearchAction;
 import org.opensearch.action.search.TransportSearchScrollAction;
@@ -677,10 +675,9 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
 
         // point in time actions
         actions.register(CreatePitAction.INSTANCE, TransportCreatePitAction.class);
-        actions.register(GetAllPitsAction.INSTANCE, TransportGetAllPitsAction.class);
         actions.register(DeletePitAction.INSTANCE, TransportDeletePitAction.class);
         actions.register(PitSegmentsAction.INSTANCE, TransportPitSegmentsAction.class);
-        actions.register(NodesGetAllPitsAction.INSTANCE, TransportNodesGetAllPitsAction.class);
+        actions.register(GetAllPitsAction.INSTANCE, TransportGetAllPitsAction.class);
 
         // Remote Store
         actions.register(RestoreRemoteStoreAction.INSTANCE, TransportRestoreRemoteStoreAction.class);
@@ -859,7 +856,7 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
 
         // Point in time API
         registerHandler.accept(new RestCreatePitAction());
-        registerHandler.accept(new RestDeletePitAction(nodesInCluster));
+        registerHandler.accept(new RestDeletePitAction());
         registerHandler.accept(new RestGetAllPitsAction(nodesInCluster));
         registerHandler.accept(new RestPitSegmentsAction(nodesInCluster));
 

server/src/main/java/org/opensearch/action/admin/indices/segments/TransportPitSegmentsAction.java

@@ -8,6 +8,7 @@
 package org.opensearch.action.admin.indices.segments;
 
 import org.opensearch.action.ActionListener;
+import org.opensearch.action.search.ListPitInfo;
 import org.opensearch.action.search.PitService;
 import org.opensearch.action.search.SearchContextId;
 import org.opensearch.action.search.SearchContextIdForNode;
@@ -45,6 +46,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import static org.opensearch.action.search.SearchContextId.decode;
 
@@ -93,10 +95,11 @@ public TransportPitSegmentsAction(
     @Override
     protected void doExecute(Task task, PitSegmentsRequest request, ActionListener<IndicesSegmentResponse> listener) {
         List<String> pitIds = request.getPitIds();
-        // when security plugin intercepts the request, if PITs are not present in the cluster the PIT IDs in request will be empty
-        // and in this case return empty response
-        if (pitIds.isEmpty()) {
-            listener.onResponse(new IndicesSegmentResponse(new ShardSegments[] {}, 0, 0, 0, new ArrayList<>()));
+        if (pitIds.size() == 1 && "_all".equals(pitIds.get(0))) {
+            pitService.getAllPits(ActionListener.wrap(response -> {
+                request.clearAndSetPitIds(response.getPitInfos().stream().map(ListPitInfo::getPitId).collect(Collectors.toList()));
+                super.doExecute(task, request, listener);
+            }, listener::onFailure));
         } else {
             super.doExecute(task, request, listener);
         }

server/src/main/java/org/opensearch/action/search/GetAllPitNodesRequest.java

@@ -21,22 +21,11 @@
  */
 public class GetAllPitNodesRequest extends BaseNodesRequest<GetAllPitNodesRequest> {
 
-    // Security plugin intercepts and sets the response with permitted PIT contexts
-    private GetAllPitNodesResponse getAllPitNodesResponse;
-
     @Inject
     public GetAllPitNodesRequest(DiscoveryNode... concreteNodes) {
         super(concreteNodes);
     }
 
-    public void setGetAllPitNodesResponse(GetAllPitNodesResponse getAllPitNodesResponse) {
-        this.getAllPitNodesResponse = getAllPitNodesResponse;
-    }
-
-    public GetAllPitNodesResponse getGetAllPitNodesResponse() {
-        return getAllPitNodesResponse;
-    }
-
     public GetAllPitNodesRequest(StreamInput in) throws IOException {
         super(in);
     }

server/src/main/java/org/opensearch/action/search/GetAllPitNodesResponse.java

@@ -68,10 +68,10 @@ public GetAllPitNodesResponse(List<ListPitInfo> listPitInfos, GetAllPitNodesResp
     public GetAllPitNodesResponse(
         List<ListPitInfo> listPitInfos,
         ClusterName clusterName,
-        List<GetAllPitNodeResponse> getAllPitNodeResponse,
+        List<GetAllPitNodeResponse> getAllPitNodeResponseList,
         List<FailedNodeException> failures
     ) {
-        super(clusterName, getAllPitNodeResponse, failures);
+        super(clusterName, getAllPitNodeResponseList, failures);
         pitInfos.addAll(listPitInfos);
     }
 

server/src/main/java/org/opensearch/action/search/GetAllPitsAction.java

@@ -11,7 +11,7 @@
 import org.opensearch.action.ActionType;
 
 /**
- * Action type for listing all PIT reader contexts
+ * Action type for retrieving all PIT reader contexts from nodes
  */
 public class GetAllPitsAction extends ActionType<GetAllPitNodesResponse> {
     public static final GetAllPitsAction INSTANCE = new GetAllPitsAction();

server/src/main/java/org/opensearch/action/search/PitService.java

@@ -170,16 +170,14 @@ public Map<String, String[]> getIndicesForPits(List<String> pitIds) {
     /**
      * Get all active point in time contexts
      */
-    public void getAllPits(GetAllPitNodesResponse getAllPitNodesResponse, ActionListener<GetAllPitNodesResponse> getAllPitsListener) {
+    public void getAllPits(ActionListener<GetAllPitNodesResponse> getAllPitsListener) {
         final List<DiscoveryNode> nodes = new ArrayList<>();
         for (ObjectCursor<DiscoveryNode> cursor : clusterService.state().nodes().getDataNodes().values()) {
             DiscoveryNode node = cursor.value;
             nodes.add(node);
         }
-        logger.debug("Number of active PITs in cluster: " + getAllPitNodesResponse.getPitInfos().size());
         DiscoveryNode[] disNodesArr = nodes.toArray(new DiscoveryNode[nodes.size()]);
         GetAllPitNodesRequest getAllPitNodesRequest = new GetAllPitNodesRequest(disNodesArr);
-        getAllPitNodesRequest.setGetAllPitNodesResponse(getAllPitNodesResponse);
         transportService.sendRequest(
             transportService.getLocalNode(),
             GetAllPitsAction.NAME,

server/src/main/java/org/opensearch/action/search/TransportDeletePitAction.java

@@ -11,7 +11,6 @@
 import org.opensearch.action.ActionListener;
 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.tasks.Task;
@@ -21,6 +20,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Transport action for deleting point in time searches - supports deleting list and all point in time searches
@@ -34,9 +34,6 @@ public TransportDeletePitAction(
         TransportService transportService,
         ActionFilters actionFilters,
         NamedWriteableRegistry namedWriteableRegistry,
-        TransportSearchAction transportSearchAction,
-        ClusterService clusterService,
-        SearchTransportService searchTransportService,
         PitService pitService
     ) {
         super(DeletePitAction.NAME, transportService, actionFilters, DeletePitRequest::new);
@@ -50,10 +47,8 @@ public TransportDeletePitAction(
     @Override
     protected void doExecute(Task task, DeletePitRequest request, ActionListener<DeletePitResponse> listener) {
         List<String> pitIds = request.getPitIds();
-        // when security plugin intercepts the request, if PITs are not present in the cluster the PIT IDs in request will be empty
-        // and in this case return empty response
-        if (pitIds.isEmpty()) {
-            listener.onResponse(new DeletePitResponse(new ArrayList<>()));
+        if (pitIds.size() == 1 && "_all".equals(pitIds.get(0))) {
+            deleteAllPits(listener);
         } else {
             deletePits(listener, request);
         }
@@ -75,4 +70,24 @@ private void deletePits(ActionListener<DeletePitResponse> listener, DeletePitReq
         }
         pitService.deletePitContexts(nodeToContextsMap, listener);
     }
+
+    /**
+     * Delete all active PIT reader contexts leveraging list all PITs
+     *
+     * For Cross cluster PITs :
+     * - mixed cluster PITs ( PIT comprising local and remote ) will be fully deleted. Since there will atleast be
+     * one reader context with PIT ID present in local cluster, 'Get all PITs' will retrieve the PIT ID with which
+     * we can completely delete the PIT contexts in both local and remote cluster.
+     * - fully remote PITs will not be deleted as 'Get all PITs' operates on local cluster only and no PIT info can
+     * be retrieved when it's fully remote.
+     */
+    private void deleteAllPits(ActionListener<DeletePitResponse> listener) {
+        // Get all PITs and execute delete operation for the PITs.
+        pitService.getAllPits(ActionListener.wrap(getAllPitNodesResponse -> {
+            DeletePitRequest deletePitRequest = new DeletePitRequest(
+                getAllPitNodesResponse.getPitInfos().stream().map(r -> r.getPitId()).collect(Collectors.toList())
+            );
+            deletePits(listener, deletePitRequest);
+        }, listener::onFailure));
+    }
 }

server/src/main/java/org/opensearch/action/search/TransportGetAllPitsAction.java

@@ -8,29 +8,79 @@
 
 package org.opensearch.action.search;
 
-import org.opensearch.action.ActionListener;
+import org.opensearch.action.FailedNodeException;
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.action.support.HandledTransportAction;
+import org.opensearch.action.support.nodes.TransportNodesAction;
+import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
-import org.opensearch.tasks.Task;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.search.SearchService;
+import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
+import java.io.IOException;
+import java.util.List;
+
 /**
- * Transport action to get all active PIT contexts across the cluster
+ * Transport action to get all active PIT contexts across all nodes
  */
-public class TransportGetAllPitsAction extends HandledTransportAction<GetAllPitNodesRequest, GetAllPitNodesResponse> {
-    private final PitService pitService;
+public class TransportGetAllPitsAction extends TransportNodesAction<
+    GetAllPitNodesRequest,
+    GetAllPitNodesResponse,
+    GetAllPitNodeRequest,
+    GetAllPitNodeResponse> {
+    private final SearchService searchService;
 
     @Inject
-    public TransportGetAllPitsAction(ActionFilters actionFilters, TransportService transportService, PitService pitService) {
-        super(GetAllPitsAction.NAME, transportService, actionFilters, in -> new GetAllPitNodesRequest(in));
-        this.pitService = pitService;
+    public TransportGetAllPitsAction(
+        ThreadPool threadPool,
+        ClusterService clusterService,
+        TransportService transportService,
+        ActionFilters actionFilters,
+        SearchService searchService
+    ) {
+        super(
+            GetAllPitsAction.NAME,
+            threadPool,
+            clusterService,
+            transportService,
+            actionFilters,
+            GetAllPitNodesRequest::new,
+            GetAllPitNodeRequest::new,
+            ThreadPool.Names.SAME,
+            GetAllPitNodeResponse.class
+        );
+        this.searchService = searchService;
+    }
+
+    @Override
+    protected GetAllPitNodesResponse newResponse(
+        GetAllPitNodesRequest request,
+        List<GetAllPitNodeResponse> getAllPitNodeResponses,
+        List<FailedNodeException> failures
+    ) {
+        return new GetAllPitNodesResponse(clusterService.getClusterName(), getAllPitNodeResponses, failures);
+    }
+
+    @Override
+    protected GetAllPitNodeRequest newNodeRequest(GetAllPitNodesRequest request) {
+        return new GetAllPitNodeRequest();
+    }
+
+    @Override
+    protected GetAllPitNodeResponse newNodeResponse(StreamInput in) throws IOException {
+        return new GetAllPitNodeResponse(in);
     }
 
-    protected void doExecute(Task task, GetAllPitNodesRequest request, ActionListener<GetAllPitNodesResponse> listener) {
-        // If security plugin intercepts the request, it'll replace all PIT IDs with permitted PIT IDs
-        if (request.getGetAllPitNodesResponse() != null) {
-            listener.onResponse(request.getGetAllPitNodesResponse());
-        }
+    /**
+     * This retrieves all active PITs in the node
+     */
+    @Override
+    protected GetAllPitNodeResponse nodeOperation(GetAllPitNodeRequest request) {
+        GetAllPitNodeResponse nodeResponse = new GetAllPitNodeResponse(
+            transportService.getLocalNode(),
+            searchService.getAllPITReaderContexts()
+        );
+        return nodeResponse;
     }
 }