AMBARI-8351. Restart ganglia starts off claiming aborted - completes anyway (dlysnichenko)

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

@@ -48,17 +48,12 @@ import org.apache.ambari.server.agent.ExecutionCommand;
 import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.configuration.Configuration;
+import org.apache.ambari.server.controller.internal.RequestOperationLevel;
 import org.apache.ambari.server.controller.internal.RequestResourceFilter;
+import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.metadata.ActionMetadata;
 import org.apache.ambari.server.state.Cluster;
-import org.apache.ambari.server.state.Clusters;
-import org.apache.ambari.server.state.CommandScriptDefinition;
-import org.apache.ambari.server.state.ComponentInfo;
-import org.apache.ambari.server.state.ConfigHelper;
-import org.apache.ambari.server.state.CustomCommandDefinition;
-import org.apache.ambari.server.state.Host;
-import org.apache.ambari.server.state.HostComponentAdminState;
-import org.apache.ambari.server.state.MaintenanceState;
+import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.PropertyInfo.PropertyType;
 import org.apache.ambari.server.state.RepositoryInfo;
 import org.apache.ambari.server.state.Service;
@@ -68,6 +63,15 @@ import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.StackInfo;
 import org.apache.ambari.server.state.State;
+import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.CommandScriptDefinition;
+import org.apache.ambari.server.state.ComponentInfo;
+import org.apache.ambari.server.state.ConfigHelper;
+import org.apache.ambari.server.state.CustomCommandDefinition;
+import org.apache.ambari.server.state.Host;
+import org.apache.ambari.server.state.HostComponentAdminState;
+import org.apache.ambari.server.state.MaintenanceState;
+
 import org.apache.ambari.server.state.stack.OsFamily;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
 import org.apache.ambari.server.utils.StageUtils;
@@ -232,11 +236,20 @@ public class AmbariCustomCommandExecutionHelper {
               }
             }
     );
