Repository: ambari Updated Branches: refs/heads/branch-2.1 e606c9f95 -> 504b00b83
AMBARI-12889 - Distribute Repository For Upgrade With Unhealthy Hosts (jonathanhurley) Project: http://git-wip-us.apache.org/repos/asf/ambari/repo Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/c7c873b5 Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/c7c873b5 Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/c7c873b5 Branch: refs/heads/branch-2.1 Commit: c7c873b5a9ca16570ee95616fa67d0684eebb8e1 Parents: e606c9f Author: Jonathan Hurley <[email protected]> Authored: Wed Aug 26 17:10:45 2015 -0400 Committer: Jonathan Hurley <[email protected]> Committed: Fri Aug 28 18:16:02 2015 -0400 ---------------------------------------------------------------------- .../ClusterStackVersionResourceProvider.java | 67 ++++++++++++------ .../org/apache/ambari/server/state/Cluster.java | 24 +++++-- .../server/state/cluster/ClusterImpl.java | 42 ++++++------ ...ClusterStackVersionResourceProviderTest.java | 11 ++- .../server/state/cluster/ClusterTest.java | 71 ++++++++++++++++++-- 5 files changed, 162 insertions(+), 53 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ambari/blob/c7c873b5/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java index 6590d00..be0af16 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProvider.java @@ -20,6 +20,7 @@ package org.apache.ambari.server.controller.internal; import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.JDK_LOCATION; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -70,7 +71,9 @@ import org.apache.ambari.server.orm.entities.RepositoryVersionEntity; import org.apache.ambari.server.orm.entities.StackEntity; import org.apache.ambari.server.serveraction.upgrades.FinalizeUpgradeAction; 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.MaintenanceState; import org.apache.ambari.server.state.RepositoryVersionState; import org.apache.ambari.server.state.ServiceComponentHost; import org.apache.ambari.server.state.ServiceInfo; @@ -300,17 +303,19 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou desiredRepoVersion = (String) propertyMap.get(CLUSTER_STACK_VERSION_REPOSITORY_VERSION_PROPERTY_ID); Cluster cluster; - Map<String, Host> hostsForCluster; - AmbariManagementController managementController = getManagementController(); AmbariMetaInfo ami = managementController.getAmbariMetaInfo(); + try { - cluster = managementController.getClusters().getCluster(clName); - hostsForCluster = managementController.getClusters().getHostsForCluster(clName); + Clusters clusters = managementController.getClusters(); + cluster = clusters.getCluster(clName); } catch (AmbariException e) { throw new NoSuchParentResourceException(e.getMessage(), e); } + // get all of the host eligible for stack distribution + List<Host> hosts = getHostsForStackDistribution(cluster); + final StackId stackId; if (propertyMap.containsKey(CLUSTER_STACK_VERSION_STACK_PROPERTY_ID) && propertyMap.containsKey(CLUSTER_STACK_VERSION_VERSION_PROPERTY_ID)) { @@ -352,13 +357,13 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou RequestStageContainer req = createRequest(); - Iterator<Host> hostsForClusterIter = hostsForCluster.values().iterator(); + Iterator<Host> hostIterator = hosts.iterator(); Map<String, String> hostLevelParams = new HashMap<String, String>(); hostLevelParams.put(JDK_LOCATION, getManagementController().getJdkResourceUrl()); String hostParamsJson = StageUtils.getGson().toJson(hostLevelParams); int maxTasks = configuration.getAgentPackageParallelCommandsLimit(); - int hostCount = hostsForCluster.size(); + int hostCount = hosts.size(); int batchCount = (int) (Math.ceil((double)hostCount / maxTasks)); long stageId = req.getLastStageId() + 1; @@ -394,25 +399,27 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou stages.add(stage); // Populate with commands for host - for (int i = 0; i < maxTasks && hostsForClusterIter.hasNext(); i++) { - Host host = hostsForClusterIter.next(); + for (int i = 0; i < maxTasks && hostIterator.hasNext(); i++) { + Host host = hostIterator.next(); addHostVersionInstallCommandsToStage(desiredRepoVersion, cluster, managementController, ami, stackId, perOsRepos, stage, host); } } + req.addStages(stages); try { - ClusterVersionEntity existingCSVer = clusterVersionDAO.findByClusterAndStackAndVersion( + ClusterVersionEntity clusterVersionEntity = clusterVersionDAO.findByClusterAndStackAndVersion( clName, stackId, desiredRepoVersion); - if (existingCSVer == null) { + if (clusterVersionEntity == null) { try { // Create/persist new cluster stack version cluster.createClusterVersion(stackId, desiredRepoVersion, managementController.getAuthName(), RepositoryVersionState.INSTALLING); - existingCSVer = clusterVersionDAO.findByClusterAndStackAndVersion( + + clusterVersionEntity = clusterVersionDAO.findByClusterAndStackAndVersion( clName, stackId, desiredRepoVersion); } catch (AmbariException e) { throw new SystemException( @@ -427,7 +434,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou } // Will also initialize all Host Versions in an INSTALLING state. - cluster.inferHostVersions(existingCSVer); + cluster.transitionHostsToInstalling(clusterVersionEntity); req.persist(); @@ -438,11 +445,9 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou } private void addHostVersionInstallCommandsToStage(final String desiredRepoVersion, - Cluster cluster, AmbariManagementController managementController, - AmbariMetaInfo ami, - final StackId stackId, - Map<String, List<RepositoryEntity>> perOsRepos, - Stage stage, Host host) throws SystemException { + Cluster cluster, AmbariManagementController managementController, AmbariMetaInfo ami, + final StackId stackId, Map<String, List<RepositoryEntity>> perOsRepos, Stage stage, Host host) + throws SystemException { // Determine repositories for host final List<RepositoryEntity> repoInfo = perOsRepos.get(host.getOsFamily()); if (repoInfo == null) { @@ -450,7 +455,8 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou "not defined. Repo version=%s, stackId=%s", host.getOsFamily(), desiredRepoVersion, stackId)); } - // For every host at cluster, determine packages for all installed services + + // determine packages for all services that are installed on host List<ServiceOsSpecific.Package> packages = new ArrayList<ServiceOsSpecific.Package>(); Set<String> servicesOnHost = new HashSet<String>(); List<ServiceComponentHost> components = cluster.getServiceComponentHosts(host.getHostName()); @@ -475,6 +481,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou } } } + final String packageList = gson.toJson(packages); final String repoList = gson.toJson(repoInfo); @@ -501,7 +508,6 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou } } - private RequestStageContainer createRequest() { ActionManager actionManager = getManagementController().getActionManager(); @@ -631,4 +637,27 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou protected Set<String> getPKPropertyIds() { return pkPropertyIds; } + + /** + * Gets all of the hosts in a cluster which are not in "maintenance mode" and + * are considered to be healthy. In the case of stack distribution, a host + * must be explicitely marked as being in maintenance mode for it to be + * considered as unhealthy. + * + * @param cluster + * the cluster (not {@code null}). + * @return the list of hosts that are not in maintenance mode and are + * elidgable to have a stack distributed to them. + */ + private List<Host> getHostsForStackDistribution(Cluster cluster) { + Collection<Host> hosts = cluster.getHosts(); + List<Host> healthyHosts = new ArrayList<>(hosts.size()); + for (Host host : hosts) { + if (host.getMaintenanceState(cluster.getClusterId()) == MaintenanceState.OFF) { + healthyHosts.add(host); + } + } + + return healthyHosts; + } } http://git-wip-us.apache.org/repos/asf/ambari/blob/c7c873b5/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java index fe669bd..916bc49 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/state/Cluster.java @@ -180,15 +180,25 @@ public interface Cluster { RepositoryVersionState desiredState) throws AmbariException; /** - * Create/update host versions for all of the hosts within a cluster based on state of cluster stack version. - * The difference of this method compared to {@link Cluster#mapHostVersions} - * is that it affects all hosts (not only missing hosts). Also, current method contains some additional logics to allow only INSTALLING - * state for hosts. - * @param sourceClusterVersion cluster version to be queried for a stack name/version info and desired RepositoryVersionState. The only valid state - * of a cluster version is {@link RepositoryVersionState#INSTALLING} + * Creates or updates host versions for all of the hosts within a cluster + * based on state of cluster stack version. This is used to transition all + * hosts into the {@link RepositoryVersionState#INSTALLING} state. + * <p/> + * The difference between this method compared to + * {@link Cluster#mapHostVersions} is that it affects all hosts (not only + * missing hosts). + * <p/> + * Hosts that are in maintenance mode will not be included. These hosts have + * been explicitely marked as being in maintenance andd are not included in + * this operation. + * + * @param sourceClusterVersion + * cluster version to be queried for a stack name/version info and + * desired RepositoryVersionState. The only valid state of a cluster + * version is {@link RepositoryVersionState#INSTALLING} * @throws AmbariException */ - void inferHostVersions(ClusterVersionEntity sourceClusterVersion) + void transitionHostsToInstalling(ClusterVersionEntity sourceClusterVersion) throws AmbariException; /** http://git-wip-us.apache.org/repos/asf/ambari/blob/c7c873b5/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java index 86f5f32..4fe24e9 100644 --- a/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java +++ b/ambari-server/src/main/java/org/apache/ambari/server/state/cluster/ClusterImpl.java @@ -1133,27 +1133,18 @@ public class ClusterImpl implements Cluster { } /** - * Because this is a top-down approach, it should only be called for the purposes of bootstrapping data, such as - * installing a brand new cluster (through Blueprints). - * @param sourceClusterVersion cluster version to be queried for a stack name/version info and desired RepositoryVersionState. - * The only valid state of this cluster version is {@link RepositoryVersionState#INSTALLING} - * @throws AmbariException + * {@inheritDoc} */ @Override - public void inferHostVersions(ClusterVersionEntity sourceClusterVersion) throws AmbariException { + public void transitionHostsToInstalling(ClusterVersionEntity sourceClusterVersion) throws AmbariException { if (sourceClusterVersion == null) { throw new AmbariException("Could not find current stack version of cluster " + getClusterName()); } - RepositoryVersionState desiredState = sourceClusterVersion.getState(); - - @SuppressWarnings("serial") - Set<RepositoryVersionState> validStates = new HashSet<RepositoryVersionState>(){{ - add(RepositoryVersionState.INSTALLING); - }}; - - if (!validStates.contains(desiredState)) { - throw new AmbariException("The state must be one of " + validStates); + if (RepositoryVersionState.INSTALLING != sourceClusterVersion.getState()) { + throw new AmbariException("Unable to transition cluster hosts into " + + RepositoryVersionState.INSTALLING + + ". The only valid state is INSTALLING"); } Map<String, Host> hosts = clusters.getHostsForCluster(getClusterName()); @@ -1169,17 +1160,26 @@ public class ClusterImpl implements Cluster { getClusterName(), repoVersionStackId, sourceClusterVersion.getRepositoryVersion().getVersion()); - if (existingHostVersionEntities != null) { - for (HostVersionEntity entity : existingHostVersionEntities) { - existingHostsWithClusterStackAndVersion.add(entity.getHostName()); - existingHostStackVersions.put(entity.getHostName(), entity); - } + // for each host that already has a stack and version, keep track of them + for (HostVersionEntity entity : existingHostVersionEntities) { + String hostName = entity.getHostName(); + existingHostsWithClusterStackAndVersion.add(hostName); + existingHostStackVersions.put(hostName, entity); } + // find any hosts that do not have the stack/repo version already Sets.SetView<String> hostsMissingRepoVersion = Sets.difference( hosts.keySet(), existingHostsWithClusterStackAndVersion); for (String hostname : hosts.keySet()) { + // if the host is in maintenance mode, that's an explicit marker which + // indicates that it should not be transitioned to INSTALLING; these + // hosts are excluded from being transitioned into INSTALLING + Host host = hosts.get(hostname); + if (host.getMaintenanceState(getClusterId()) != MaintenanceState.OFF) { + continue; + } + if (hostsMissingRepoVersion.contains(hostname)) { // Create new host stack version HostEntity hostEntity = hostDAO.findByName(hostname); @@ -1190,7 +1190,7 @@ public class ClusterImpl implements Cluster { } else { // Update existing host stack version HostVersionEntity hostVersionEntity = existingHostStackVersions.get(hostname); - hostVersionEntity.setState(desiredState); + hostVersionEntity.setState(RepositoryVersionState.INSTALLING); hostVersionDAO.merge(hostVersionEntity); } } http://git-wip-us.apache.org/repos/asf/ambari/blob/c7c873b5/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProviderTest.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProviderTest.java b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProviderTest.java index ece45f9..7b37483 100644 --- a/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProviderTest.java +++ b/ambari-server/src/test/java/org/apache/ambari/server/controller/internal/ClusterStackVersionResourceProviderTest.java @@ -75,11 +75,13 @@ import org.apache.ambari.server.state.Cluster; import org.apache.ambari.server.state.Clusters; import org.apache.ambari.server.state.ConfigHelper; import org.apache.ambari.server.state.Host; +import org.apache.ambari.server.state.MaintenanceState; import org.apache.ambari.server.state.ServiceComponentHost; import org.apache.ambari.server.state.ServiceInfo; import org.apache.ambari.server.state.ServiceOsSpecific; import org.apache.ambari.server.state.StackId; import org.apache.ambari.server.state.cluster.ClusterImpl; +import org.easymock.EasyMock; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -171,6 +173,9 @@ public class ClusterStackVersionResourceProviderTest { Host host = createNiceMock(hostname, Host.class); expect(host.getHostName()).andReturn(hostname).anyTimes(); expect(host.getOsFamily()).andReturn("redhat6").anyTimes(); + expect(host.getMaintenanceState(EasyMock.anyLong())).andReturn( + MaintenanceState.OFF).anyTimes(); + replay(host); hostsForCluster.put(hostname, host); } @@ -208,8 +213,12 @@ public class ClusterStackVersionResourceProviderTest { eq(managementController))).andReturn(csvResourceProvider).anyTimes(); expect(clusters.getCluster(anyObject(String.class))).andReturn(cluster); - expect(clusters.getHostsForCluster(anyObject(String.class))).andReturn(hostsForCluster); + expect(clusters.getHostsForCluster(anyObject(String.class))).andReturn( + hostsForCluster).anyTimes(); + String clusterName = "Cluster100"; + expect(cluster.getClusterId()).andReturn(1L).anyTimes(); + expect(cluster.getHosts()).andReturn(hostsForCluster.values()).atLeastOnce(); expect(cluster.getCurrentStackVersion()).andReturn(stackId); expect(cluster.getServiceComponentHosts(anyObject(String.class))).andReturn(schs).anyTimes(); http://git-wip-us.apache.org/repos/asf/ambari/blob/c7c873b5/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java ---------------------------------------------------------------------- diff --git a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java index f047b33..edd6267 100644 --- a/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java +++ b/ambari-server/src/test/java/org/apache/ambari/server/state/cluster/ClusterTest.java @@ -42,8 +42,6 @@ import java.util.Set; import javax.persistence.EntityManager; import javax.persistence.RollbackException; -import junit.framework.Assert; - import org.apache.ambari.server.AmbariException; import org.apache.ambari.server.agent.AgentEnv; import org.apache.ambari.server.agent.AgentEnv.Directory; @@ -88,6 +86,7 @@ import org.apache.ambari.server.state.DesiredConfig; 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.MaintenanceState; import org.apache.ambari.server.state.RepositoryVersionState; import org.apache.ambari.server.state.Service; import org.apache.ambari.server.state.ServiceComponent; @@ -117,6 +116,8 @@ import com.google.inject.persist.Transactional; import com.google.inject.persist.UnitOfWork; import com.google.inject.util.Modules; +import junit.framework.Assert; + public class ClusterTest { private Clusters clusters; @@ -1423,14 +1424,24 @@ public class ClusterTest { assertNotNull(c1.getCurrentClusterVersion()); } + /** + * Tests that hosts can be correctly transitioned into the "INSTALLING" state. + * This method also tests that hosts in MM will not be transitioned, as per + * the contract of + * {@link Cluster#transitionHostsToInstalling(ClusterVersionEntity)}. + * + * @throws Exception + */ @Test - public void testInferHostVersions() throws Exception { + public void testTransitionHostVersions() throws Exception { createDefaultCluster(); StackId stackId = new StackId("HDP", "0.2"); helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion()); + c1.createClusterVersion(stackId, "0.2", "admin", RepositoryVersionState.INSTALLING); + ClusterVersionEntity entityHDP2 = null; for (ClusterVersionEntity entity : c1.getAllClusterVersions()) { StackEntity repoVersionStackEntity = entity.getRepositoryVersion().getStack(); @@ -1442,12 +1453,13 @@ public class ClusterTest { break; } } + assertNotNull(entityHDP2); List<HostVersionEntity> hostVersionsH1Before = hostVersionDAO.findByClusterAndHost("c1", "h1"); assertEquals(1, hostVersionsH1Before.size()); - c1.inferHostVersions(entityHDP2); + c1.transitionHostsToInstalling(entityHDP2); List<HostVersionEntity> hostVersionsH1After = hostVersionDAO.findByClusterAndHost("c1", "h1"); assertEquals(2, hostVersionsH1After.size()); @@ -1462,10 +1474,11 @@ public class ClusterTest { break; } } + assertTrue(checked); // Test for update of existing host stack version - c1.inferHostVersions(entityHDP2); + c1.transitionHostsToInstalling(entityHDP2); hostVersionsH1After = hostVersionDAO.findByClusterAndHost("c1", "h1"); assertEquals(2, hostVersionsH1After.size()); @@ -1480,7 +1493,55 @@ public class ClusterTest { break; } } + assertTrue(checked); + + // reset all to INSTALL_FAILED + List<HostVersionEntity> hostVersionEntities = hostVersionDAO.findAll(); + for (HostVersionEntity hostVersionEntity : hostVersionEntities) { + hostVersionEntity.setState(RepositoryVersionState.INSTALL_FAILED); + hostVersionDAO.merge(hostVersionEntity); + } + + // verify they have been transition to INSTALL_FAILED + hostVersionEntities = hostVersionDAO.findAll(); + for (HostVersionEntity hostVersionEntity : hostVersionEntities) { + assertEquals(RepositoryVersionState.INSTALL_FAILED, hostVersionEntity.getState()); + } + + // put 1 host in maintenance mode + Collection<Host> hosts = c1.getHosts(); + Iterator<Host> iterator = hosts.iterator(); + Host hostInMaintenanceMode = iterator.next(); + Host hostNotInMaintenanceMode = iterator.next(); + hostInMaintenanceMode.setMaintenanceState(c1.getClusterId(), MaintenanceState.ON); + + // transition host versions to INSTALLING + c1.transitionHostsToInstalling(entityHDP2); + + List<HostVersionEntity> hostInMaintModeVersions = hostVersionDAO.findByClusterAndHost("c1", + hostInMaintenanceMode.getHostName()); + + List<HostVersionEntity> otherHostVersions = hostVersionDAO.findByClusterAndHost("c1", + hostNotInMaintenanceMode.getHostName()); + + // verify the MM host is in INSTALL_FAILED still + for (HostVersionEntity hostVersionEntity : hostInMaintModeVersions) { + StackEntity repoVersionStackEntity = hostVersionEntity.getRepositoryVersion().getStack(); + if (repoVersionStackEntity.getStackName().equals("HDP") + && repoVersionStackEntity.getStackVersion().equals("0.2")) { + assertEquals(RepositoryVersionState.INSTALL_FAILED, hostVersionEntity.getState()); + } + } + + // verify the other host is in INSTALLING + for (HostVersionEntity hostVersionEntity : otherHostVersions) { + StackEntity repoVersionStackEntity = hostVersionEntity.getRepositoryVersion().getStack(); + if (repoVersionStackEntity.getStackName().equals("HDP") + && repoVersionStackEntity.getStackVersion().equals("0.2")) { + assertEquals(RepositoryVersionState.INSTALLING, hostVersionEntity.getState()); + } + } } @Test
