AMBARI-4161. Move calculation of host status from HostResourceProvider to HeartBeatHandler.(odiachenko)

ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java

@@ -17,6 +17,7 @@
  */
 package org.apache.ambari.server.agent;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -35,7 +36,10 @@ import org.apache.ambari.server.metadata.ActionMetadata;
 import org.apache.ambari.server.state.AgentVersion;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
+import org.apache.ambari.server.state.ComponentInfo;
 import org.apache.ambari.server.state.Host;
+import org.apache.ambari.server.state.HostHealthStatus;
+import org.apache.ambari.server.state.HostHealthStatus.HealthStatus;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.ServiceComponent;
@@ -119,7 +123,6 @@ public class HeartBeatHandler {
       LOG.error("CurrentResponseId unknown for " + hostname + " - send register command");
       return createRegisterCommand();
     }
-
     LOG.debug("Received heartbeat from host"
         + ", hostname=" + hostname
         + ", currentResponseId=" + currentResponseId
@@ -178,15 +181,97 @@ public class HeartBeatHandler {
 
     // Examine heartbeart for component live status reports
     processStatusReports(heartbeat, hostname, clusterFsm);
+    
+    //Calculate host status
+    processHostStatus(heartbeat, hostname);
 
     // Send commands if node is active
     if (hostObject.getState().equals(HostState.HEALTHY)) {
       sendCommands(hostname, response);
       annotateResponse(hostname, response);
-    }
+    }    
     return response;
   }
 