-    if (! ignoredHosts.isEmpty()) {
+
+    // Filter unhealthy hosts
+    Set<String> filteredHosts = filterUnhealthHosts(candidateHosts, actionExecutionContext, resourceFilter);
+
+    if (!ignoredHosts.isEmpty()) {
       String message = String.format("Some hosts (%s) have been ignored " +
                       "because components on them are in Maintenance state.",
               ignoredHosts);
       LOG.debug(message);
+    } else if (!filteredHosts.isEmpty()) {
+      String message = String.format("Some hosts (%s) have been ignored " +
+            "because they are in unknown state",
+         filteredHosts);
+      LOG.warn(message);
     } else if (candidateHosts.isEmpty()) {
       String message = "Invalid request : No hosts specified.";
       throw new AmbariException(message);
@@ -245,9 +258,9 @@ public class AmbariCustomCommandExecutionHelper {
     StackId stackId = cluster.getDesiredStackVersion();
     AmbariMetaInfo ambariMetaInfo = managementController.getAmbariMetaInfo();
     ServiceInfo serviceInfo = ambariMetaInfo.getService(
-        stackId.getStackName(), stackId.getStackVersion(), serviceName);
+       stackId.getStackName(), stackId.getStackVersion(), serviceName);
     StackInfo stackInfo = ambariMetaInfo.getStack
-        (stackId.getStackName(), stackId.getStackVersion());
+       (stackId.getStackName(), stackId.getStackVersion());
 
     CustomCommandDefinition customCommandDefinition = null;
     ComponentInfo ci = serviceInfo.getComponentByName(componentName);
@@ -877,4 +890,74 @@ public class AmbariCustomCommandExecutionHelper {
 
     return repoInfo;
   }
+
+  private ServiceComponent getServiceComponent ( ActionExecutionContext actionExecutionContext,
+                                                RequestResourceFilter resourceFilter){
+    try {
+      Cluster cluster = clusters.getCluster(actionExecutionContext.getClusterName());
+      Service service = cluster.getService(resourceFilter.getServiceName());
+
+      return service.getServiceComponent(resourceFilter.getComponentName());
+    } catch (Exception e) {
+      LOG.debug(String.format( "Unknown error appears during getting service component: %s", e.getMessage()));
+    }
+    return null;
+  }
+
+  /**
+   * Filter host according to status of host/host components
+   * @param hostname Host name to check
+   * @param actionExecutionContext Received request to execute a command
+   * @param resourceFilter Resource filter
+   * @return True if host need to be filtered, False if Not
+   * @throws AmbariException
+   */
+  private boolean filterUnhealthHostItem(String hostname,
+                                         ActionExecutionContext actionExecutionContext,
+                                         RequestResourceFilter resourceFilter) throws AmbariException {
+
+    RequestOperationLevel operationLevel = actionExecutionContext.getOperationLevel();
+    ServiceComponent serviceComponent = getServiceComponent(actionExecutionContext, resourceFilter);
+    if (serviceComponent != null && operationLevel != null
+                                && operationLevel.getLevel() == Resource.Type.Service // compare operation is allowed only for Service operation level
+                                && actionExecutionContext.getResourceFilters().size() > 1  // Check if operation was started in a chain
+                                && !serviceComponent.isMasterComponent()
+       ){
+
+      return !(clusters.getHost(hostname).getState() == HostState.HEALTHY);
+    } else if (serviceComponent != null && operationLevel != null
+                                        && operationLevel.getLevel() == Resource.Type.Host  // compare operation is allowed only for host component operation level
+                                        && actionExecutionContext.getResourceFilters().size() > 1  // Check if operation was started in a chain
+                                        && serviceComponent.getServiceComponentHosts().containsKey(hostname)  // Check if host is assigned to host component
+                                        && !serviceComponent.isMasterComponent()
+       ){
+
+      State hostState = serviceComponent.getServiceComponentHosts().get(hostname).getState();
+
+      return hostState == State.UNKNOWN;
+    }
+    return false;
+  }
+
+
+  /**
+   * Filter hosts according to status of host/host components
+   * @param hosts Host name set to filter
+   * @param actionExecutionContext Received request to execute a command
+   * @param resourceFilter Resource filter
+   * @return Set of excluded hosts
+   * @throws AmbariException
+   */
+  private Set<String> filterUnhealthHosts(Set<String> hosts,
+                                          ActionExecutionContext actionExecutionContext,
+                                          RequestResourceFilter resourceFilter) throws AmbariException {
+    Set<String> removedHosts = new HashSet<String>();
+    for (String hostname : hosts) {
+      if (filterUnhealthHostItem(hostname, actionExecutionContext, resourceFilter)){
+        removedHosts.add(hostname);
+      }
+    }
+    hosts.removeAll(removedHosts);
+    return removedHosts;
+  }
 }

ambari-server/src/test/java/org/apache/ambari/server/controller/AmbariCustomCommandExecutionHelperTest.java

@@ -19,6 +19,7 @@ package org.apache.ambari.server.controller;
 
 import static org.mockito.Matchers.any;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -35,8 +36,10 @@ import org.apache.ambari.server.actionmanager.Stage;
 import org.apache.ambari.server.agent.ExecutionCommand;
 import org.apache.ambari.server.api.services.AmbariMetaInfo;
 import org.apache.ambari.server.controller.internal.ComponentResourceProviderTest;
+import org.apache.ambari.server.controller.internal.RequestOperationLevel;
 import org.apache.ambari.server.controller.internal.RequestResourceFilter;
 import org.apache.ambari.server.controller.internal.ServiceResourceProviderTest;
+import org.apache.ambari.server.controller.spi.Resource;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.state.Clusters;
@@ -143,7 +146,133 @@ public class AmbariCustomCommandExecutionHelperTest {
     }
   }
 
