AMBARI-9540. RU: Fail in DOWNGRADE finalize (ncole)

ambari-server/src/main/java/org/apache/ambari/server/serveraction/upgrades/FinalizeUpgradeAction.java

@@ -17,7 +17,15 @@
  */
 package org.apache.ambari.server.serveraction.upgrades;
 
-import com.google.inject.Inject;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.text.MessageFormat;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
 
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
@@ -34,17 +42,12 @@ import org.apache.ambari.server.serveraction.AbstractServerAction;
 import org.apache.ambari.server.state.Cluster;
 import org.apache.ambari.server.state.Clusters;
 import org.apache.ambari.server.state.RepositoryVersionState;
-import  org.apache.ambari.server.state.StackId;
+import org.apache.ambari.server.state.StackId;
 import org.apache.ambari.server.state.UpgradeState;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostSummary;
 import org.apache.commons.lang.StringUtils;
 
-import java.text.MessageFormat;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ConcurrentMap;
+import com.google.inject.Inject;
 
 /**
  * Action that represents finalizing the Upgrade by completing any database changes.
@@ -71,19 +74,40 @@ public class FinalizeUpgradeAction extends AbstractServerAction {
   private AmbariMetaInfo ambariMetaInfo;
 
   @Override
-  public CommandReport execute(
-      ConcurrentMap<String, Object> requestSharedDataContext)
+  public CommandReport execute(ConcurrentMap<String, Object> requestSharedDataContext)
       throws AmbariException, InterruptedException {
 
+    Map<String, String> commandParams = getExecutionCommand().getCommandParams();
+
+    boolean isDowngrade = commandParams.containsKey("upgrade_direction") &&
+        "downgrade".equals(commandParams.get("upgrade_direction").toLowerCase());
+
+    String version = commandParams.get("version");
+    String clusterName = getExecutionCommand().getClusterName();
+
+    if (isDowngrade) {
+      return executeDowngrade(clusterName, version);
+    } else {
+      return executeUpgrade(clusterName, version);
+    }
+  }
+
+  /**
+   * Execution path for upgrade.
+   * @param clusterName the name of the cluster the upgrade is for
+   * @param version     the target version of the upgrade
+   * @return the command report
+   */
+  private CommandReport executeUpgrade(String clusterName, String version)
+    throws AmbariException, InterruptedException {
+
     Set<RepositoryVersionState> allowedStates = new HashSet<RepositoryVersionState>();
     allowedStates.add(RepositoryVersionState.UPGRADED);
 
-    StringBuffer outSB = new StringBuffer();
-    StringBuffer errSB = new StringBuffer();
+    StringBuilder outSB = new StringBuilder();
+    StringBuilder errSB = new StringBuilder();
 
     try {
-      String version = this.getExecutionCommand().getCommandParams().get("version");
-      String clusterName = this.getExecutionCommand().getClusterName();
       outSB.append(MessageFormat.format("Begin finalizing the upgrade of cluster {0} to version {1}\n", clusterName, version));
 
       Cluster cluster = clusters.getCluster(clusterName);
@@ -182,4 +206,96 @@ public class FinalizeUpgradeAction extends AbstractServerAction {
       return createCommandReport(-1, HostRoleStatus.FAILED, "{}", outSB.toString(), errSB.toString());
     }
   }
+
+  /**
+   * Execution path for downgrade.
+   * @param clusterName the name of the cluster the downgrade is for
+   * @param version     the target version of the downgrade
+   * @return the command report
+   */
+  private CommandReport executeDowngrade(String clusterName, String version)
+      throws AmbariException, InterruptedException {
+
+    StringBuilder out = new StringBuilder();
+    StringBuilder err = new StringBuilder();
+
+    try {
+      Cluster cluster = clusters.getCluster(clusterName);
+      StackId stackId = cluster.getDesiredStackVersion();
+
+      // !!! find and make sure the cluster_version EXCEPT current are set back
+      out.append(String.format("Searching for current version for %s\n", clusterName));
+      ClusterVersionEntity clusterVersion = clusterVersionDAO.findByClusterAndStateCurrent(clusterName);
+      if (null == clusterVersion) {
+        throw new AmbariException("Could not find current cluster version");
+      }
+
+      out.append(String.format("Comparing downgrade version %s to current cluster version %s\n",
+          version,
+          clusterVersion.getRepositoryVersion().getVersion()));
+
+      if (!version.equals(clusterVersion.getRepositoryVersion().getVersion())) {
+        throw new AmbariException(
+            String.format("Downgrade version %s is not the current cluster version of %s",
+                version, clusterVersion.getRepositoryVersion().getVersion()));
+      } else {
+        out.append(String.format("Downgrade version is the same as current.  Searching " +
+          "for cluster versions that do not match %s\n", version));
+      }
+
+      Set<String> badVersions = new HashSet<String>();
+      // update the cluster version
+      for (ClusterVersionEntity cve : clusterVersionDAO.findByCluster(clusterName)) {
+        switch (cve.getState()) {
+          case UPGRADE_FAILED:
+          case UPGRADED:
+          case UPGRADING: {
+              badVersions.add(cve.getRepositoryVersion().getVersion());
+              cve.setState(RepositoryVersionState.INSTALLED);
+              clusterVersionDAO.merge(cve);
+              break;
+            }
+          default:
+            break;
+        }
+      }
+      out.append(String.format("Found %d other version(s) not matching downgrade: %s\n",
+          badVersions.size(), StringUtils.join(badVersions, ", ")));
+
+      Set<String> badHosts = new HashSet<String>();
+      for (String badVersion : badVersions) {
+        List<HostVersionEntity> hostVersions = hostVersionDAO.findByClusterStackAndVersion(
+            clusterName, stackId.getStackId(), badVersion);
+
+        for (HostVersionEntity hostVersion : hostVersions) {
+          badHosts.add(hostVersion.getHostName());
+          hostVersion.setState(RepositoryVersionState.INSTALLED);
+          hostVersionDAO.merge(hostVersion);
+        }
+      }
+
+      out.append(String.format("Found %d hosts not matching downgrade version: %s\n",
+          badHosts.size(), version));
+
+      for (String badHost : badHosts) {
+        List<HostComponentStateEntity> hostComponentStates = hostComponentStateDAO.findByHost(badHost);
+        for (HostComponentStateEntity hostComponentState : hostComponentStates) {
+          hostComponentState.setUpgradeState(UpgradeState.NONE);
+          hostComponentStateDAO.merge(hostComponentState);
+        }
+      }
+
+      return createCommandReport(0, HostRoleStatus.COMPLETED, "{}",
+          out.toString(), err.toString());
+
+    } catch (Exception e) {
+      StringWriter sw = new StringWriter();
+      e.printStackTrace(new PrintWriter(sw));
+      err.append(sw.toString());
+
+      return createCommandReport(-1, HostRoleStatus.FAILED, "{}",
+          out.toString(), err.toString());
+    }
+  }
+
 }