+  protected void processHostStatus(HeartBeat heartbeat, String hostname) throws AmbariException {
+
+    Host host = clusterFsm.getHost(hostname);
+
+    
+    HealthStatus healthStatus = host.getHealthStatus().getHealthStatus();
+    
+    if (!healthStatus.equals(HostHealthStatus.HealthStatus.UNKNOWN)) {
+
+      List<ComponentStatus> componentStatuses = heartbeat.getComponentStatus();
+      //Host status info could be calculated only if agent returned statuses in heartbeat
+      if (componentStatuses.size() > 0) {
+
+        int masterCount = 0;
+        int mastersRunning = 0;
+        int slaveCount = 0;
+        int slavesRunning = 0;
+
+        Map<String, StackId> stackIdsByClusters =
+            new HashMap<String, StackId>();
+
+        for (ComponentStatus componentStatus : componentStatuses) {
+
+          String clusterName = componentStatus.getClusterName();
+
+          StackId stackId;
+          if (stackIdsByClusters.containsKey(clusterName)) {
+            stackId = stackIdsByClusters.get(clusterName);
+
+          } else {
+            Cluster cluster = clusterFsm.getCluster(clusterName);
+            stackId = cluster.getDesiredStackVersion();
+            stackIdsByClusters.put(clusterName, stackId);
+          }
+
+          ComponentInfo componentInfo =
+              ambariMetaInfo.getComponent(stackId.getStackName(),
+                  stackId.getStackVersion(), componentStatus.getServiceName(),
+                  componentStatus.getComponentName());
+
+          String status = componentStatus.getStatus();
+
+          String category = componentInfo.getCategory();
+
+          if (category.equals("MASTER")) {
+            ++masterCount;
+            if (status.equals("STARTED")) {
+              ++mastersRunning;
+            }
+          } else if (category.equals("SLAVE")) {
+            ++slaveCount;
+            if (status.equals("STARTED")) {
+              ++slavesRunning;
+            }
+          }
+        }
+        
+        if (masterCount == mastersRunning && slaveCount == slavesRunning) {
+          healthStatus = HostHealthStatus.HealthStatus.HEALTHY;
+        } else if (masterCount > 0 && mastersRunning < masterCount) {
+          healthStatus = HostHealthStatus.HealthStatus.UNHEALTHY;
+        } else {
+          healthStatus = HostHealthStatus.HealthStatus.ALERT;
+        }
+        
+        host.setStatus(healthStatus.name());
+        host.persist();
+      }
+      
+      //If host doesn't belongs to any cluster
+      if ((clusterFsm.getClustersForHost(host.getHostName())).size() == 0) {
+        healthStatus = HostHealthStatus.HealthStatus.HEALTHY;
+        host.setStatus(healthStatus.name());
+        host.persist();
+      } 
+      
+    }
+  }
+
   protected void processCommandReports(
       HeartBeat heartbeat, String hostname, Clusters clusterFsm, long now)
       throws AmbariException {

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

@@ -139,13 +139,18 @@ public class HostResponse {
    */
   private Map<String, HostConfig> desiredHostConfigs;
 
+  /**
+   * Host status, calculated on host components statuses
+   */
+  private String status;
+
   public HostResponse(String hostname, String clusterName,
                       String ipv4, String ipv6, int cpuCount, int phCpuCount, String osArch, String osType,
                       String osInfo, long availableMemBytes, long totalMemBytes,
                       List<DiskInfo> disksInfo, long lastHeartbeatTime,
                       long lastRegistrationTime, String rackInfo,
                       Map<String, String> hostAttributes, AgentVersion agentVersion,
-                      HostHealthStatus healthStatus, String hostState) {
+                      HostHealthStatus healthStatus, String hostState, String status) {
     super();
     this.hostname = hostname;
     this.clusterName = clusterName;
@@ -166,6 +171,7 @@ public class HostResponse {
     this.agentVersion = agentVersion;
     this.healthStatus = healthStatus;
     this.setHostState(hostState);
+    this.status = status;
   }
 
   //todo: why are we passing in empty strings for host/cluster name instead of null?
@@ -175,7 +181,7 @@ public class HostResponse {
         "", 0, 0, new ArrayList<DiskInfo>(),
         0, 0, "",
         new HashMap<String, String>(),
-        null, null, null);
+        null, null, null, null);
   }
 
   /**
@@ -505,4 +511,12 @@ public class HostResponse {
   public void setDesiredHostConfigs(Map<String, HostConfig> desiredHostConfigs) {
     this.desiredHostConfigs = desiredHostConfigs;
   }
+
+  public String getStatus() {
+    return status;
+  }
+
+  public void setStatus(String status) {
+    this.status = status;
+  }
 }

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

@@ -181,13 +181,6 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
     for (HostResponse response : responses) {
       Resource resource = new ResourceImpl(Resource.Type.Host);
 
-      String hostStatus;
-      try {
-        hostStatus = calculateHostStatus(response);
-      } catch (AmbariException e) {
-        throw new SystemException("", e);
-      }
-
       // TODO : properly handle more than one cluster
       if (response.getClusterName() != null
           && !response.getClusterName().isEmpty()) {
@@ -219,7 +212,7 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
       setResourceProperty(resource, HOST_LAST_REGISTRATION_TIME_PROPERTY_ID,
           response.getLastRegistrationTime(), requestedIds);
       setResourceProperty(resource, HOST_HOST_STATUS_PROPERTY_ID,
-          hostStatus,requestedIds);
+          response.getStatus(),requestedIds);
       setResourceProperty(resource, HOST_HOST_HEALTH_REPORT_PROPERTY_ID,
           response.getHealthStatus().getHealthReport(), requestedIds);
       setResourceProperty(resource, HOST_DISK_INFO_PROPERTY_ID,
@@ -443,7 +436,7 @@ class HostResourceProvider extends AbstractControllerResourceProvider {
     return response;
   }
 
-  protected static synchronized Set<HostResponse> getHosts(AmbariManagementController controller, HostRequest request)
+  protected static Set<HostResponse> getHosts(AmbariManagementController controller, HostRequest request)
       throws AmbariException {
 
     //TODO/FIXME host can only belong to a single cluster so get host directly from Cluster

ambari-server/src/main/java/org/apache/ambari/server/state/Host.java

@@ -282,6 +282,17 @@ public interface Host {
    * @param timeInState the timeInState to set
    */
   public void setTimeInState(long timeInState);
+  
+  /**
+   * Get Current Host Status
+   * @return String
+   */
+  public String getStatus();
+  /**
+   * Set the Status of the Host
+   * @param status Host Status
+   */
+  public void setStatus(String status);
 
   public HostResponse convertToResponse();
 

ambari-server/src/main/java/org/apache/ambari/server/state/host/HostImpl.java

@@ -37,7 +37,6 @@ import org.apache.ambari.server.orm.dao.HostConfigMappingDAO;
 import org.apache.ambari.server.orm.dao.HostDAO;
 import org.apache.ambari.server.orm.dao.HostStateDAO;
 import org.apache.ambari.server.orm.entities.ClusterEntity;
-import org.apache.ambari.server.orm.entities.ConfigGroupHostMappingEntity;
 import org.apache.ambari.server.orm.entities.HostConfigMappingEntity;
 import org.apache.ambari.server.orm.entities.HostEntity;
 import org.apache.ambari.server.orm.entities.HostStateEntity;
@@ -115,6 +114,10 @@ public class HostImpl implements Host {
   private static final String SWAPSIZE = "swap_size";
   private static final String SWAPFREE = "swap_free";
   private static final String TIMEZONE = "timezone";
+  
+  
+  //In-memory status, based on host components states
+  private String status;
 
   private static final StateMachineFactory
     <HostImpl, HostState, HostEventType, HostEvent>
@@ -853,6 +856,10 @@ public class HostImpl implements Host {
     try {
       writeLock.lock();
       hostStateEntity.setHealthStatus(gson.toJson(healthStatus));
+      
+      if (healthStatus.getHealthStatus().equals(HealthStatus.UNKNOWN))
+        setStatus(HealthStatus.UNKNOWN.name());
+      
       saveIfPersisted();
     } finally {
       writeLock.unlock();
@@ -992,6 +999,23 @@ public class HostImpl implements Host {
       writeLock.unlock();
     }
   }
+  
+  
+  @Override
+  public String getStatus() {
+    return status;
+  }
+
+  @Override
+  public void setStatus(String status) {
+    try {
+      writeLock.lock();
+      this.status = status;
+    }
+    finally {
+      writeLock.unlock();
+    }
+  }
 
   @Override
   public HostResponse convertToResponse() {
@@ -1018,6 +1042,7 @@ public class HostImpl implements Host {
       r.setTotalMemBytes(getTotalMemBytes());
       r.setPublicHostName(getPublicHostName());
       r.setHostState(getState().toString());
+      r.setStatus(getStatus());
 
       return r;
     }

ambari-server/src/test/java/org/apache/ambari/server/agent/TestHeartbeatHandler.java

@@ -73,6 +73,7 @@ import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
 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.HostHealthStatus;
 import org.apache.ambari.server.state.HostState;
 import org.apache.ambari.server.state.Service;
 import org.apache.ambari.server.state.ServiceComponentHost;
@@ -1307,6 +1308,101 @@ public class TestHeartbeatHandler {
     assertEquals("Stack version of SCH should not change after fail report",
             State.INSTALL_FAILED, serviceComponentHost2.getState());
   }
+  
+  @Test
+  public void testProcessStatusReports() throws Exception {
+    ActionManager am = getMockActionManager();
+    Clusters fsm = clusters;
+
+    Cluster cluster = getDummyCluster();
+    Host hostObject = clusters.getHost(DummyHostname1);
+    clusters.mapHostToCluster(hostObject.getHostName(), cluster.getClusterName());
+
+    ActionQueue aq = new ActionQueue();
+
+    HeartBeatHandler handler = new HeartBeatHandler(fsm, aq, am, injector);
+    Register reg = new Register();
+    HostInfo hi = new HostInfo();
+    hi.setHostName(DummyHostname1);
+    hi.setOS(DummyOs);
+    hi.setOSRelease(DummyOSRelease);
+    reg.setHostname(DummyHostname1);
+    reg.setHardwareProfile(hi);
+    reg.setAgentVersion(metaInfo.getServerVersion());
+    handler.handleRegistration(reg);
+
+    hostObject.setState(HostState.UNHEALTHY);
+
+    aq.enqueue(DummyHostname1, new StatusCommand());
+    
+    //All components are up
+    HeartBeat hb1 = new HeartBeat();
+    hb1.setResponseId(0);
+    hb1.setNodeStatus(new HostStatus(Status.HEALTHY, DummyHostStatus));
+    hb1.setHostname(DummyHostname1);
+    List<ComponentStatus> componentStatus = new ArrayList<ComponentStatus>();
+    ComponentStatus dataNodeStatus = new ComponentStatus();
+    dataNodeStatus.setClusterName(cluster.getClusterName());
+    dataNodeStatus.setServiceName(HDFS);
+    dataNodeStatus.setComponentName(DATANODE);
+    dataNodeStatus.setStatus("STARTED");
+    componentStatus.add(dataNodeStatus);
+    ComponentStatus nameNodeStatus = new ComponentStatus();
+    nameNodeStatus.setClusterName(cluster.getClusterName());
+    nameNodeStatus.setServiceName(HDFS);
+    nameNodeStatus.setComponentName(NAMENODE);
+    nameNodeStatus.setStatus("STARTED");
+    componentStatus.add(nameNodeStatus);
+    hb1.setComponentStatus(componentStatus);
+    handler.handleHeartBeat(hb1);
+    assertEquals(HostHealthStatus.HealthStatus.HEALTHY.name(), hostObject.getStatus());
+    
+    //Some slaves are down, masters are up
+    HeartBeat hb2 = new HeartBeat();
+    hb2.setResponseId(1);
+    hb2.setNodeStatus(new HostStatus(Status.HEALTHY, DummyHostStatus));
+    hb2.setHostname(DummyHostname1);
+    componentStatus = new ArrayList<ComponentStatus>();
+    dataNodeStatus = new ComponentStatus();
+    dataNodeStatus.setClusterName(cluster.getClusterName());
+    dataNodeStatus.setServiceName(HDFS);
+    dataNodeStatus.setComponentName(DATANODE);
+    dataNodeStatus.setStatus("INSTALLED");
+    componentStatus.add(dataNodeStatus);
+    nameNodeStatus = new ComponentStatus();
+    nameNodeStatus.setClusterName(cluster.getClusterName());
+    nameNodeStatus.setServiceName(HDFS);
+    nameNodeStatus.setComponentName(NAMENODE);
+    nameNodeStatus.setStatus("STARTED");
+    componentStatus.add(nameNodeStatus);
+    hb2.setComponentStatus(componentStatus);
+    handler.handleHeartBeat(hb2);
+    assertEquals(HostHealthStatus.HealthStatus.ALERT.name(), hostObject.getStatus());
+    
+    //Some masters are down
+    HeartBeat hb3 = new HeartBeat();
+    hb3.setResponseId(2);
+    hb3.setNodeStatus(new HostStatus(Status.HEALTHY, DummyHostStatus));
+    hb3.setHostname(DummyHostname1);
+    componentStatus = new ArrayList<ComponentStatus>();
+    dataNodeStatus = new ComponentStatus();
+    dataNodeStatus.setClusterName(cluster.getClusterName());
+    dataNodeStatus.setServiceName(HDFS);
+    dataNodeStatus.setComponentName(DATANODE);
+    dataNodeStatus.setStatus("INSTALLED");
+    componentStatus.add(dataNodeStatus);
+    nameNodeStatus = new ComponentStatus();
+    nameNodeStatus.setClusterName(cluster.getClusterName());
+    nameNodeStatus.setServiceName(HDFS);
+    nameNodeStatus.setComponentName(NAMENODE);
+    nameNodeStatus.setStatus("INSTALLED");
+    componentStatus.add(nameNodeStatus);
+    hb3.setComponentStatus(componentStatus);
+    handler.handleHeartBeat(hb3);
+    assertEquals(HostHealthStatus.HealthStatus.UNHEALTHY.name(), hostObject.getStatus());
+    
+    
+  }
 
   @Test
   public void testIgnoreCustomActionReport() throws AmbariException, InvalidStateTransitionException {

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

@@ -41,6 +41,7 @@ import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.ComponentInfo;
 import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.HostHealthStatus;
+import org.apache.ambari.server.state.HostHealthStatus.HealthStatus;
 import org.apache.ambari.server.state.StackId;
 import org.easymock.Capture;
 import org.junit.Assert;
@@ -253,6 +254,7 @@ public class HostResourceProviderTest {
     expect(hostResponse1.getClusterName()).andReturn("").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
+    expect(hostResponse1.getStatus()).andReturn(HealthStatus.HEALTHY.name()).anyTimes();
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
@@ -344,8 +346,6 @@ public class HostResourceProviderTest {
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(cluster.getDesiredStackVersion()).andReturn(stackId);
-
     expect(host1.getHostName()).andReturn("Host100").anyTimes();
 
     expect(host1.convertToResponse()).andReturn(hostResponse1);
@@ -353,6 +353,7 @@ public class HostResourceProviderTest {
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
+    expect(hostResponse1.getStatus()).andReturn(HealthStatus.HEALTHY.name()).anyTimes();
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
@@ -444,8 +445,6 @@ public class HostResourceProviderTest {
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(cluster.getDesiredStackVersion()).andReturn(stackId);
-
     expect(host1.getHostName()).andReturn("Host100").anyTimes();
 
     expect(host1.convertToResponse()).andReturn(hostResponse1);
@@ -453,6 +452,7 @@ public class HostResourceProviderTest {
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
+    expect(hostResponse1.getStatus()).andReturn(HealthStatus.UNHEALTHY.name()).anyTimes();
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();
@@ -539,6 +539,7 @@ public class HostResourceProviderTest {
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
+    expect(hostResponse1.getStatus()).andReturn(HealthStatus.UNKNOWN.name()).anyTimes();
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.UNKNOWN).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("UNKNOWN").anyTimes();
@@ -624,8 +625,6 @@ public class HostResourceProviderTest {
 
     expect(clusters.getClustersForHost("Host100")).andReturn(clusterSet).anyTimes();
 
-    expect(cluster.getDesiredStackVersion()).andReturn(stackId);
-
     expect(host1.getHostName()).andReturn("Host100").anyTimes();
 
     expect(host1.convertToResponse()).andReturn(hostResponse1);
@@ -633,6 +632,7 @@ public class HostResourceProviderTest {
     expect(hostResponse1.getClusterName()).andReturn("Cluster100").anyTimes();
     expect(hostResponse1.getHostname()).andReturn("Host100").anyTimes();
     expect(hostResponse1.getHealthStatus()).andReturn(healthStatus).anyTimes();
+    expect(hostResponse1.getStatus()).andReturn(HealthStatus.ALERT.name()).anyTimes();
 
     expect(healthStatus.getHealthStatus()).andReturn(HostHealthStatus.HealthStatus.HEALTHY).anyTimes();
     expect(healthStatus.getHealthReport()).andReturn("HEALTHY").anyTimes();