-  
+  @Test
+  public void testHostsFilterHealthy(){
+    try {
+      createClusterFixture();
+
+      Map<String, String> requestProperties = new HashMap<String, String>() {
+        {
+          put("context" , "Restart all components for GANGLIA");
+          put("operation_level/level", "SERVICE");
+          put("operation_level/service_name", "GANGLIA");
+          put("operation_level/cluster_name", "c1");
+        }
+      };
+
+      ExecuteActionRequest actionRequest = new ExecuteActionRequest(
+         "c1", "RESTART", null,
+         Arrays.asList(
+            new RequestResourceFilter("GANGLIA", "GANGLIA_SERVER", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6402"))
+         ),
+         new RequestOperationLevel(Resource.Type.Service, "c1", "GANGLIA", null, null),
+         new HashMap<String, String>(){{}},
+         false);
+
+      controller.createAction(actionRequest, requestProperties);
+
+      //clusters.getHost("c6402").setState(HostState.HEARTBEAT_LOST);
+
+      Mockito.verify(am, Mockito.times(1)).sendActions(stagesCaptor.capture(), any(ExecuteActionRequest.class));
+
+      List<Stage> stages = stagesCaptor.getValue();
+      Assert.assertEquals(1, stages.size());
+
+      Stage stage = stages.get(0);
+       // Check if was generated command, one for each host
+      Assert.assertEquals(2, stage.getHostRoleCommands().size());
+    }catch (Exception e) {
+      Assert.fail(e.getMessage());
+    }
+  }
+
+  @Test
+  public void testHostsFilterUnhealthyHost(){
+    try {
+      createClusterFixture();
+
+      // Set custom status to host
+      clusters.getHost("c6402").setState(HostState.HEARTBEAT_LOST);
+      Map<String, String> requestProperties = new HashMap<String, String>() {
+        {
+          put("context" , "Restart all components for GANGLIA");
+          put("operation_level/level", "SERVICE");
+          put("operation_level/service_name", "GANGLIA");
+          put("operation_level/cluster_name", "c1");
+        }
+      };
+
+      ExecuteActionRequest actionRequest = new ExecuteActionRequest(
+         "c1", "RESTART", null,
+         Arrays.asList(
+            new RequestResourceFilter("GANGLIA", "GANGLIA_SERVER", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6402"))
+         ),
+         new RequestOperationLevel(Resource.Type.Service, "c1", "GANGLIA", null, null),
+         new HashMap<String, String>(){{}},
+         false);
+
+      controller.createAction(actionRequest, requestProperties);
+
+      Mockito.verify(am, Mockito.times(1)).sendActions(stagesCaptor.capture(), any(ExecuteActionRequest.class));
+
+      List<Stage> stages = stagesCaptor.getValue();
+      Assert.assertEquals(1, stages.size());
+
+      Stage stage = stages.get(0);
+      // Check if was generated command for one health host
+      Assert.assertEquals(1, stage.getHostRoleCommands().size());
+    }catch (Exception e) {
+      Assert.fail(e.getMessage());
+    }
+  }
+
+  @Test
+  public void testHostsFilterUnhealthyComponent(){
+    try {
+      createClusterFixture();
+
+      // Set custom status to host
+      clusters.getCluster("c1").getService("GANGLIA").getServiceComponent("GANGLIA_MONITOR").getServiceComponentHost("c6402")
+         .setState(State.UNKNOWN);
+      Map<String, String> requestProperties = new HashMap<String, String>() {
+        {
+          put("context" , "Restart all components for GANGLIA");
+          put("operation_level/level", "SERVICE");
+          put("operation_level/service_name", "GANGLIA");
+          put("operation_level/cluster_name", "c1");
+        }
+      };
+
+      ExecuteActionRequest actionRequest = new ExecuteActionRequest(
+         "c1", "RESTART", null,
+         Arrays.asList(
+            new RequestResourceFilter("GANGLIA", "GANGLIA_SERVER", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6401")),
+            new RequestResourceFilter("GANGLIA", "GANGLIA_MONITOR", Collections.singletonList("c6402"))
+         ),
+         new RequestOperationLevel(Resource.Type.Host, "c1", "GANGLIA", null, null),
+         new HashMap<String, String>(){{}},
+         false);
+
+      controller.createAction(actionRequest, requestProperties);
+
+      Mockito.verify(am, Mockito.times(1)).sendActions(stagesCaptor.capture(), any(ExecuteActionRequest.class));
+
+      List<Stage> stages = stagesCaptor.getValue();
+      Assert.assertEquals(1, stages.size());
+
+      Stage stage = stages.get(0);
+      // Check if was generated command for one health host
+      Assert.assertEquals(1, stage.getHostRoleCommands().size());
+    }catch (Exception e) {
+      Assert.fail(e.getMessage());
+    }
+  }
+
   private void createClusterFixture() throws AmbariException {
     createCluster("c1");
     addHost("c6401","c1");
@@ -151,14 +280,21 @@ public class AmbariCustomCommandExecutionHelperTest {
     
     clusters.getCluster("c1");
     createService("c1", "YARN", null);
-    
-    createServiceComponent("c1","YARN","RESOURCEMANAGER", State.INIT);
-    createServiceComponent("c1","YARN","NODEMANAGER", State.INIT);
+    createService("c1", "GANGLIA", null);
+
+    createServiceComponent("c1", "YARN","RESOURCEMANAGER", State.INIT);
+    createServiceComponent("c1", "YARN", "NODEMANAGER", State.INIT);
+    createServiceComponent("c1", "GANGLIA", "GANGLIA_SERVER", State.INIT);
+    createServiceComponent("c1", "GANGLIA", "GANGLIA_MONITOR", State.INIT);
     
     createServiceComponentHost("c1","YARN","RESOURCEMANAGER","c6401", null);
     createServiceComponentHost("c1","YARN","NODEMANAGER","c6401", null);
+    createServiceComponentHost("c1","GANGLIA","GANGLIA_SERVER","c6401", State.INIT);
+    createServiceComponentHost("c1","GANGLIA","GANGLIA_MONITOR","c6401", State.INIT);
     
     createServiceComponentHost("c1","YARN","NODEMANAGER","c6402", null);
+    createServiceComponentHost("c1","GANGLIA","GANGLIA_MONITOR","c6402", State.INIT);
+
   }
   private void addHost(String hostname, String clusterName) throws AmbariException {
     clusters.addHost(hostname);