AMBARI-14041. RU: Downgrade does not restart failed service component (ncole)

ambari-common/src/main/python/resource_management/libraries/script/script.py

@@ -131,7 +131,6 @@ class Script(object):
       del Script.structuredOut["securityStateErrorInfo"]
 
   def put_structured_out(self, sout):
-    curr_content = Script.structuredOut.copy()
     Script.structuredOut.update(sout)
     try:
       with open(self.stroutfile, 'w') as fp:
@@ -462,10 +461,12 @@ class Script(object):
       pass
 
     restart_type = ""
+    direction = None
     if config is not None:
       command_params = config["commandParams"] if "commandParams" in config else None
       if command_params is not None:
         restart_type = command_params["restart_type"] if "restart_type" in command_params else ""
+        direction = command_params["upgrade_direction"] if "upgrade_direction" in command_params else None
 
     upgrade_type = None
     if restart_type.lower() == "rolling_upgrade":
@@ -475,6 +476,14 @@ class Script(object):
 
     is_stack_upgrade = upgrade_type is not None
 
+    # need this before actually executing so that failures still report upgrade info
+    if is_stack_upgrade:
+      upgrade_info = {"upgrade_type": restart_type}
+      if direction is not None:
+        upgrade_info["direction"] = direction.upper()
+
+      Script.structuredOut.update(upgrade_info)
+
     if componentCategory and componentCategory.strip().lower() == 'CLIENT'.lower():
       if is_stack_upgrade:
         # Remain backward compatible with the rest of the services that haven't switched to using
@@ -525,6 +534,7 @@ class Script(object):
     if self.should_expose_component_version("restart"):
       self.save_component_version_to_structured_out()
 