ambari-server/src/test/java/org/apache/ambari/server/serveraction/upgrades/UpgradeActionTest.java

@@ -0,0 +1,176 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ambari.server.serveraction.upgrades;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.ambari.server.actionmanager.ExecutionCommandWrapper;
+import org.apache.ambari.server.actionmanager.HostRoleCommand;
+import org.apache.ambari.server.actionmanager.HostRoleStatus;
+import org.apache.ambari.server.agent.CommandReport;
+import org.apache.ambari.server.agent.ExecutionCommand;
+import org.apache.ambari.server.orm.GuiceJpaInitializer;
+import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
+import org.apache.ambari.server.orm.OrmTestHelper;
+import org.apache.ambari.server.orm.dao.ClusterVersionDAO;
+import org.apache.ambari.server.orm.dao.HostDAO;
+import org.apache.ambari.server.orm.dao.HostVersionDAO;
+import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
+import org.apache.ambari.server.orm.entities.ClusterVersionEntity;
+import org.apache.ambari.server.orm.entities.HostVersionEntity;
+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.RepositoryVersionState;
+import org.apache.ambari.server.state.StackId;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.persist.PersistService;
+
+/**
+ * Tests upgrade-related server side actions
+ */
+public class UpgradeActionTest {
+  private static final String UPGRADE_VERSION = "2.2.1.0-2270";
+  private static final String DOWNGRADE_VERSION = "2.2.0.0-2041";
+
+  private Injector m_injector;
+
+  @Before
+  public void setup() throws Exception {
+    m_injector = Guice.createInjector(new InMemoryDefaultTestModule());
+    m_injector.getInstance(GuiceJpaInitializer.class);
+
+  }
+
+  @After
+  public void teardown() throws Exception {
+    m_injector.getInstance(PersistService.class).stop();
+  }
+
+  private void makeDowngradeCluster() throws Exception {
+    String clusterName = "c1";
+    String hostName = "h1";
+
+    Clusters clusters = m_injector.getInstance(Clusters.class);
+    clusters.addCluster(clusterName);
+
+    StackId stackId = new StackId("HDP-2.1.1");
+
+    Cluster c = clusters.getCluster(clusterName);
+    c.setDesiredStackVersion(stackId);
+
+    // add a host component
+    clusters.addHost(hostName);
+
+    Host host = clusters.getHost(hostName);
+
+    Map<String, String> hostAttributes = new HashMap<String, String>();
+    hostAttributes.put("os_family", "redhat");
+    hostAttributes.put("os_release_version", "6");
+    host.setHostAttributes(hostAttributes);
+    host.persist();
+
+
+    OrmTestHelper helper = m_injector.getInstance(OrmTestHelper.class);
+
+    helper.getOrCreateRepositoryVersion(stackId.getStackId(), DOWNGRADE_VERSION);
+    helper.getOrCreateRepositoryVersion(stackId.getStackId(), UPGRADE_VERSION);
+
+    RepositoryVersionDAO repoVersionDao = m_injector.getInstance(RepositoryVersionDAO.class);
+    HostVersionDAO hostVersionDao = m_injector.getInstance(HostVersionDAO.class);
+
+    c.createClusterVersion(stackId.getStackId(), DOWNGRADE_VERSION, "admin",
+        RepositoryVersionState.UPGRADING);
+    c.createClusterVersion(stackId.getStackId(), UPGRADE_VERSION, "admin",
+        RepositoryVersionState.INSTALLING);
+
+    c.transitionClusterVersion(stackId.getStackId(), DOWNGRADE_VERSION, RepositoryVersionState.CURRENT);
+    c.transitionClusterVersion(stackId.getStackId(), UPGRADE_VERSION, RepositoryVersionState.INSTALLED);
+    c.transitionClusterVersion(stackId.getStackId(), UPGRADE_VERSION, RepositoryVersionState.UPGRADING);
+
+    c.mapHostVersions(Collections.singleton(hostName), c.getCurrentClusterVersion(),
+        RepositoryVersionState.CURRENT);
+
+    HostDAO hostDAO = m_injector.getInstance(HostDAO.class);
+
+    HostVersionEntity entity = new HostVersionEntity();
+    entity.setHostEntity(hostDAO.findByName(hostName));
+    entity.setHostName(hostName);
+    entity.setRepositoryVersion(
+        repoVersionDao.findByStackAndVersion(stackId.getStackId(), UPGRADE_VERSION));
+    entity.setState(RepositoryVersionState.UPGRADING);
+    hostVersionDao.create(entity);
+  }
+
+
+  @Test
+  public void testFinalizeDowngrade() throws Exception {
+    makeDowngradeCluster();
+
+    Map<String, String> commandParams = new HashMap<String, String>();
+    commandParams.put("upgrade_direction", "downgrade");
+    commandParams.put("version", DOWNGRADE_VERSION);
+
+    ExecutionCommand executionCommand = new ExecutionCommand();
+    executionCommand.setCommandParams(commandParams);
+    executionCommand.setClusterName("c1");
+
+    HostRoleCommand hostRoleCommand = new HostRoleCommand(null, null, null, null);
+    hostRoleCommand.setExecutionCommandWrapper(new ExecutionCommandWrapper(executionCommand));
+
+    FinalizeUpgradeAction action = m_injector.getInstance(FinalizeUpgradeAction.class);
+    action.setExecutionCommand(executionCommand);
+    action.setHostRoleCommand(hostRoleCommand);
+
+    CommandReport report = action.execute(null);
+    assertNotNull(report);
+    assertEquals(HostRoleStatus.COMPLETED.name(), report.getStatus());
+
+    HostVersionDAO hostVersionDao = m_injector.getInstance(HostVersionDAO.class);
+
+    for (HostVersionEntity entity : hostVersionDao.findByClusterAndHost("c1", "h1")) {
+      if (entity.getRepositoryVersion().getVersion().equals(DOWNGRADE_VERSION)) {
+        assertEquals(RepositoryVersionState.CURRENT, entity.getState());
+      } else if (entity.getRepositoryVersion().getVersion().equals(UPGRADE_VERSION)) {
+        assertEquals(RepositoryVersionState.INSTALLED, entity.getState());
+      }
+    }
+
+    ClusterVersionDAO clusterVersionDao = m_injector.getInstance(ClusterVersionDAO.class);
+    for (ClusterVersionEntity entity : clusterVersionDao.findByCluster("c1")) {
+      if (entity.getRepositoryVersion().getVersion().equals(DOWNGRADE_VERSION)) {
+        assertEquals(RepositoryVersionState.CURRENT, entity.getState());
+      } else if (entity.getRepositoryVersion().getVersion().equals(UPGRADE_VERSION)) {
+        assertEquals(RepositoryVersionState.INSTALLED, entity.getState());
+      }
+    }
+
+  }
+
+
+}