AMBARI-6673 Expose ability to cancel requests via API (Dmitro Lisnichenko via dsen)

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/ActionManager.java

@@ -219,4 +219,9 @@ public class ActionManager {
   public String getRequestContext(long requestId) {
     return db.getRequestContext(requestId);
   }
+
+  public void cancelRequest(Request request) {
+    // TODO: implement (dlysnichenko)
+    // TODO : what if cluster name == null?
+  }
 }

ambari-server/src/main/java/org/apache/ambari/server/actionmanager/Request.java

@@ -53,7 +53,6 @@ public class Request {
   private long createTime;
   private long startTime;
   private long endTime;
-  private HostRoleStatus status; // not persisted yet
   private String inputs;
   private List<RequestResourceFilter> resourceFilters;
   private RequestOperationLevel operationLevel;
@@ -160,7 +159,6 @@ public class Request {
 
     this.requestType = entity.getRequestType();
     this.commandName = entity.getCommandName();
-    this.status = entity.getStatus();
     if (entity.getRequestScheduleEntity() != null) {
       this.requestScheduleId = entity.getRequestScheduleEntity().getScheduleId();
     }
@@ -373,11 +371,4 @@ public class Request {
         '}';
   }
 
-  public HostRoleStatus getStatus() {
-    return status;
-  }
-
-  public void setStatus(HostRoleStatus status) {
-    this.status = status;
-  }
 }

ambari-server/src/main/java/org/apache/ambari/server/api/services/RequestService.java

@@ -24,6 +24,7 @@ import org.apache.ambari.server.controller.spi.Resource;
 
 import javax.ws.rs.GET;
 import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
@@ -103,6 +104,23 @@ public class RequestService extends BaseService {
     return new TaskService(m_clusterName, requestId);
   }
 
+  /**
+   * Handles: PUT /clusters/{clusterId}/requests/{requestId} or /requests/{requestId}
+   * Change state of existing requests. Usually used to cancel running requests
+   *
+   * @param body        http body
+   * @param headers     http headers
+   * @param ui          uri info
+   * @return information regarding the created services
+   */
+  @PUT
+  @Path("{requestId}")
+  @Produces("text/plain")
+  public Response updateRequests(String body, @Context HttpHeaders headers, @Context UriInfo ui,
+                                 @PathParam("requestId") String requestId) {
+    return handleRequest(headers, body, ui, Request.Type.PUT, createRequestResource(m_clusterName, requestId));
+  }
+
   /**
    * Handles: POST /clusters/{clusterId}/requests or /requests
    * Create multiple services.

ambari-server/src/main/java/org/apache/ambari/server/controller/RequestRequest.java

@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.controller;
+
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+
+/**
+ * Encapsulates all data about update request that came to RequestResourceProvider
+ */
+public class RequestRequest {
+
+  public RequestRequest(String clusterName, long requestId) {
+    this.clusterName = clusterName;
+    this.requestId = requestId;
+  }
+
+  private String clusterName;
+
+  private long requestId;
+
+  private HostRoleStatus status;
+
+  private String abortReason;
+
+
+  public HostRoleStatus getStatus() {
+    return status;
+  }
+
+  public String getClusterName() {
+    return clusterName;
+  }
+
+  public void setClusterName(String clusterName) {
+    this.clusterName = clusterName;
+  }
+
+  public long getRequestId() {
+    return requestId;
+  }
+
+  public void setRequestId(int requestId) {
+    this.requestId = requestId;
+  }
+
+  public void setStatus(HostRoleStatus status) {
+    this.status = status;
+  }
+
+  public String getAbortReason() {
+    return abortReason;
+  }
+
+  public void setAbortReason(String abortReason) {
+    this.abortReason = abortReason;
+  }
+
+  @Override
+  public String toString() {
+    return "RequestRequest{" +
+            "clusterName='" + clusterName + '\'' +
+            ", requestId=" + requestId +
+            ", status=" + status +
+            ", abortReason='" + abortReason + '\'' +
+            '}';
+  }
+}

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/RequestResourceProvider.java

@@ -24,6 +24,7 @@ import org.apache.ambari.server.actionmanager.HostRoleStatus;
 import org.apache.ambari.server.api.services.BaseRequest;
 import org.apache.ambari.server.controller.AmbariManagementController;
 import org.apache.ambari.server.controller.ExecuteActionRequest;
+import org.apache.ambari.server.controller.RequestRequest;
 import org.apache.ambari.server.controller.RequestStatusResponse;
 import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
 import org.apache.ambari.server.controller.spi.NoSuchResourceException;
@@ -42,6 +43,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -56,6 +58,7 @@ public class RequestResourceProvider extends AbstractControllerResourceProvider
   protected static final String REQUEST_CLUSTER_NAME_PROPERTY_ID = "Requests/cluster_name";
   protected static final String REQUEST_ID_PROPERTY_ID = "Requests/id";
   protected static final String REQUEST_STATUS_PROPERTY_ID = "Requests/request_status";
+  protected static final String REQUEST_ABORT_REASON_PROPERTY_ID = "Requests/abort_reason";
   protected static final String REQUEST_CONTEXT_ID = "Requests/request_context";
   public static final String REQUEST_SOURCE_SCHEDULE = "Requests/request_schedule";
   public static final String REQUEST_SOURCE_SCHEDULE_ID = "Requests/request_schedule/schedule_id";
@@ -159,9 +162,79 @@ public class RequestResourceProvider extends AbstractControllerResourceProvider
   }
 
   @Override