+
   # TODO, remove after all services have switched to post_upgrade_restart
   def post_rolling_restart(self, env):
     """

ambari-server/src/main/java/org/apache/ambari/server/agent/HeartBeatHandler.java

@@ -86,6 +86,8 @@ import org.apache.ambari.server.state.host.HostRegistrationRequestEvent;
 import org.apache.ambari.server.state.host.HostStatusUpdatesReceivedEvent;
 import org.apache.ambari.server.state.host.HostUnhealthyHeartbeatEvent;
 import org.apache.ambari.server.state.scheduler.RequestExecution;
+import org.apache.ambari.server.state.stack.upgrade.Direction;
+import org.apache.ambari.server.state.stack.upgrade.UpgradeType;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpFailedEvent;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
 import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpSucceededEvent;
@@ -558,7 +560,7 @@ public class HeartBeatHandler {
                 //do nothing, pass this data further for processing
               }
 
-              String newVersion = structuredOutput == null ? null : structuredOutput.getVersion();
+              String newVersion = structuredOutput == null ? null : structuredOutput.version;
 
               // Pass true to always publish a version event.  It is safer to recalculate the version even if we don't
               // detect a difference in the value.  This is useful in case that a manual database edit is done while
@@ -566,7 +568,7 @@ public class HeartBeatHandler {
               handleComponentVersionReceived(cl, scHost, newVersion, true);
             }
 
-            // Updating stack version, if needed
+            // Updating stack version, if needed (this is not actually for express/rolling upgrades!)
             if (scHost.getState().equals(State.UPGRADING)) {
               scHost.setStackVersion(scHost.getDesiredStackVersion());
             } else if ((report.getRoleCommand().equals(RoleCommand.START.toString()) ||
@@ -611,6 +613,19 @@ public class HeartBeatHandler {
                   hostname, now));
             }
           } else if (report.getStatus().equals("FAILED")) {
+
+            if (StringUtils.isNotBlank(report.getStructuredOut())) {
+              try {
+                ComponentVersionStructuredOut structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
+
+                if (null != structuredOutput.upgradeDirection && structuredOutput.upgradeDirection.isUpgrade()) {
+                  scHost.setUpgradeState(UpgradeState.FAILED);
+                }
+              } catch (JsonSyntaxException ex) {
+                LOG.warn("Structured output was found, but not parseable: {}", report.getStructuredOut());
+              }
+            }
+
             LOG.warn("Operation failed - may be retried. Service component host: "
                 + schName + ", host: " + hostname + " Action id" + report.getActionId());
             if (actionManager.isInProgressCommand(report)) {
@@ -1184,13 +1199,12 @@ public class HeartBeatHandler {
     @SerializedName("version")
     private String version;
 
-    public String getVersion() {
-      return version;
-    }
+    @SerializedName("upgrade_type")
+    private UpgradeType upgradeType = null;
+
+    @SerializedName("direction")
+    private Direction upgradeDirection = null;
 
-    public void setVersion(String version) {
-      this.version = version;
-    }
   }
 
   /**

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariManagementControllerImpl.java

@@ -1432,7 +1432,7 @@ public class AmbariManagementControllerImpl implements AmbariManagementControlle
             break;
           } else {
             for (Entry<String, String> property : requestConfigProperties.entrySet()) {
-              if (!property.getValue().equals(clusterConfigProperties.get(property.getKey()))) {
+              if (!StringUtils.equals(property.getValue(),clusterConfigProperties.get(property.getKey()))) {
                 isConfigurationCreationNeeded =true;
                 break;
               }

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

@@ -18,7 +18,11 @@
 package org.apache.ambari.server.serveraction.upgrades;
 import static org.apache.ambari.server.agent.ExecutionCommand.KeyNames.VERSION;
 
-import com.google.inject.Inject;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
 import org.apache.ambari.server.agent.CommandReport;
@@ -33,11 +37,7 @@ import org.apache.ambari.server.state.StackInfo;
 import org.apache.ambari.server.state.stack.UpgradePack;
 import org.apache.ambari.server.state.stack.upgrade.Direction;
 
-import java.io.PrintWriter;
-import java.io.StringWriter;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentMap;
+import com.google.inject.Inject;
 
 /**
  * Action that represents updating the Desired Stack Id during the middle of a stack upgrade (typically NonRolling).
@@ -113,7 +113,7 @@ public class UpdateDesiredStackAction extends AbstractServerAction {
     try {
       Cluster cluster = clusters.getCluster(clusterName);
       StackId currentClusterStackId = cluster.getCurrentStackVersion();
-      out.append(String.format("Params: %s %s %s %s %s %s",
+      out.append(String.format("Params: %s %s %s %s %s %s\n",
           clusterName, originalStackId.getStackId(), targetStackId.getStackId(), version, direction.getText(false), upgradePack.getName()));
 
       out.append(String.format("Checking if can update the Desired Stack Id to %s. The cluster's current Stack Id is %s\n", targetStackId.getStackId(), currentClusterStackId.getStackId()));

ambari-server/src/main/java/org/apache/ambari/server/stack/MasterHostResolver.java

@@ -35,6 +35,7 @@ import org.apache.ambari.server.state.Host;
 import org.apache.ambari.server.state.MaintenanceState;
 import org.apache.ambari.server.state.ServiceComponent;
 import org.apache.ambari.server.state.ServiceComponentHost;
+import org.apache.ambari.server.state.UpgradeState;
 import org.apache.ambari.server.utils.HTTPUtils;
 import org.apache.ambari.server.utils.HostAndPort;
 import org.apache.ambari.server.utils.StageUtils;
@@ -206,7 +207,8 @@ public class MasterHostResolver {
         // possible
         if (maintenanceState != MaintenanceState.OFF) {
           unhealthyHosts.add(sch);
-        } else if (null == m_version || null == sch.getVersion() || !sch.getVersion().equals(m_version)) {
+        } else if (null == m_version || null == sch.getVersion() ||
+            !sch.getVersion().equals(m_version) || sch.getUpgradeState() == UpgradeState.FAILED) {
           upgradeHosts.add(hostName);
         }
       }

ambari-server/src/main/java/org/apache/ambari/server/state/ServiceComponentHost.java

@@ -136,8 +136,16 @@ public interface ServiceComponentHost {
    */
   public void setDesiredSecurityState(SecurityState securityState) throws AmbariException;
 
+  /**
+   * @param upgradeState the upgrade state
+   */
   public void setUpgradeState(UpgradeState upgradeState);
 
+  /**
+   * @return the upgrade state
+   */
+  public UpgradeState getUpgradeState();
+
   public StackId getStackVersion();
 
   public void setStackVersion(StackId stackVersion);

ambari-server/src/main/java/org/apache/ambari/server/state/stack/upgrade/UpgradeType.java

@@ -19,6 +19,8 @@ package org.apache.ambari.server.state.stack.upgrade;
 
 import javax.xml.bind.annotation.XmlEnumValue;
 
+import com.google.gson.annotations.SerializedName;
+
 /**
  * Indicates the type of Upgrade performed.
  */
@@ -27,10 +29,13 @@ public enum UpgradeType {
    * Services are up the entire time
    */
   @XmlEnumValue("ROLLING")
+  @SerializedName("rolling_upgrade")
   ROLLING,
+
   /**
    * All services are stopped, then started
    */
   @XmlEnumValue("NON_ROLLING")
+  @SerializedName("nonrolling_upgrade")
   NON_ROLLING;
 }

ambari-server/src/main/java/org/apache/ambari/server/state/svccomphost/ServiceComponentHostImpl.java

@@ -20,7 +20,6 @@ package org.apache.ambari.server.state.svccomphost;
 
 import java.text.MessageFormat;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -78,7 +77,6 @@ import org.apache.ambari.server.state.fsm.SingleArcTransition;
 import org.apache.ambari.server.state.fsm.StateMachine;
 import org.apache.ambari.server.state.fsm.StateMachineFactory;
 import org.apache.ambari.server.state.stack.upgrade.RepositoryVersionHelper;
-import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -173,8 +171,8 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
   // define the state machine of a HostServiceComponent for runnable
   // components
 
-  .addTransition(State.INIT, 
-      State.INSTALLING, 
+  .addTransition(State.INIT,
+      State.INSTALLING,
       ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
       new ServiceComponentHostOpStartedTransition())
 
@@ -198,12 +196,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
       new ServiceComponentHostOpInProgressTransition())
 
-  .addTransition(State.INSTALLING, 
+  .addTransition(State.INSTALLING,
       State.INSTALLING,
       ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLING, 
+  .addTransition(State.INSTALLING,
       State.INSTALL_FAILED,
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
       new ServiceComponentHostOpCompletedTransition())
@@ -213,49 +211,49 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALL_FAILED, 
+  .addTransition(State.INSTALL_FAILED,
       State.INSTALLING,
       ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
       new ServiceComponentHostOpStartedTransition())
 
   // Allow transition on abort
-  .addTransition(State.INSTALL_FAILED, 
+  .addTransition(State.INSTALL_FAILED,
       State.INSTALL_FAILED,
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
       new ServiceComponentHostOpCompletedTransition())
 
-  .addTransition(State.INSTALLED, 
-      State.STARTING, 
+  .addTransition(State.INSTALLED,
+      State.STARTING,
       ServiceComponentHostEventType.HOST_SVCCOMP_START,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLED, 
+  .addTransition(State.INSTALLED,
       State.UNINSTALLING,
       ServiceComponentHostEventType.HOST_SVCCOMP_UNINSTALL,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLED, 
+  .addTransition(State.INSTALLED,
       State.INSTALLING,
       ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLED, 
+  .addTransition(State.INSTALLED,
       State.STOPPING,
       ServiceComponentHostEventType.HOST_SVCCOMP_STOP,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLED, 
+  .addTransition(State.INSTALLED,
       State.UPGRADING,
       ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
       new ServiceComponentHostOpStartedTransition())
 
-  .addTransition(State.INSTALLED, 
+  .addTransition(State.INSTALLED,
       State.INSTALLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
       new ServiceComponentHostOpInProgressTransition())
 
   .addTransition(State.INSTALLED,
-      State.STARTED, 
+      State.STARTED,
       ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
       new ServiceComponentHostOpCompletedTransition())
 
@@ -270,12 +268,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       new ServiceComponentHostOpInProgressTransition())
 
   .addTransition(State.STARTING,
-      State.STARTING, 
+      State.STARTING,
       ServiceComponentHostEventType.HOST_SVCCOMP_START,
       new ServiceComponentHostOpStartedTransition())
 
   .addTransition(State.STARTING,
-      State.STARTED, 
+      State.STARTED,
       ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
       new ServiceComponentHostOpCompletedTransition())
 
@@ -290,12 +288,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       new ServiceComponentHostOpStartedTransition())
 
   .addTransition(State.STARTED,
-      State.STARTED, 
+      State.STARTED,
       ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
       new ServiceComponentHostOpCompletedTransition())
 
   .addTransition(State.STARTED,
-      State.STOPPING, 
+      State.STOPPING,
       ServiceComponentHostEventType.HOST_SVCCOMP_STOP,
       new ServiceComponentHostOpStartedTransition())
 
@@ -305,7 +303,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       new ServiceComponentHostOpInProgressTransition())
 
   .addTransition(State.STARTED,
-      State.INSTALLED, 
+      State.INSTALLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_STOPPED,
       new ServiceComponentHostOpCompletedTransition())
 
@@ -349,7 +347,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
       new ServiceComponentHostOpInProgressTransition())
 
-  .addTransition(State.UPGRADING, 
+  .addTransition(State.UPGRADING,
       State.INSTALLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
       new ServiceComponentHostOpCompletedTransition())
@@ -420,12 +418,12 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       new ServiceComponentHostOpCompletedTransition())
 
   .addTransition(State.DISABLED,
-      State.DISABLED, 
+      State.DISABLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
       new ServiceComponentHostOpCompletedTransition())
 
   .addTransition(State.UNKNOWN,
-      State.DISABLED, 
+      State.DISABLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
       new ServiceComponentHostOpCompletedTransition())
 
@@ -434,7 +432,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
       ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
       new ServiceComponentHostOpCompletedTransition())
 
-  .addTransition(State.INSTALL_FAILED, 
+  .addTransition(State.INSTALL_FAILED,
       State.DISABLED,
       ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
       new ServiceComponentHostOpCompletedTransition())
@@ -949,7 +947,7 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
     }
   }
 
-  /***
+  /**
    * To be called during the upgrade of a specific Component in a host.
    * The potential upgrade states are NONE (default), PENDING, IN_PROGRESS, FAILED.
    * If the upgrade completes successfully, the upgradeState should be set back to NONE.
@@ -978,6 +976,29 @@ public class ServiceComponentHostImpl implements ServiceComponentHost {
     }
   }
 
+  @Override
+  public UpgradeState getUpgradeState() {
+    readLock.lock();
+
+    try {
+      HostComponentStateEntity stateEntity = getStateEntity();
+      if (stateEntity != null) {
+        return stateEntity.getUpgradeState();
+      } else {
+        LOG.warn("Trying to fetch a state entity from an object that may " +
+          "have been previously deleted, serviceName = " + getServiceName() + ", " +
+          "componentName = " + getServiceComponentName() + ", " +
+          "hostName = " + getHostName());
+      }
+
+    } finally {
+      readLock.unlock();
+    }
+
+    return UpgradeState.NONE;
+  }
+
+
   @Override
   public void handleEvent(ServiceComponentHostEvent event)
       throws InvalidStateTransitionException {

ambari-server/src/test/java/org/apache/ambari/server/state/UpgradeHelperTest.java

@@ -60,7 +60,6 @@ import org.apache.ambari.server.state.stack.upgrade.UpgradeType;
 import org.easymock.EasyMock;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import com.google.gson.Gson;
@@ -1092,6 +1091,67 @@ public class UpgradeHelperTest {
     assertEquals(1, task.getHosts().size());
   }
 
+  @Test
+  public void testResolverWithFailedUpgrade() throws Exception {
+    Clusters clusters = injector.getInstance(Clusters.class);
+    ServiceFactory serviceFactory = injector.getInstance(ServiceFactory.class);
+
+    String clusterName = "c1";
+
+    StackId stackId = new StackId("HDP-2.1.1");
+    clusters.addCluster(clusterName, stackId);
+    Cluster c = clusters.getCluster(clusterName);
+
+    helper.getOrCreateRepositoryVersion(stackId,
+        c.getDesiredStackVersion().getStackVersion());
+
+    c.createClusterVersion(stackId,
+        c.getDesiredStackVersion().getStackVersion(), "admin",
+        RepositoryVersionState.UPGRADING);
+
+    for (int i = 0; i < 2; i++) {
+      String hostName = "h" + (i+1);
+      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();
+      clusters.mapHostToCluster(hostName, clusterName);
+    }
+
+    // !!! add services
+    c.addService(serviceFactory.createNew(c, "ZOOKEEPER"));
+
+    Service s = c.getService("ZOOKEEPER");
+    ServiceComponent sc = s.addServiceComponent("ZOOKEEPER_SERVER");
+
+    ServiceComponentHost sch1 =sc.addServiceComponentHost("h1");
+    sch1.setVersion("2.1.1.0-1234");
+
+    ServiceComponentHost sch2 = sc.addServiceComponentHost("h2");
+    sch2.setVersion("2.1.1.0-1234");
+
+    List<ServiceComponentHost> schs = c.getServiceComponentHosts("ZOOKEEPER", "ZOOKEEPER_SERVER");
+    assertEquals(2, schs.size());
+
+    MasterHostResolver mhr = new MasterHostResolver(null, c, "2.1.1.0-1234");
+
+    HostsType ht = mhr.getMasterAndHosts("ZOOKEEPER", "ZOOKEEPER_SERVER");
+    assertEquals(0, ht.hosts.size());
+
+    // !!! if one of them is failed, it should be scheduled
+    sch2.setUpgradeState(UpgradeState.FAILED);
+
+    ht = mhr.getMasterAndHosts("ZOOKEEPER", "ZOOKEEPER_SERVER");
+
+    assertEquals(1, ht.hosts.size());
+    assertEquals("h2", ht.hosts.iterator().next());
+  }
 
 
   private class MockModule implements Module {

ambari-server/src/test/python/stacks/2.0.6/HDFS/test_namenode.py

@@ -24,7 +24,7 @@ import tempfile
 import time
 from stacks.utils.RMFTestCase import *
 from mock.mock import MagicMock, patch, call
-import resource_management
+from resource_management.libraries.script.script import Script
 from resource_management.core import shell
 from resource_management.core.exceptions import Fail
 
@@ -1395,6 +1395,13 @@ class TestNamenode(RMFTestCase):
                      hdp_stack_version = self.STACK_VERSION,
                      target = RMFTestCase.TARGET_COMMON_SERVICES)
 
+    self.assertFalse(0 == len(Script.structuredOut))
+    self.assertTrue(Script.structuredOut.has_key("upgrade_type"))
+    self.assertTrue(Script.structuredOut.has_key("direction"))
+    self.assertEquals("rolling_upgrade", Script.structuredOut["upgrade_type"])
+    self.assertEquals("UPGRADE", Script.structuredOut["direction"])
+
+
   @patch("utils.get_namenode_states")
   def test_upgrade_restart_eu(self, get_namenode_states_mock):
     active_namenodes = [('nn1', 'c6401.ambari.apache.org:50070')]