HeartBeatHandler { public RegistrationResponse handleRegistration(Register register) throws InvalidStateTransitionException, AmbariException { String hostname = register.getHostname(); int currentPingPort = register.getCurrentPingPort(); long now = System.currentTimeMillis(); String agentVersion = register.getAgentVersion(); String serverVersion = ambariMetaInfo.getServerVersion(); if (!VersionUtils.areVersionsEqual(serverVersion, agentVersion, true)) { LOG.warn("Received registration request from host with non compatible" + " agent version" + ", hostname=" + hostname + ", agentVersion=" + agentVersion + ", serverVersion=" + serverVersion); throw new AmbariException("Cannot register host with non compatible" + " agent version" + ", hostname=" + hostname + ", agentVersion=" + agentVersion + ", serverVersion=" + serverVersion); } String agentOsType = getOsType(register.getHardwareProfile().getOS(), register.getHardwareProfile().getOSRelease()); LOG.info("agentOsType = "+agentOsType ); if (!ambariMetaInfo.isOsSupported(agentOsType)) { LOG.warn("Received registration request from host with not supported" + " os type" + ", hostname=" + hostname + ", serverOsType=" + config.getServerOsType() + ", agentOsType=" + agentOsType); throw new AmbariException("Cannot register host with not supported" + " os type" + ", hostname=" + hostname + ", serverOsType=" + config.getServerOsType() + ", agentOsType=" + agentOsType); } Host hostObject; try { hostObject = clusterFsm.getHost(hostname); } catch (HostNotFoundException ex) { clusterFsm.addHost(hostname); hostObject = clusterFsm.getHost(hostname); } hostObject.setState(HostState.INIT); hostObject.setCurrentPingPort(currentPingPort); List<StatusCommand> cmds = heartbeatMonitor.generateStatusCommands(hostname); for (StatusCommand command: cmds) { command.getCommandParams().put("request_version", String.valueOf(true)); } hostObject.setPrefix(register.getPrefix()); hostObject.handleEvent(new HostRegistrationRequestEvent(hostname, null != register.getPublicHostname() ? register.getPublicHostname() : hostname, new AgentVersion(register.getAgentVersion()), now, register.getHardwareProfile(), register.getAgentEnv())); RegistrationResponse response = new RegistrationResponse(); if (cmds.isEmpty()) { hostObject.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now)); } response.setStatusCommands(cmds); response.setResponseStatus(RegistrationStatus.OK); List<AlertDefinitionCommand> alertDefinitionCommands = getRegistrationAlertDefinitionCommands(hostname); response.setAlertDefinitionCommands(alertDefinitionCommands); response.setAgentConfig(config.getAgentConfigsMap()); if(response.getAgentConfig() != null) { LOG.debug("Agent configuration map set to " + response.getAgentConfig()); } Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname); if (clusters.size() > 0) { String clusterName = clusters.iterator().next().getClusterName(); RecoveryConfig rc = recoveryConfigHelper.getRecoveryConfig(clusterName, hostname); response.setRecoveryConfig(rc); if(response.getRecoveryConfig() != null) { LOG.info("Recovery configuration set to " + response.getRecoveryConfig().toString()); } } Long requestId = 0L; hostResponseIds.put(hostname, requestId); response.setResponseId(requestId); return response; } @Inject  HeartBeatHandler(Clusters fsm, ActionQueue aq, ActionManager am,
                          Injector injector); void start(); void setHeartbeatProcessor(HeartbeatProcessor heartbeatProcessor); HeartbeatProcessor getHeartbeatProcessor(); HeartBeatResponse handleHeartBeat(HeartBeat heartbeat); String getOsType(String os, String osRelease); RegistrationResponse handleRegistration(Register register); ComponentsResponse handleComponents(String clusterName); void stop();  }