-  public RequestStatus updateResources(Request request, Predicate predicate)
-      throws SystemException, UnsupportedPropertyException, NoSuchResourceException, NoSuchParentResourceException {
-    throw new UnsupportedOperationException("Not currently supported.");
+  public RequestStatus updateResources(Request requestInfo, Predicate predicate)
+          throws SystemException, UnsupportedPropertyException,
+          NoSuchResourceException, NoSuchParentResourceException {
+    AmbariManagementController amc = getManagementController();
+    final Set<RequestRequest> requests = new HashSet<RequestRequest>();
+
+    Iterator<Map<String,Object>> iterator = requestInfo.getProperties().iterator();
+    if (iterator.hasNext()) {
+      for (Map<String, Object> propertyMap : getPropertyMaps(iterator.next(), predicate)) {
+        requests.add(getRequest(propertyMap));
+      }
+    }
+    // Validate
+    List<org.apache.ambari.server.actionmanager.Request> targets =
+            new ArrayList<org.apache.ambari.server.actionmanager.Request>();
+    for (RequestRequest updateRequest : requests) {
+      ActionManager actionManager = amc.getActionManager();
+      List<org.apache.ambari.server.actionmanager.Request> internalRequests =
+              actionManager.getRequests(Collections.singletonList(updateRequest.getRequestId()));
+      if (internalRequests.size() == 0) {
+        throw new IllegalArgumentException(
+                String.format("Request %s does not exist", updateRequest.getRequestId()));
+      }
+      // There should be only one request with this id (or no request at all)
+      org.apache.ambari.server.actionmanager.Request internalRequest = internalRequests.get(0);
+      // Validate update request (check constraints on state value and presense of abort reason)
+      if (updateRequest.getAbortReason() == null || updateRequest.getAbortReason().isEmpty()) {
+        throw new IllegalArgumentException("Abort reason can not be empty.");
+      }
+
+      HostRoleStatus internalRequestStatus = calculateSummaryStatus(
+              calculateRequestStatusCounters(internalRequest),
+              internalRequest.getCommands().size()
+      );
+      if (updateRequest.getStatus() != HostRoleStatus.ABORTED) {
+        throw new IllegalArgumentException(
+                String.format("%s is wrong value. The only allowed value " +
+                                "for updating request status is ABORTED",
+                        updateRequest.getStatus()));
+      }
+      if (internalRequestStatus.isCompletedState()) {
+        throw new IllegalArgumentException(
+                String.format("Can not set request that is in %s state to %s state.",
+                        internalRequestStatus.toString(), updateRequest.getStatus()));
+      }
+      // Validation passed
+      targets.add(internalRequest);
+    }
+    // Perform update
+    for (org.apache.ambari.server.actionmanager.Request target : targets) {
+      amc.getActionManager().cancelRequest(target);
+    }
+    return getRequestStatus(null);
+  }
+
+  private RequestRequest getRequest(Map<String, Object> propertyMap) {
+    // Cluster name may be empty for custom actions
+    String clusterNameStr = (String) propertyMap.get(REQUEST_CLUSTER_NAME_PROPERTY_ID);
+    String requestIdStr = (String) propertyMap.get(REQUEST_ID_PROPERTY_ID);
+    long requestId = Integer.valueOf(requestIdStr);
+    String requestStatusStr = (String) propertyMap.get(REQUEST_STATUS_PROPERTY_ID);
+    HostRoleStatus requestStatus = null;
+    if (requestStatusStr != null) {
+      // This conversion may throw IllegalArgumentException, it is OK
+      // in this case it will be mapped to HTTP 400 Bad Request
+      requestStatus = HostRoleStatus.valueOf(requestStatusStr);
+    }
+    String abortReason = (String) propertyMap.get(REQUEST_ABORT_REASON_PROPERTY_ID);
+    RequestRequest requestRequest = new RequestRequest(clusterNameStr, requestId);
+    requestRequest.setStatus(requestStatus);
+    requestRequest.setAbortReason(abortReason);
+    return requestRequest;
+
   }
 
   @Override
@@ -351,67 +424,76 @@ public class RequestResourceProvider extends AbstractControllerResourceProvider
     } else {
       setResourceProperty(resource, REQUEST_SOURCE_SCHEDULE, null, requestedPropertyIds);
     }
