AMBARI-18619. Optimize Service Checks to it picks a random host and prefers hosts with 0 active commands (alejandro)

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

@@ -53,6 +53,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Random;
 import java.util.Set;
 import java.util.TreeMap;
 
@@ -60,6 +61,7 @@ import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.Role;
 import org.apache.ambari.server.RoleCommand;
 import org.apache.ambari.server.actionmanager.HostRoleCommand;
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
 import org.apache.ambari.server.actionmanager.Stage;
 import org.apache.ambari.server.agent.AgentCommand.AgentCommandType;
 import org.apache.ambari.server.agent.ExecutionCommand;
@@ -71,6 +73,7 @@ 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.orm.dao.ClusterVersionDAO;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
 import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
 import org.apache.ambari.server.orm.entities.OperatingSystemEntity;
 import org.apache.ambari.server.orm.entities.RepositoryEntity;
@@ -169,6 +172,9 @@ public class AmbariCustomCommandExecutionHelper {
   @Inject
   private ClusterVersionDAO clusterVersionDAO;
 
+  @Inject
+  private HostRoleCommandDAO hostRoleCommandDAO;
+
   protected static final String SERVICE_CHECK_COMMAND_NAME = "SERVICE_CHECK";
   protected static final String START_COMMAND_NAME = "START";
   protected static final String RESTART_COMMAND_NAME = "RESTART";
@@ -500,31 +506,36 @@ public class AmbariCustomCommandExecutionHelper {
       smokeTestRole = actionExecutionContext.getActionName();
     }
 
-    long nowTimestamp = System.currentTimeMillis();
-    Map<String, String> actionParameters = actionExecutionContext.getParameters();
-    final Set<String> candidateHosts;
+    Set<String> candidateHosts;
     final Map<String, ServiceComponentHost> serviceHostComponents;
 
     if (componentName != null) {
-      serviceHostComponents = cluster.getService(serviceName).getServiceComponent(
-          componentName).getServiceComponentHosts();
+      serviceHostComponents = cluster.getService(serviceName).getServiceComponent(componentName).getServiceComponentHosts();
 
       if (serviceHostComponents.isEmpty()) {
-        throw new AmbariException("Hosts not found, component="
-            + componentName + ", service = " + serviceName
-            + ", cluster = " + clusterName);
+        throw new AmbariException(MessageFormat.format("No hosts found for service: {0}, component: {1} in cluster: {2}",
+            serviceName, componentName, clusterName));
       }
 
+      // If specified a specific host, run on it as long as it contains the component.
+      // Otherwise, use candidates that contain the component.
       List<String> candidateHostsList = resourceFilter.getHostNames();
       if (candidateHostsList != null && !candidateHostsList.isEmpty()) {
         candidateHosts = new HashSet<String>(candidateHostsList);
+
+        // Get the intersection.
+        candidateHosts.retainAll(serviceHostComponents.keySet());
+
+        if (candidateHosts.isEmpty()) {
+          throw new AmbariException(MessageFormat.format("The resource filter for hosts does not contain components for " +
+                  "service: {0}, component: {1} in cluster: {2}", serviceName, componentName, clusterName));
+        }
       } else {
         candidateHosts = serviceHostComponents.keySet();
       }
     } else {
-      // TODO: this code branch looks unreliable(taking random component)
-      Map<String, ServiceComponent> serviceComponents =
-              cluster.getService(serviceName).getServiceComponents();
+      // TODO: This code branch looks unreliable (taking random component, should prefer the clients)
+      Map<String, ServiceComponent> serviceComponents = cluster.getService(serviceName).getServiceComponents();
 
       // Filter components without any HOST
       Iterator<String> serviceComponentNameIterator = serviceComponents.keySet().iterator();
@@ -536,18 +547,17 @@ public class AmbariCustomCommandExecutionHelper {
       }
 
       if (serviceComponents.isEmpty()) {
-        throw new AmbariException("Components not found, service = "
-            + serviceName + ", cluster = " + clusterName);
+        throw new AmbariException(MessageFormat.format("Did not find any hosts with components for service: {0} in cluster: {1}",
+            serviceName, clusterName));
       }
 
+      // Pick a random service (should prefer clients).
       ServiceComponent serviceComponent = serviceComponents.values().iterator().next();
-
       serviceHostComponents = serviceComponent.getServiceComponentHosts();
       candidateHosts = serviceHostComponents.keySet();
     }
 
-    // filter out hosts that are in maintenance mode - they should never be
-    // included in service checks
+    // Filter out hosts that are in maintenance mode - they should never be included in service checks
     Set<String> hostsInMaintenanceMode = new HashSet<String>();
     if (actionExecutionContext.isMaintenanceModeHostExcluded()) {
       Iterator<String> iterator = candidateHosts.iterator();
@@ -562,10 +572,10 @@ public class AmbariCustomCommandExecutionHelper {
       }
     }
 
-    // pick a random healthy host from the remaining set, throwing an exception
-    // if there are none to choose from
-    String hostName = managementController.getHealthyHost(candidateHosts);
-    if (hostName == null) {
+    // Filter out hosts that are not healthy, i.e., all hosts should be heartbeating.
+    // Pick one randomly. If there are none, throw an exception.
+    List<String> healthyHostNames = managementController.selectHealthyHosts(candidateHosts);
+    if (healthyHostNames.isEmpty()) {
       String message = MessageFormat.format(
           "While building a service check command for {0}, there were no healthy eligible hosts: unhealthy[{1}], maintenance[{2}]",
           serviceName, StringUtils.join(candidateHosts, ','),
@@ -574,11 +584,55 @@ public class AmbariCustomCommandExecutionHelper {
       throw new AmbariException(message);
     }
 
-    addServiceCheckAction(stage, hostName, smokeTestRole, nowTimestamp, serviceName, componentName,
+    String preferredHostName = selectRandomHostNameWithPreferenceOnAvailability(healthyHostNames);
+
+    long nowTimestamp = System.currentTimeMillis();
+    Map<String, String> actionParameters = actionExecutionContext.getParameters();
+    addServiceCheckAction(stage, preferredHostName, smokeTestRole, nowTimestamp, serviceName, componentName,
         actionParameters, actionExecutionContext.isRetryAllowed(),
         actionExecutionContext.isFailureAutoSkipped());
   }
 
+  /**
+   * Assuming all hosts are healthy and not in maintenance mode. Rank the hosts based on availability.
+   * Let S = all hosts with 0 PENDING/RUNNING/QUEUED/IN-PROGRESS tasks
+   * Let S' be all such other hosts.
+   *
+   * If S is non-empty, pick a random host from it. If S is empty and S' is non-empty, pick a random host from S'.
+   * @param candidateHostNames All possible host names
+   * @return Random host with a preference for those that are available to process commands immediately.
+   */
+  private String selectRandomHostNameWithPreferenceOnAvailability(List<String> candidateHostNames) throws AmbariException {
+    if (null == candidateHostNames || candidateHostNames.isEmpty()) {
+      return null;
+    }
+    if (candidateHostNames.size() == 1) {
+      return candidateHostNames.get(0);
+    }
+
+    List<String> hostsWithZeroCommands = new ArrayList<>();
+    List<String> hostsWithInProgressCommands = new ArrayList<>();
+
+    Map<Long, Integer> hostIdToCount = hostRoleCommandDAO.getHostIdToCountOfCommandsWithStatus(HostRoleStatus.IN_PROGRESS_STATUSES);
+    for (String hostName : candidateHostNames) {
+      Host host = clusters.getHost(hostName);
+
+      if (hostIdToCount.containsKey(host.getHostId()) && hostIdToCount.get(host.getHostId()) > 0) {
+        hostsWithInProgressCommands.add(hostName);
+      } else {
+        hostsWithZeroCommands.add(hostName);
+      }
+    }
+
+    List<String> preferredList = !hostsWithZeroCommands.isEmpty() ? hostsWithZeroCommands : hostsWithInProgressCommands;
+    if (!preferredList.isEmpty()) {
+      int randomIndex = new Random().nextInt(preferredList.size());
+      return preferredList.get(randomIndex);
+    }
+
+    return null;
+  }
+
   /**
    * Creates and populates service check EXECUTION_COMMAND for host. Not all
    * EXECUTION_COMMAND parameters are populated here because they are not needed

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

@@ -692,6 +692,17 @@ public interface AmbariManagementController {
    */
   String getMysqljdbcUrl();
 
+  /**
+   * Filters hosts to only select healthy ones that are heartbeating.
+   * <p/>
+   * The host's {@link HostState} is used to determine if a host is healthy.
+   *
+   * @return a List of healthy hosts, or an empty List if none exist.
+   * @throws AmbariException
+   * @see {@link HostState#HEALTHY}
+   */
+  List<String> selectHealthyHosts(Set<String> hostList) throws AmbariException;
+
   /**
    * Chooses a healthy host from the list of candidate hosts randomly. If there
    * are no healthy hosts, then this method will return {@code null}.

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

@@ -3761,18 +3761,49 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
             + cluster.getClusterName() + ", service=" + service.getName());
   }
 
-
+  /**
+   * Filters hosts to only select healthy ones that are heartbeating.
+   * <p/>
+   * The host's {@link HostState} is used to determine if a host is healthy.
+   *
+   * @return a List of healthy hosts, or an empty List if none exist.
+   * @throws AmbariException
+   * @see {@link HostState#HEALTHY}
+   */
   @Override
-  public String getHealthyHost(Set<String> hostList) throws AmbariException {
-    String hostName = null;
+  public List<String> selectHealthyHosts(Set<String> hostList) throws AmbariException {
+    List<String> healthyHosts = new ArrayList();
+
     for (String candidateHostName : hostList) {
-      hostName = candidateHostName;
-      Host candidateHost = clusters.getHost(hostName);
+      Host candidateHost = clusters.getHost(candidateHostName);
       if (candidateHost.getState() == HostState.HEALTHY) {
-        break;
+        healthyHosts.add(candidateHostName);
       }
     }
-    return hostName;
+
+    return healthyHosts;
+  }
+
+  /**
+   * Chooses a healthy host from the list of candidate hosts randomly. If there
+   * are no healthy hosts, then this method will return {@code null}.
+   * <p/>
+   * The host's {@link HostState} is used to determine if a host is healthy.
+   *
+   * @return a random healthy host, or {@code null}.
+   * @throws AmbariException
+   * @see {@link HostState#HEALTHY}
+   */
+  @Override
+  public String getHealthyHost(Set<String> hostList) throws AmbariException {
+    List<String> healthyHosts = selectHealthyHosts(hostList);
+
+    if (!healthyHosts.isEmpty()) {
+      Collections.shuffle(healthyHosts);
+      return healthyHosts.get(0);
+    }
+
+    return null;
   }
 
   @Override

ambari-server/src/main/java/org/apache/ambari/server/orm/dao/HostRoleCommandDAO.java

@@ -434,6 +434,26 @@ public class HostRoleCommandDAO {
     return daoUtils.selectList(query, requestId, stageId);
   }
 
+  @RequiresSession
+  public Map<Long, Integer> getHostIdToCountOfCommandsWithStatus(Collection<HostRoleStatus> statuses) {
+    Map<Long, Integer> hostIdToCount = new HashMap<>();
+
+    String queryName = "SELECT command.hostId FROM HostRoleCommandEntity command WHERE command.status IN :statuses";
+    TypedQuery<Long> query = entityManagerProvider.get().createQuery(queryName, Long.class);
+    query.setParameter("statuses", statuses);
+    List<Long> results = query.getResultList();
+
+    for (Long hostId : results) {
+      if (hostIdToCount.containsKey(hostId)) {
+        hostIdToCount.put(hostId, hostIdToCount.get(hostId) + 1);
+      } else {
+        hostIdToCount.put(hostId, 1);
+      }
+    }
+
+    return hostIdToCount;
+  }
+
   @RequiresSession
   public List<HostRoleCommandEntity> findByHostRole(String hostName, long requestId, long stageId, String role) {
 

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

@@ -47,11 +47,14 @@ import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 import org.apache.ambari.server.security.TestAuthenticationFactory;
 import org.apache.ambari.server.security.authorization.AuthorizationException;
+import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.SecurityType;
+import org.apache.ambari.server.state.Service;
+import org.apache.ambari.server.state.ServiceComponent;
 import org.apache.ambari.server.state.ServiceInfo;
 import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.State;
@@ -389,6 +392,45 @@ public class AmbariCustomCommandExecutionHelperTest {
     Assert.fail("Expected an exception since there are no hosts which can run the Flume service check");
   }
 
+  /**
+   * Perform a Service Check for ZOOKEEPER/ZOOKEEPER_CLIENT without specifying a host to run in the request.
+   * This should cause Ambari to randomly pick one of the ZOOKEEPER_CLIENT hosts.
+   * The current logic first excludes hosts in maintenance mode or that are not healthy (i.e., not heartbeating).
+   * From that candidate list, if any hosts have 0 IN-PROGRESS tasks, it randomly picks from that set.
+   * Otherwise, it picks from all candidate hosts.
+   * @throws Exception
+   */
+  @Test
+  public void testServiceCheckPicksRandomHost() throws Exception {
+    AmbariCustomCommandExecutionHelper ambariCustomCommandExecutionHelper = injector.getInstance(AmbariCustomCommandExecutionHelper.class);
+
+    Cluster c1 = clusters.getCluster("c1");
+    Service s = c1.getService("ZOOKEEPER");
+    ServiceComponent sc = s.getServiceComponent("ZOOKEEPER_CLIENT");
+    Assert.assertTrue(sc.getServiceComponentHosts().keySet().size() > 1);
+
+    // There are multiple hosts with ZK Client.
+    List<RequestResourceFilter> requestResourceFilter = new ArrayList<RequestResourceFilter>() {{
+      add(new RequestResourceFilter("ZOOKEEPER", "ZOOKEEPER_CLIENT", null));
+    }};
+    ActionExecutionContext actionExecutionContext = new ActionExecutionContext("c1", "SERVICE_CHECK", requestResourceFilter);
+    Stage stage = EasyMock.niceMock(Stage.class);
+    ExecutionCommandWrapper execCmdWrapper = EasyMock.niceMock(ExecutionCommandWrapper.class);
+    ExecutionCommand execCmd = EasyMock.niceMock(ExecutionCommand.class);
+
+    EasyMock.expect(stage.getClusterName()).andReturn("c1");
+    //
+    EasyMock.expect(stage.getExecutionCommandWrapper(EasyMock.eq("c1-c6401"), EasyMock.anyString())).andReturn(execCmdWrapper);
+    EasyMock.expect(stage.getExecutionCommandWrapper(EasyMock.eq("c1-c6402"), EasyMock.anyString())).andReturn(execCmdWrapper);
+    EasyMock.expect(execCmdWrapper.getExecutionCommand()).andReturn(execCmd);
+    EasyMock.expect(execCmd.getForceRefreshConfigTagsBeforeExecution()).andReturn(true);
+
+    HashSet<String> localComponents = new HashSet<>();
+    EasyMock.expect(execCmd.getLocalComponents()).andReturn(localComponents).anyTimes();
+    EasyMock.replay(stage, execCmdWrapper, execCmd);
+
+    ambariCustomCommandExecutionHelper.addExecutionCommandsToStage(actionExecutionContext, stage, new HashMap<String, String>());
+  }
 
   @Test
   public void testIsTopologyRefreshRequired() throws Exception {
@@ -460,11 +502,11 @@ public class AmbariCustomCommandExecutionHelperTest {
     createServiceComponentHost(clusterName, "YARN", "NODEMANAGER", hostC6401, null);
     createServiceComponentHost(clusterName, "GANGLIA", "GANGLIA_SERVER", hostC6401, State.INIT);
     createServiceComponentHost(clusterName, "GANGLIA", "GANGLIA_MONITOR", hostC6401, State.INIT);
+    createServiceComponentHost(clusterName, "ZOOKEEPER", "ZOOKEEPER_CLIENT", hostC6401, State.INIT);
 
-    createServiceComponentHost(clusterName, "YARN", "NODEMANAGER", hostPrefix + "-c6402", null);
-    createServiceComponentHost(clusterName, "GANGLIA", "GANGLIA_MONITOR", hostPrefix + "-c6402", State.INIT);
-    createServiceComponentHost(clusterName, "ZOOKEEPER", "ZOOKEEPER_CLIENT", hostPrefix + "-c6402", State.INIT);
-
+    createServiceComponentHost(clusterName, "YARN", "NODEMANAGER", hostC6402, null);
+    createServiceComponentHost(clusterName, "GANGLIA", "GANGLIA_MONITOR", hostC6402, State.INIT);
+    createServiceComponentHost(clusterName, "ZOOKEEPER", "ZOOKEEPER_CLIENT", hostC6402, State.INIT);
   }
   private void addHost(String hostname, String clusterName) throws AmbariException {
     clusters.addHost(hostname);

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

@@ -6126,22 +6126,15 @@ public class AmbariManagementControllerTest {
     }
     Assert.assertEquals("Expect only one service check.", 1, commandCount);
 
-    // When both are unhealthy then just pick one
+    // When both are unhealthy then it should raise an exception.
     clusters.getHost(host3).setState(HostState.WAITING_FOR_HOST_STATUS_UPDATES);
     clusters.getHost(host2).setState(HostState.INIT);
-    response = controller.createAction(actionRequest, requestProperties);
-    commands = actionDB.getRequestTasks(response.getRequestId());
-    commandCount = 0;
-    for(HostRoleCommand command : commands) {
-      if(command.getRoleCommand() == RoleCommand.SERVICE_CHECK &&
-          command.getRole() == Role.HDFS_SERVICE_CHECK) {
-        Assert.assertTrue(command.getHostName().equals(host3) ||
-            command.getHostName().equals(host2));
-        commandCount++;
-      }
+    try {
+      response = controller.createAction(actionRequest, requestProperties);
+      assertTrue("Exception should have been raised.", false);
+    } catch (AmbariException e) {
+      assertTrue(e.getMessage().contains("there were no healthy eligible hosts"));
     }
-    Assert.assertEquals("Expect only one service check.", 1, commandCount);
-    Assert.assertEquals("", response.getRequestContext());
   }
 
   @Test
@@ -9402,11 +9395,13 @@ public class AmbariManagementControllerTest {
       clusters.addHost(HOST1);
       Host host = clusters.getHost(HOST1);
       setOsFamily(host, "redhat", "6.3");
+      clusters.getHost(HOST1).setState(HostState.HEALTHY);
       host.persist();
 
       clusters.addHost(HOST2);
       host = clusters.getHost(HOST2);
       setOsFamily(host, "redhat", "6.3");
+      clusters.getHost(HOST1).setState(HostState.HEALTHY);
       host.persist();
 
       AmbariManagementController amc = injector.getInstance(AmbariManagementController.class);

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

@@ -37,6 +37,7 @@ import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.utilities.PredicateBuilder;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
 import org.apache.ambari.server.orm.DBAccessor;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
 import org.apache.ambari.server.orm.dao.UserDAO;
 import org.apache.ambari.server.orm.dao.WidgetDAO;
 import org.apache.ambari.server.orm.dao.WidgetLayoutDAO;
@@ -394,6 +395,7 @@ public class ActiveWidgetLayoutResourceProviderTest extends EasyMockSupport {
         bind(CredentialStoreService.class).to(CredentialStoreServiceImpl.class);
         bind(UserDAO.class).toInstance(createMock(UserDAO.class));
         bind(WidgetLayoutDAO.class).toInstance(createMock(WidgetLayoutDAO.class));
+        bind(HostRoleCommandDAO.class).toInstance(createMock(HostRoleCommandDAO.class));
       }
     });
   }

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

@@ -38,6 +38,7 @@ import org.apache.ambari.server.controller.spi.SystemException;
 import org.apache.ambari.server.controller.utilities.PredicateBuilder;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
 import org.apache.ambari.server.orm.DBAccessor;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
 import org.apache.ambari.server.orm.dao.PermissionDAO;
 import org.apache.ambari.server.orm.dao.ResourceTypeDAO;
 import org.apache.ambari.server.orm.dao.UserDAO;
@@ -405,6 +406,7 @@ public class UserAuthorizationResourceProviderTest extends EasyMockSupport {
         bind(UserDAO.class).toInstance(createMock(UserDAO.class));
         bind(ResourceTypeDAO.class).toInstance(createMock(ResourceTypeDAO.class));
         bind(PermissionDAO.class).toInstance(createMock(PermissionDAO.class));
+        bind(HostRoleCommandDAO.class).toInstance(createMock(HostRoleCommandDAO.class));
       }
     });
   }

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

@@ -37,6 +37,7 @@ import org.apache.ambari.server.controller.spi.ResourceProvider;
 import org.apache.ambari.server.controller.utilities.PredicateBuilder;
 import org.apache.ambari.server.controller.utilities.PropertyHelper;
 import org.apache.ambari.server.orm.DBAccessor;
+import org.apache.ambari.server.orm.dao.HostRoleCommandDAO;
 import org.apache.ambari.server.scheduler.ExecutionScheduler;
 import org.apache.ambari.server.security.TestAuthenticationFactory;
 import org.apache.ambari.server.security.authorization.AuthorizationException;
@@ -240,6 +241,7 @@ public class UserResourceProviderTest extends EasyMockSupport {
         bind(Users.class).toInstance(createMock(Users.class));
         bind(AmbariManagementController.class).to(AmbariManagementControllerImpl.class);
         bind(CredentialStoreService.class).to(CredentialStoreServiceImpl.class);
+        bind(HostRoleCommandDAO.class).toInstance(createMock(HostRoleCommandDAO.class));
       }
     });
   }