in ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java [1109:1372]
private List<Stage> doStageCreation(Cluster cluster,
Map<State, List<Service>> changedServices,
Map<State, List<ServiceComponent>> changedComps,
Map<String, Map<State, List<ServiceComponentHost>>> changedScHosts,
Map<String, String> requestParameters, String requestContext,
boolean runSmokeTest, boolean reconfigureClients)
throws AmbariException {
// TODO handle different transitions?
// Say HDFS to stopped and MR to started, what order should actions be done
// in?
// TODO additional validation?
// verify all configs
// verify all required components
if ((changedServices == null || changedServices.isEmpty())
&& (changedComps == null || changedComps.isEmpty())
&& (changedScHosts == null || changedScHosts.isEmpty())) {
return null;
}
// smoke test any service that goes from installed to started
Set<String> smokeTestServices = getServicesForSmokeTests(cluster,
changedServices, changedScHosts, runSmokeTest);
if (reconfigureClients) {
// Re-install client only hosts to reattach changed configs on service
// restart
addClientSchForReinstall(cluster, changedServices, changedScHosts);
}
if (!changedScHosts.isEmpty()
|| !smokeTestServices.isEmpty()) {
long nowTimestamp = System.currentTimeMillis();
Long requestId = actionManager.getNextRequestId();
// FIXME cannot work with a single stage
// multiple stages may be needed for reconfigure
long stageId = 0;
Map<String, List<String>> clusterHostInfo = StageUtils.getClusterHostInfo(
clusters.getHostsForCluster(cluster.getClusterName()), cluster, hostsMap, injector.getInstance(Configuration.class));
String clusterHostInfoJson = StageUtils.getGson().toJson(clusterHostInfo);
Stage stage = createNewStage(cluster, requestId, requestContext, clusterHostInfoJson);
stage.setStageId(stageId);
//HACK
String jobtrackerHost = getJobTrackerHost(cluster);
for (String compName : changedScHosts.keySet()) {
for (State newState : changedScHosts.get(compName).keySet()) {
for (ServiceComponentHost scHost :
changedScHosts.get(compName).get(newState)) {
// Do not create role command for hosts that are not responding
if (scHost.getHostState().equals(HostState.HEARTBEAT_LOST)) {
LOG.info("Command is not created for servicecomponenthost "
+ ", clusterName=" + cluster.getClusterName()
+ ", clusterId=" + cluster.getClusterId()
+ ", serviceName=" + scHost.getServiceName()
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", hostState=" + scHost.getHostState()
+ ", targetNewState=" + newState);
continue;
}
RoleCommand roleCommand;
State oldSchState = scHost.getState();
ServiceComponentHostEvent event;
switch (newState) {
case INSTALLED:
if (oldSchState == State.INIT
|| oldSchState == State.UNINSTALLED
|| oldSchState == State.INSTALLED
|| oldSchState == State.INSTALLING
|| oldSchState == State.UNKNOWN
|| oldSchState == State.INSTALL_FAILED) {
roleCommand = RoleCommand.INSTALL;
event = new ServiceComponentHostInstallEvent(
scHost.getServiceComponentName(), scHost.getHostName(),
nowTimestamp,
scHost.getDesiredStackVersion().getStackId());
} else if (oldSchState == State.STARTED
|| oldSchState == State.INSTALLED
|| oldSchState == State.STOPPING) {
roleCommand = RoleCommand.STOP;
event = new ServiceComponentHostStopEvent(
scHost.getServiceComponentName(), scHost.getHostName(),
nowTimestamp);
} else if (oldSchState == State.UPGRADING) {
roleCommand = RoleCommand.UPGRADE;
event = new ServiceComponentHostUpgradeEvent(
scHost.getServiceComponentName(), scHost.getHostName(),
nowTimestamp, scHost.getDesiredStackVersion().getStackId());
} else {
throw new AmbariException("Invalid transition for"
+ " servicecomponenthost"
+ ", clusterName=" + cluster.getClusterName()
+ ", clusterId=" + cluster.getClusterId()
+ ", serviceName=" + scHost.getServiceName()
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", currentState=" + oldSchState
+ ", newDesiredState=" + newState);
}
break;
case STARTED:
StackId stackId = scHost.getDesiredStackVersion();
ComponentInfo compInfo = ambariMetaInfo.getComponentCategory(
stackId.getStackName(), stackId.getStackVersion(), scHost.getServiceName(),
scHost.getServiceComponentName());
if (oldSchState == State.INSTALLED
|| oldSchState == State.STARTING) {
roleCommand = RoleCommand.START;
event = new ServiceComponentHostStartEvent(
scHost.getServiceComponentName(), scHost.getHostName(),
nowTimestamp);
} else {
String error = "Invalid transition for"
+ " servicecomponenthost"
+ ", clusterName=" + cluster.getClusterName()
+ ", clusterId=" + cluster.getClusterId()
+ ", serviceName=" + scHost.getServiceName()
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", currentState=" + oldSchState
+ ", newDesiredState=" + newState;
if (compInfo.isMaster()) {
throw new AmbariException(error);
} else {
LOG.info("Ignoring: " + error);
continue;
}
}
break;
case UNINSTALLED:
if (oldSchState == State.INSTALLED
|| oldSchState == State.UNINSTALLING) {
roleCommand = RoleCommand.UNINSTALL;
event = new ServiceComponentHostStartEvent(
scHost.getServiceComponentName(), scHost.getHostName(),
nowTimestamp);
} else {
throw new AmbariException("Invalid transition for"
+ " servicecomponenthost"
+ ", clusterName=" + cluster.getClusterName()
+ ", clusterId=" + cluster.getClusterId()
+ ", serviceName=" + scHost.getServiceName()
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", currentState=" + oldSchState
+ ", newDesiredState=" + newState);
}
break;
case INIT:
throw new AmbariException("Unsupported transition to INIT for"
+ " servicecomponenthost"
+ ", clusterName=" + cluster.getClusterName()
+ ", clusterId=" + cluster.getClusterId()
+ ", serviceName=" + scHost.getServiceName()
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", currentState=" + oldSchState
+ ", newDesiredState=" + newState);
default:
throw new AmbariException("Unsupported state change operation"
+ ", newState=" + newState.toString());
}
if (LOG.isDebugEnabled()) {
LOG.debug("Create a new host action"
+ ", requestId=" + requestId
+ ", componentName=" + scHost.getServiceComponentName()
+ ", hostname=" + scHost.getHostName()
+ ", roleCommand=" + roleCommand.name());
}
// [ type -> [ key, value ] ]
Map<String, Map<String, String>> configurations = new TreeMap<String, Map<String, String>>();
Host host = clusters.getHost(scHost.getHostName());
Map<String, Map<String, String>> configTags =
findConfigurationTagsWithOverrides(cluster, host.getHostName());
// HACK - Set configs on the ExecCmd
if (!scHost.getHostName().equals(jobtrackerHost)) {
if (configTags.get(Configuration.GLOBAL_CONFIG_TAG) != null) {
configHelper.applyCustomConfig(
configurations, Configuration.GLOBAL_CONFIG_TAG,
Configuration.RCA_ENABLED_PROPERTY, "false", false);
}
}
createHostAction(cluster, stage, scHost, configurations, configTags,
roleCommand, requestParameters, event);
}
}
}
for (String serviceName : smokeTestServices) {
Service s = cluster.getService(serviceName);
// find service component host
String clientHost = getClientHostForRunningAction(cluster, s);
String smokeTestRole =
actionMetadata.getServiceCheckAction(serviceName);
if (clientHost == null || smokeTestRole == null) {
LOG.info("Nothing to do for service check as could not find role or"
+ " or host to run check on"
+ ", clusterName=" + cluster.getClusterName()
+ ", serviceName=" + serviceName
+ ", clientHost=" + clientHost
+ ", serviceCheckRole=" + smokeTestRole);
continue;
}
stage.addHostRoleExecutionCommand(clientHost,
Role.valueOf(smokeTestRole),
RoleCommand.EXECUTE,
new ServiceComponentHostOpInProgressEvent(null, clientHost,
nowTimestamp), cluster.getClusterName(), serviceName);
// [ type -> [ key, value ] ]
Map<String, Map<String, String>> configurations = new TreeMap<String, Map<String,String>>();
Map<String, Map<String, String>> configTags =
findConfigurationTagsWithOverrides(cluster, clientHost);
stage.getExecutionCommandWrapper(clientHost,
smokeTestRole).getExecutionCommand()
.setConfigurations(configurations);
stage.getExecutionCommandWrapper(clientHost,
smokeTestRole).getExecutionCommand()
.setConfigurationTags(configTags);
// Generate cluster host info
stage.getExecutionCommandWrapper(clientHost, smokeTestRole)
.getExecutionCommand()
.setClusterHostInfo(StageUtils.getClusterHostInfo(
clusters.getHostsForCluster(cluster.getClusterName()), cluster, hostsMap,
injector.getInstance(Configuration.class)));
Map<String,String> hostParams = new HashMap<String, String>();
hostParams.put("stack_version", cluster.getDesiredStackVersion().getStackVersion());
// smoke tests need stack version
stage.getExecutionCommandWrapper(clientHost,
smokeTestRole).getExecutionCommand()
.setHostLevelParams(hostParams);
}
RoleCommandOrder rco = getRoleCommandOrder(cluster);
RoleGraph rg = new RoleGraph(rco);
rg.build(stage);
return rg.getStages();
}
return null;
}