EmbeddedVHM extends VHMProcess implements ProgressReporter { protected void setNumTTVMsForCluster(VHMInputMessage input) { CompoundStatus thisStatus = new CompoundStatus("setNumTTVMsForCluster"); CompoundStatus vmChooserStatus = null; CompoundStatus edPolicyStatus = null; try { HadoopCluster cluster = new HadoopCluster(input.getClusterName(), input.getJobTrackerAddress()); EnableDisableTTPolicy edPolicy = _config._enableDisablePolicy; VMChooserAlgorithm vmChooser = _config._vmChooser; _pLog.registerProgress(10); _log.log(Level.INFO, "Getting cluster inventory information..."); VMDTO[] allTTs = getAllTTsForAllHosts(thisStatus, input.getTTFolderNames(), input.getSerengetiRootFolder(), input.getClusterName()); if ((allTTs == null) || (allTTs.length == 0)) { processCompoundStatuses(thisStatus, vmChooserStatus, edPolicyStatus); return; } TTStatesForHost[] ttStatesForHosts = _config._enableDisablePolicy.getStateForTTs(allTTs); int totalEnabled = 0; for (TTStatesForHost ttStatesForHost : ttStatesForHosts) { totalEnabled += ttStatesForHost.getEnabled().size(); } int targetTTs = 0; if (input.getTargetTTs() == UNLIMIT_CMD) { _log.log(Level.INFO, "Request to unlimit TT VMs"); targetTTs = allTTs.length; } else { targetTTs = input.getTargetTTs(); } int delta = (targetTTs - totalEnabled); _log.log(Level.INFO, "Total TT VMs = "+allTTs.length+", total powered-on TT VMs = "+totalEnabled+", target powered-on TT VMs = "+targetTTs); _pLog.registerProgress(30); if (input.getTargetTTs() == UNLIMIT_CMD) { VMDTO[] ttsToEnable = chooseAllTTVMs(ttStatesForHosts); _pLog.registerProgress(40); vmChooserStatus = new CompoundStatus("Null VMChooser"); edPolicyStatus = edPolicy.enableTTs(ttsToEnable, ttsToEnable.length, cluster); _pLog.registerProgress(90); } else if (delta > 0) { _log.log(Level.INFO, "Target TT VMs to enable/disable = "+delta); VMCAResult chooserResult = vmChooser.chooseVMsToEnable(ttStatesForHosts, allTTs.length, delta); VMDTO[] ttsToEnable = chooserResult.getChosenVMs(); vmChooserStatus = chooserResult.getChooserStatus(); _pLog.registerProgress(40); edPolicyStatus = edPolicy.enableTTs(ttsToEnable, (ttsToEnable.length + totalEnabled), cluster); _pLog.registerProgress(90); } else if (delta < 0) { _log.log(Level.INFO, "Target TT VMs to enable/disable = "+delta); VMCAResult chooserResult = vmChooser.chooseVMsToDisable(ttStatesForHosts, allTTs.length, 0 - delta); VMDTO[] ttsToDisable = chooserResult.getChosenVMs(); vmChooserStatus = chooserResult.getChooserStatus(); _pLog.registerProgress(40); edPolicyStatus = edPolicy.disableTTs(ttsToDisable, (totalEnabled - ttsToDisable.length), cluster); _pLog.registerProgress(90); } } catch (Exception e) { _log.log(Level.SEVERE, "Unexpected error in core VHM: "+e); thisStatus.registerTaskFailed(true, e.getMessage()); } processCompoundStatuses(thisStatus, vmChooserStatus, edPolicyStatus); }  void init(VHMConfig config, VCActions vc, MQActions mq); void start(); void stop(); @Override void reportProgress(int percentage, String message); void reportError(String message); void reportCompletion();  }
@Test public void testWithFakeQueue() throws IOException { File file = new File("/tmp/vhm.json.msg"); String jsonMsg = FileUtils.readFileToString(file); VHMInputMessage input = new VHMJsonInputMessage(jsonMsg.getBytes()); _test.setNumTTVMsForCluster(input); }