@Test public void testRegistrationRecoveryConfigMaintenanceMode() throws Exception, InvalidStateTransitionException { ActionManager am = actionManagerTestHelper.getMockActionManager(); replay(am); Clusters fsm = clusters; HeartBeatHandler handler = new HeartBeatHandler(fsm, new ActionQueue(), am, injector); Cluster cluster = heartbeatTestHelper.getDummyCluster(); Service hdfs = cluster.addService(HDFS); hdfs.addServiceComponent(DATANODE).setRecoveryEnabled(true); hdfs.getServiceComponent(DATANODE); hdfs.getServiceComponent(DATANODE).addServiceComponentHost(DummyHostname1); hdfs.addServiceComponent(NAMENODE).setRecoveryEnabled(true); hdfs.getServiceComponent(NAMENODE); hdfs.getServiceComponent(NAMENODE).addServiceComponentHost(DummyHostname1); hdfs.addServiceComponent(HDFS_CLIENT).setRecoveryEnabled(true); hdfs.getServiceComponent(HDFS_CLIENT); hdfs.getServiceComponent(HDFS_CLIENT).addServiceComponentHost(DummyHostname1); Host hostObject = clusters.getHost(DummyHostname1); hostObject.setIPv4("ipv4"); hostObject.setIPv6("ipv6"); ServiceComponentHost schHdfsClient = hdfs.getServiceComponent(HDFS_CLIENT).getServiceComponentHost(DummyHostname1); schHdfsClient.setMaintenanceState(MaintenanceState.ON); Register reg = new Register(); HostInfo hi = new HostInfo(); hi.setHostName(DummyHostname1); hi.setOS(DummyOsType); reg.setHostname(DummyHostname1); reg.setCurrentPingPort(DummyCurrentPingPort); reg.setHardwareProfile(hi); reg.setAgentVersion(metaInfo.getServerVersion()); reg.setPrefix(Configuration.PREFIX_DIR); RegistrationResponse rr = handler.handleRegistration(reg); RecoveryConfig rc = rr.getRecoveryConfig(); assertEquals(rc.getMaxCount(), "4"); assertEquals(rc.getType(), "AUTO_START"); assertEquals(rc.getMaxLifetimeCount(), "10"); assertEquals(rc.getRetryGap(), "2"); assertEquals(rc.getWindowInMinutes(), "23"); assertEquals(rc.getEnabledComponents(), "DATANODE,NAMENODE"); }
@Test public void testRegistrationAgentConfig() throws Exception, InvalidStateTransitionException { ActionManager am = actionManagerTestHelper.getMockActionManager(); replay(am); Clusters fsm = clusters; HeartBeatHandler handler = new HeartBeatHandler(fsm, new ActionQueue(), am, injector); clusters.addHost(DummyHostname1); Host hostObject = clusters.getHost(DummyHostname1); hostObject.setIPv4("ipv4"); hostObject.setIPv6("ipv6"); Register reg = new Register(); HostInfo hi = new HostInfo(); hi.setHostName(DummyHostname1); hi.setOS(DummyOsType); reg.setHostname(DummyHostname1); reg.setCurrentPingPort(DummyCurrentPingPort); reg.setHardwareProfile(hi); reg.setAgentVersion(metaInfo.getServerVersion()); reg.setPrefix(Configuration.PREFIX_DIR); RegistrationResponse rr = handler.handleRegistration(reg); Map<String, String> config = rr.getAgentConfig(); assertFalse(config.isEmpty()); assertTrue(config.containsKey(Configuration.CHECK_REMOTE_MOUNTS.getKey())); assertTrue("false".equals(config.get(Configuration.CHECK_REMOTE_MOUNTS.getKey()))); assertTrue(config.containsKey(Configuration.CHECK_MOUNTS_TIMEOUT.getKey())); assertTrue("0".equals(config.get(Configuration.CHECK_MOUNTS_TIMEOUT.getKey()))); assertTrue("true".equals(config.get(Configuration.ENABLE_AUTO_AGENT_CACHE_UPDATE.getKey()))); }
@Test public void testRegistrationWithBadVersion() throws Exception, InvalidStateTransitionException { ActionManager am = actionManagerTestHelper.getMockActionManager(); replay(am); Clusters fsm = clusters; HeartBeatHandler handler = new HeartBeatHandler(fsm, new ActionQueue(), am, injector); clusters.addHost(DummyHostname1); Host hostObject = clusters.getHost(DummyHostname1); hostObject.setIPv4("ipv4"); hostObject.setIPv6("ipv6"); Register reg = new Register(); HostInfo hi = new HostInfo(); hi.setHostName(DummyHostname1); hi.setOS(DummyOsType); reg.setHostname(DummyHostname1); reg.setHardwareProfile(hi); reg.setAgentVersion(""); reg.setPrefix(Configuration.PREFIX_DIR); try { handler.handleRegistration(reg); fail ("Expected failure for non compatible agent version"); } catch (AmbariException e) { log.debug("Error:" + e.getMessage()); Assert.assertTrue(e.getMessage().contains( "Cannot register host with non compatible agent version")); } reg.setAgentVersion(null); try { handler.handleRegistration(reg); fail ("Expected failure for non compatible agent version"); } catch (AmbariException e) { log.debug("Error:" + e.getMessage()); Assert.assertTrue(e.getMessage().contains( "Cannot register host with non compatible agent version")); } }
@Test public void testInvalidOSRegistration() throws Exception, InvalidStateTransitionException { ActionManager am = actionManagerTestHelper.getMockActionManager(); replay(am); Clusters fsm = clusters; HeartBeatHandler handler = new HeartBeatHandler(fsm, new ActionQueue(), am, injector); clusters.addHost(DummyHostname1); Host hostObject = clusters.getHost(DummyHostname1); hostObject.setIPv4("ipv4"); hostObject.setIPv6("ipv6"); Register reg = new Register(); HostInfo hi = new HostInfo(); hi.setHostName(DummyHostname1); hi.setOS("MegaOperatingSystem"); reg.setHostname(DummyHostname1); reg.setHardwareProfile(hi); reg.setAgentVersion(metaInfo.getServerVersion()); try { handler.handleRegistration(reg); fail ("Expected failure for non matching os type"); } catch (AmbariException e) { } }
@Test public void testIncompatibleAgentRegistration() throws Exception, InvalidStateTransitionException { ActionManager am = actionManagerTestHelper.getMockActionManager(); replay(am); Clusters fsm = clusters; HeartBeatHandler handler = new HeartBeatHandler(fsm, new ActionQueue(), am, injector); clusters.addHost(DummyHostname1); Host hostObject = clusters.getHost(DummyHostname1); hostObject.setIPv4("ipv4"); hostObject.setIPv6("ipv6"); Register reg = new Register(); HostInfo hi = new HostInfo(); hi.setHostName(DummyHostname1); hi.setOS(DummyOsType); reg.setHostname(DummyHostname1); reg.setHardwareProfile(hi); reg.setAgentVersion("0.0.0"); try { handler.handleRegistration(reg); fail ("Expected failure for non compatible agent version"); } catch (AmbariException e) { } }