+    int taskCount = request.getCommands().size();
 
-    List<HostRoleCommand> commands = request.getCommands();
-
-    int taskCount = commands.size();
-    int completedTaskCount = 0;
-    int queuedTaskCount = 0;
-    int pendingTaskCount = 0;
-    int failedTaskCount = 0;
-    int abortedTaskCount = 0;
-    int timedOutTaskCount = 0;
-
-    for (HostRoleCommand hostRoleCommand : commands) {
-      HostRoleStatus status = hostRoleCommand.getStatus();
-      if (status.isCompletedState()) {
-        completedTaskCount++;
-
-        switch (status) {
-          case ABORTED:
-            abortedTaskCount++;
-            break;
-          case FAILED:
-            failedTaskCount++;
-            break;
-          case TIMEDOUT:
-            timedOutTaskCount++;
-            break;
-        }
-      } else if (status.equals(HostRoleStatus.QUEUED)) {
-        queuedTaskCount++;
-      } else if (status.equals(HostRoleStatus.PENDING)) {
-        pendingTaskCount++;
-      }
-    }
-
-    int inProgressTaskCount = taskCount - completedTaskCount - queuedTaskCount - pendingTaskCount;
+    Map<HostRoleStatus, Integer> hostRoleStatusCounters = calculateRequestStatusCounters(request);
+    HostRoleStatus requestStatus = calculateSummaryStatus(hostRoleStatusCounters, taskCount);
 
-    LOG.debug("taskCount={}, inProgressTaskCount={}, completedTaskCount={}, queuedTaskCount={}, " +
-      "pendingTaskCount={}, failedTaskCount={}, abortedTaskCount={}, timedOutTaskCount={}",
-      taskCount, inProgressTaskCount, completedTaskCount, queuedTaskCount, pendingTaskCount,
-      failedTaskCount, abortedTaskCount, timedOutTaskCount);
-
-    // determine request status
-    HostRoleStatus requestStatus = failedTaskCount > 0 ? HostRoleStatus.FAILED :
-        abortedTaskCount > 0 ? HostRoleStatus.ABORTED :
-            timedOutTaskCount > 0 ? HostRoleStatus.TIMEDOUT :
-                inProgressTaskCount > 0 ? HostRoleStatus.IN_PROGRESS :
-                    completedTaskCount == taskCount ? HostRoleStatus.COMPLETED :
-                        HostRoleStatus.PENDING;
     double progressPercent =
-        ((queuedTaskCount * 0.09 + inProgressTaskCount * 0.35 + completedTaskCount) / (double) taskCount) * 100.0;
+            ((hostRoleStatusCounters.get(HostRoleStatus.QUEUED) * 0.09 +
+                    hostRoleStatusCounters.get(HostRoleStatus.IN_PROGRESS) * 0.35 +
+                    hostRoleStatusCounters.get(HostRoleStatus.COMPLETED)) / (double) taskCount) * 100.0;
 
     setResourceProperty(resource, REQUEST_STATUS_PROPERTY_ID, requestStatus.toString(), requestedPropertyIds);
     setResourceProperty(resource, REQUEST_TASK_CNT_ID, taskCount, requestedPropertyIds);
