HeartbeatMonitor implements Runnable { public List<StatusCommand> generateStatusCommands(String hostname) throws AmbariException { List<StatusCommand> cmds = new ArrayList<StatusCommand>(); for (Cluster cl : clusters.getClustersForHost(hostname)) { Map<String, DesiredConfig> desiredConfigs = cl.getDesiredConfigs(); for (ServiceComponentHost sch : cl.getServiceComponentHosts(hostname)) { switch (sch.getState()) { case INIT: case INSTALLING: case STARTING: case STOPPING: continue; default: StatusCommand statusCmd = createStatusCommand(hostname, cl, sch, desiredConfigs); cmds.add(statusCmd); } } } return cmds; } HeartbeatMonitor(Clusters clusters, ActionQueue aq, ActionManager am,
                          int threadWakeupInterval, Injector injector); void shutdown(); void start(); boolean isAlive(); AgentRequests getAgentRequests(); @Override void run(); List<StatusCommand> generateStatusCommands(String hostname);  }
@Test public void testStateCommandsGeneration() throws AmbariException, InterruptedException, InvalidStateTransitionException { StackId stackId = new StackId("HDP-0.1"); Clusters clusters = injector.getInstance(Clusters.class); clusters.addHost(hostname1); setOsFamily(clusters.getHost(hostname1), "redhat", "6.3"); clusters.addHost(hostname2); setOsFamily(clusters.getHost(hostname2), "redhat", "6.3"); clusters.addCluster(clusterName, stackId); Cluster cluster = clusters.getCluster(clusterName); helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion()); cluster.createClusterVersion(stackId, stackId.getStackVersion(), "admin", RepositoryVersionState.INSTALLING); Set<String> hostNames = new HashSet<String>(){{ add(hostname1); add(hostname2); }}; ConfigFactory configFactory = injector.getInstance(ConfigFactory.class); Config config = configFactory.createNew(cluster, "hadoop-env", "version1", new HashMap<String,String>() {{ put("a", "b"); }}, new HashMap<String, Map<String,String>>()); cluster.addDesiredConfig("_test", Collections.singleton(config)); clusters.mapAndPublishHostsToCluster(hostNames, clusterName); Service hdfs = cluster.addService(serviceName); hdfs.addServiceComponent(Role.DATANODE.name()); hdfs.getServiceComponent(Role.DATANODE.name()).addServiceComponentHost(hostname1); hdfs.addServiceComponent(Role.NAMENODE.name()); hdfs.getServiceComponent(Role.NAMENODE.name()).addServiceComponentHost(hostname1); hdfs.addServiceComponent(Role.SECONDARY_NAMENODE.name()); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).addServiceComponentHost(hostname1); hdfs.getServiceComponent(Role.DATANODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); ActionQueue aq = new ActionQueue(); ActionManager am = mock(ActionManager.class); HeartbeatMonitor hm = new HeartbeatMonitor(clusters, aq, am, heartbeatMonitorWakeupIntervalMS, injector); HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector); Register reg = new Register(); reg.setHostname(hostname1); reg.setResponseId(12); reg.setTimestamp(System.currentTimeMillis() - 300); reg.setAgentVersion(ambariMetaInfo.getServerVersion()); HostInfo hi = new HostInfo(); hi.setOS("Centos5"); reg.setHardwareProfile(hi); handler.handleRegistration(reg); HeartBeat hb = new HeartBeat(); hb.setHostname(hostname1); hb.setNodeStatus(new HostStatus(HostStatus.Status.HEALTHY, "cool")); hb.setTimestamp(System.currentTimeMillis()); hb.setResponseId(12); handler.handleHeartBeat(hb); List<StatusCommand> cmds = hm.generateStatusCommands(hostname1); assertTrue("HeartbeatMonitor should generate StatusCommands for host1", cmds.size() == 3); assertEquals("HDFS", cmds.get(0).getServiceName()); boolean containsDATANODEStatus = false; boolean containsNAMENODEStatus = false; boolean containsSECONDARY_NAMENODEStatus = false; for (StatusCommand cmd : cmds) { containsDATANODEStatus |= cmd.getComponentName().equals("DATANODE"); containsNAMENODEStatus |= cmd.getComponentName().equals("NAMENODE"); containsSECONDARY_NAMENODEStatus |= cmd.getComponentName().equals("SECONDARY_NAMENODE"); assertTrue(cmd.getConfigurations().size() > 0); } assertEquals(true, containsDATANODEStatus); assertEquals(true, containsNAMENODEStatus); assertEquals(true, containsSECONDARY_NAMENODEStatus); cmds = hm.generateStatusCommands(hostname2); assertTrue("HeartbeatMonitor should not generate StatusCommands for host2 because it has no services", cmds.isEmpty()); }
@Test public void testStatusCommandForAnyComponents() throws Exception { StackId stackId = new StackId("HDP-0.1"); Clusters clusters = injector.getInstance(Clusters.class); clusters.addHost(hostname1); setOsFamily(clusters.getHost(hostname1), "redhat", "6.3"); clusters.addHost(hostname2); setOsFamily(clusters.getHost(hostname2), "redhat", "6.3"); clusters.addCluster(clusterName, stackId); Cluster cluster = clusters.getCluster(clusterName); helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion()); cluster.createClusterVersion(stackId, stackId.getStackVersion(), "admin", RepositoryVersionState.INSTALLING); Set<String> hostNames = new HashSet<String>() {{ add(hostname1); add(hostname2); }}; ConfigFactory configFactory = injector.getInstance(ConfigFactory.class); Config hadoopEnvConfig = configFactory.createNew(cluster, "hadoop-env", "version1", new HashMap<String, String>() {{ put("a", "b"); }}, new HashMap<String, Map<String,String>>()); Config hbaseEnvConfig = configFactory.createNew(cluster, "hbase-env", "version1", new HashMap<String, String>() {{ put("a", "b"); }}, new HashMap<String, Map<String,String>>()); cluster.addDesiredConfig("_test", Collections.singleton(hadoopEnvConfig)); clusters.mapAndPublishHostsToCluster(hostNames, clusterName); Service hdfs = cluster.addService(serviceName); hdfs.addServiceComponent(Role.DATANODE.name()); hdfs.getServiceComponent(Role.DATANODE.name()).addServiceComponentHost (hostname1); hdfs.addServiceComponent(Role.NAMENODE.name()); hdfs.getServiceComponent(Role.NAMENODE.name()).addServiceComponentHost (hostname1); hdfs.addServiceComponent(Role.SECONDARY_NAMENODE.name()); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()). addServiceComponentHost(hostname1); hdfs.addServiceComponent(Role.HDFS_CLIENT.name()); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).addServiceComponentHost (hostname1); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).addServiceComponentHost (hostname2); hdfs.getServiceComponent(Role.DATANODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).getServiceComponentHost(hostname2).setState(State.INSTALLED); hdfs.getServiceComponent(Role.DATANODE.name()).getServiceComponentHost(hostname1).setDesiredState(State.INSTALLED); hdfs.getServiceComponent(Role.NAMENODE.name()).getServiceComponentHost(hostname1).setDesiredState(State.INSTALLED); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).getServiceComponentHost(hostname1).setDesiredState(State.INSTALLED); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).getServiceComponentHost(hostname1).setDesiredState(State.INSTALLED); hdfs.getServiceComponent(Role.HDFS_CLIENT.name()).getServiceComponentHost(hostname2).setDesiredState(State.INSTALLED); ActionQueue aq = new ActionQueue(); ActionManager am = mock(ActionManager.class); HeartbeatMonitor hm = new HeartbeatMonitor(clusters, aq, am, heartbeatMonitorWakeupIntervalMS, injector); HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector); Register reg = new Register(); reg.setHostname(hostname1); reg.setResponseId(12); reg.setTimestamp(System.currentTimeMillis() - 300); reg.setAgentVersion(ambariMetaInfo.getServerVersion()); HostInfo hi = new HostInfo(); hi.setOS("Centos5"); reg.setHardwareProfile(hi); RegistrationResponse registrationResponse = handler.handleRegistration(reg); HeartBeat hb = new HeartBeat(); hb.setHostname(hostname1); hb.setNodeStatus(new HostStatus(HostStatus.Status.HEALTHY, "cool")); hb.setTimestamp(System.currentTimeMillis()); hb.setResponseId(12); handler.handleHeartBeat(hb); Map<String, Map<String, String>> statusCommandConfig = registrationResponse.getStatusCommands().get(0).getConfigurations(); assertEquals(statusCommandConfig.size(), 1); assertTrue(statusCommandConfig.containsKey("hadoop-env")); List<StatusCommand> cmds = hm.generateStatusCommands(hostname1); assertTrue("HeartbeatMonitor should generate StatusCommands for host1", cmds.size() == 4); assertEquals("HDFS", cmds.get(0).getServiceName()); boolean containsDATANODEStatus = false; boolean containsNAMENODEStatus = false; boolean containsSECONDARY_NAMENODEStatus = false; boolean containsHDFS_CLIENTStatus = false; for (StatusCommand cmd : cmds) { containsDATANODEStatus |= cmd.getComponentName().equals("DATANODE"); containsNAMENODEStatus |= cmd.getComponentName().equals("NAMENODE"); containsSECONDARY_NAMENODEStatus |= cmd.getComponentName(). equals("SECONDARY_NAMENODE"); containsHDFS_CLIENTStatus |= cmd.getComponentName().equals ("HDFS_CLIENT"); assertTrue(cmd.getConfigurations().size() > 0); assertEquals(State.INSTALLED, cmd.getDesiredState()); assertEquals(false, cmd.getHasStaleConfigs()); } assertTrue(containsDATANODEStatus); assertTrue(containsNAMENODEStatus); assertTrue(containsSECONDARY_NAMENODEStatus); assertTrue(containsHDFS_CLIENTStatus); cmds = hm.generateStatusCommands(hostname2); assertTrue("HeartbeatMonitor should generate StatusCommands for host2, " + "even if it has only client components", cmds.size() == 1); assertTrue(cmds.get(0).getComponentName().equals(Role.HDFS_CLIENT.name())); assertEquals(hostname2, cmds.get(0).getHostname()); }
@Test public void testStateCommandsWithAlertsGeneration() throws AmbariException, InterruptedException, InvalidStateTransitionException { StackId stackId = new StackId("HDP-2.0.7"); Clusters clusters = injector.getInstance(Clusters.class); clusters.addHost(hostname1); setOsFamily(clusters.getHost(hostname1), "redhat", "6.3"); clusters.addHost(hostname2); setOsFamily(clusters.getHost(hostname2), "redhat", "6.3"); clusters.addCluster(clusterName, stackId); Cluster cluster = clusters.getCluster(clusterName); cluster.setDesiredStackVersion(stackId); helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion()); cluster.createClusterVersion(stackId, stackId.getStackVersion(), "admin", RepositoryVersionState.INSTALLING); Set<String> hostNames = new HashSet<String>(){{ add(hostname1); add(hostname2); }}; clusters.mapAndPublishHostsToCluster(hostNames, clusterName); Service hdfs = cluster.addService(serviceName); hdfs.addServiceComponent(Role.DATANODE.name()); hdfs.getServiceComponent(Role.DATANODE.name()).addServiceComponentHost(hostname1); hdfs.addServiceComponent(Role.NAMENODE.name()); hdfs.getServiceComponent(Role.NAMENODE.name()).addServiceComponentHost(hostname1); hdfs.addServiceComponent(Role.SECONDARY_NAMENODE.name()); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).addServiceComponentHost(hostname1); hdfs.getServiceComponent(Role.DATANODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); hdfs.getServiceComponent(Role.SECONDARY_NAMENODE.name()).getServiceComponentHost(hostname1).setState(State.INSTALLED); ActionQueue aq = new ActionQueue(); ActionManager am = mock(ActionManager.class); HeartbeatMonitor hm = new HeartbeatMonitor(clusters, aq, am, heartbeatMonitorWakeupIntervalMS, injector); HeartBeatHandler handler = new HeartBeatHandler(clusters, aq, am, injector); Register reg = new Register(); reg.setHostname(hostname1); reg.setResponseId(12); reg.setTimestamp(System.currentTimeMillis() - 300); reg.setAgentVersion(ambariMetaInfo.getServerVersion()); HostInfo hi = new HostInfo(); hi.setOS("Centos5"); reg.setHardwareProfile(hi); handler.handleRegistration(reg); HeartBeat hb = new HeartBeat(); hb.setHostname(hostname1); hb.setNodeStatus(new HostStatus(HostStatus.Status.HEALTHY, "cool")); hb.setTimestamp(System.currentTimeMillis()); hb.setResponseId(12); handler.handleHeartBeat(hb); List<StatusCommand> cmds = hm.generateStatusCommands(hostname1); assertEquals("HeartbeatMonitor should generate StatusCommands for host1", 3, cmds.size()); assertEquals("HDFS", cmds.get(0).getServiceName()); cmds = hm.generateStatusCommands(hostname2); assertTrue("HeartbeatMonitor should not generate StatusCommands for host2 because it has no services", cmds.isEmpty()); }