-    setResourceProperty(resource, REQUEST_FAILED_TASK_CNT_ID, failedTaskCount, requestedPropertyIds);
-    setResourceProperty(resource, REQUEST_ABORTED_TASK_CNT_ID, abortedTaskCount, requestedPropertyIds);
-    setResourceProperty(resource, REQUEST_TIMED_OUT_TASK_CNT_ID, timedOutTaskCount, requestedPropertyIds);
-    setResourceProperty(resource, REQUEST_QUEUED_TASK_CNT_ID, queuedTaskCount, requestedPropertyIds);
-    setResourceProperty(resource, REQUEST_COMPLETED_TASK_CNT_ID, completedTaskCount, requestedPropertyIds);
+    setResourceProperty(resource, REQUEST_FAILED_TASK_CNT_ID,
+            hostRoleStatusCounters.get(HostRoleStatus.FAILED), requestedPropertyIds);
+    setResourceProperty(resource, REQUEST_ABORTED_TASK_CNT_ID,
+            hostRoleStatusCounters.get(HostRoleStatus.ABORTED), requestedPropertyIds);
+    setResourceProperty(resource, REQUEST_TIMED_OUT_TASK_CNT_ID,
+            hostRoleStatusCounters.get(HostRoleStatus.TIMEDOUT), requestedPropertyIds);
+    setResourceProperty(resource, REQUEST_QUEUED_TASK_CNT_ID,
+            hostRoleStatusCounters.get(HostRoleStatus.QUEUED), requestedPropertyIds);
+    setResourceProperty(resource, REQUEST_COMPLETED_TASK_CNT_ID,
+            hostRoleStatusCounters.get(HostRoleStatus.COMPLETED), requestedPropertyIds);
     setResourceProperty(resource, REQUEST_PROGRESS_PERCENT_ID, progressPercent, requestedPropertyIds);
 
     return resource;
   }
 
+  /**
+   * Returns counts of tasks that are in various states.
+   */
+  private Map<HostRoleStatus, Integer> calculateRequestStatusCounters(org.apache.ambari.server.actionmanager.Request request) {
+    List<HostRoleCommand> commands = request.getCommands();
+    Map<HostRoleStatus, Integer> counters = new HashMap<HostRoleStatus, Integer>();
+    int totalTasks = commands.size();
+    // initialize
+    for (HostRoleStatus hostRoleStatus : HostRoleStatus.values()) {
+      counters.put(hostRoleStatus, 0);
+    }
+    // calculate counts
+    for (HostRoleCommand hostRoleCommand : commands) {
+      HostRoleStatus status = hostRoleCommand.getStatus();
+      if (status.isCompletedState() &&
+            status != HostRoleStatus.COMPLETED ) { // we don't want to count twice
+        // Increase total number of completed tasks;
+        counters.put(HostRoleStatus.COMPLETED, counters.get(HostRoleStatus.COMPLETED) + 1);
+      }
+      // Increment counter for particular status
+      counters.put(status, counters.get(status) + 1);
+    }
+    // We overwrite the value to have the sum converged
+    counters.put(HostRoleStatus.IN_PROGRESS,
+            totalTasks - counters.get(HostRoleStatus.COMPLETED)-
+                    counters.get(HostRoleStatus.QUEUED)-
+                    counters.get(HostRoleStatus.PENDING));
+    return counters;
+  }
+
+  /**
+   * @param counters counts of tasks that are in various states.
+   * @param totalTasks total number of tasks in request
+   * @return summary request status based on statuses of tasks in different
+   * states.
+   */
+  private HostRoleStatus calculateSummaryStatus(Map<HostRoleStatus, Integer> counters, int totalTasks) {
+    return counters.get(HostRoleStatus.FAILED) > 0 ? HostRoleStatus.FAILED :
+            // TODO (dlysnichenko): maybe change order of FAILED and ABORTED?
+            counters.get(HostRoleStatus.ABORTED) > 0 ? HostRoleStatus.ABORTED :
+                    counters.get(HostRoleStatus.TIMEDOUT) > 0 ? HostRoleStatus.TIMEDOUT :
+                            counters.get(HostRoleStatus.IN_PROGRESS) > 0 ? HostRoleStatus.IN_PROGRESS :
+                                    counters.get(HostRoleStatus.COMPLETED) == totalTasks ? HostRoleStatus.COMPLETED :
+                                            HostRoleStatus.PENDING;
+  }
 }

ambari-server/src/main/resources/properties.json

@@ -123,6 +123,7 @@
         "Requests/completed_task_count",
         "Requests/queued_task_count",
         "Requests/progress_percent",
+        "Requests/abort_reason",
         "_"
     ],
     "RequestSchedule" : [

ambari-server/src/test/java/org/apache/ambari/server/controller/internal/RequestResourceProviderTest.java

@@ -760,14 +760,33 @@ public class RequestResourceProviderTest {
   }
 
   @Test
-  public void testUpdateResources() throws Exception {
+  public void testUpdateResources_CancelRequest() throws Exception {
     Resource.Type type = Resource.Type.Request;
 
     AmbariManagementController managementController = createMock(AmbariManagementController.class);
+    ActionManager actionManager = createNiceMock(ActionManager.class);
+    HostRoleCommand hostRoleCommand = createNiceMock(HostRoleCommand.class);
+    Clusters clusters = createNiceMock(Clusters.class);
+
+    List<HostRoleCommand> hostRoleCommands = new LinkedList<HostRoleCommand>();
+    hostRoleCommands.add(hostRoleCommand);
+
+    org.apache.ambari.server.actionmanager.Request requestMock =
+            createNiceMock(org.apache.ambari.server.actionmanager.Request.class);
+    expect(requestMock.getCommands()).andReturn(hostRoleCommands).anyTimes();
+
+    Capture<Collection<Long>> requestIdsCapture = new Capture<Collection<Long>>();
+
+    // set expectations
+    expect(managementController.getActionManager()).andReturn(actionManager).anyTimes();
+    expect(actionManager.getRequests(capture(requestIdsCapture))).
+            andReturn(Collections.singletonList(requestMock)).anyTimes();
+    expect(hostRoleCommand.getStatus()).andReturn(HostRoleStatus.IN_PROGRESS).anyTimes();
+
     RequestStatusResponse response = createNiceMock(RequestStatusResponse.class);
 
     // replay
-    replay(managementController, response);
+    replay(managementController, actionManager, hostRoleCommand, clusters, requestMock, response);
 
     ResourceProvider provider = AbstractControllerResourceProvider.getResourceProvider(
         type,
@@ -775,22 +794,65 @@ public class RequestResourceProviderTest {
         PropertyHelper.getKeyPropertyIds(type),
         managementController);
 
+    // TEST CASE: Check update request validation (abort reason not specified)
     // add the property map to a set for the request.
     Map<String, Object> properties = new LinkedHashMap<String, Object>();
-
     // create the request
     Request request = PropertyHelper.getUpdateRequest(properties, null);
-
     Predicate predicate = new PredicateBuilder().property(RequestResourceProvider.REQUEST_ID_PROPERTY_ID).
-        equals("Request100").toPredicate();
+        equals("100").toPredicate();
 
     try {
       provider.updateResources(request, predicate);
-      Assert.fail("Expected an UnsupportedOperationException");
-    } catch (UnsupportedOperationException e) {
+      Assert.fail("Expected an java.lang.IllegalArgumentException: Abort reason can not be empty.");
+    } catch (IllegalArgumentException e) {
+      // expected
+    }
+
+    // Add abort reason to previous request
+    properties.put(RequestResourceProvider.REQUEST_ABORT_REASON_PROPERTY_ID, "Some reason");
+
+    // TEST CASE: Check update request validation (new state is not specified)
+    request = PropertyHelper.getUpdateRequest(properties, null);
+    try {
+      provider.updateResources(request, predicate);
+      Assert.fail("Expected an java.lang.IllegalArgumentException: null is wrong value.");
+    } catch (IllegalArgumentException e) {
       // expected
     }
 
+    // TEST CASE: Check update request validation (new state is wrong)
+    properties.put(RequestResourceProvider.REQUEST_STATUS_PROPERTY_ID, "COMPLETED");
+    request = PropertyHelper.getUpdateRequest(properties, null);
+    try {
+      provider.updateResources(request, predicate);
+      Assert.fail("Expected an java.lang.IllegalArgumentException: COMPLETED is wrong value. " +
+              "The only allowed value for updating request status is ABORTED");
+    } catch (IllegalArgumentException e) {
+      // expected
+    }
+
+    // TEST CASE: Check update request validation (request is in wrong state)
+    // Put valid request status
+    properties.put(RequestResourceProvider.REQUEST_STATUS_PROPERTY_ID, "ABORTED");
+    for (HostRoleStatus status : HostRoleStatus.values()) {
+      reset(hostRoleCommand);
+      expect(hostRoleCommand.getStatus()).andReturn(status).anyTimes();
+      replay(hostRoleCommand);
+      request = PropertyHelper.getUpdateRequest(properties, null);
+      if (status == HostRoleStatus.IN_PROGRESS ||
+              status == HostRoleStatus.PENDING ||
+              status == HostRoleStatus.QUEUED) { // the only valid cases
+        provider.updateResources(request, predicate);
+      } else {  // In other cases, should error out
+        try {
+          provider.updateResources(request, predicate);
+          Assert.fail("Expected an java.lang.IllegalArgumentException: null is wrong value.");
+        } catch (IllegalArgumentException e) {
+          // expected
+        }
+      }
+    }
     // verify
     verify(managementController, response);
   }