AMBARI-8593. Upgrade Execute - Add pre-restart and post-restart hooks to perform hdp-select with version (alejandro)

ambari-common/src/main/python/resource_management/core/shell.py

@@ -33,15 +33,15 @@ from resource_management.core.logger import Logger
 
 SUDO_ENVIRONMENT_PLACEHOLDER = "{ENV_PLACEHOLDER}"
 
-def checked_call(command, logoutput=False, 
+def checked_call(command, verbose=False, logoutput=False,
          cwd=None, env={}, preexec_fn=None, user=None, wait_for_finish=True, timeout=None, path=None, output_file=None, sudo=False):
-  return _call(command, logoutput, True, cwd, env, preexec_fn, user, wait_for_finish, timeout, path, output_file, sudo)
+  return _call(command, verbose, logoutput, True, cwd, env, preexec_fn, user, wait_for_finish, timeout, path, output_file, sudo)
 
-def call(command, logoutput=False, 
+def call(command, verbose=False, logoutput=False,
          cwd=None, env={}, preexec_fn=None, user=None, wait_for_finish=True, timeout=None, path=None, output_file=None, sudo=False):
-  return _call(command, logoutput, False, cwd, env, preexec_fn, user, wait_for_finish, timeout, path, output_file, sudo)
+  return _call(command, verbose, logoutput, False, cwd, env, preexec_fn, user, wait_for_finish, timeout, path, output_file, sudo)
             
-def _call(command, logoutput=False, throw_on_failure=True, 
+def _call(command, verbose=False, logoutput=False, throw_on_failure=True,
          cwd=None, env={}, preexec_fn=None, user=None, wait_for_finish=True, timeout=None, path=None, output_file=None, sudo=False):
   """
   Execute shell command
@@ -74,7 +74,9 @@ def _call(command, logoutput=False, throw_on_failure=True,
     command = string_cmd_from_args_list(command)
   # replace placeholder from as_sudo / as_user if present
   command = command.replace(SUDO_ENVIRONMENT_PLACEHOLDER, get_environment_str(env), 1)
-  
+  if verbose:
+    Logger.info("Call command: " + Logger.get_protected_text(command))
+
   subprocess_command = ["/bin/bash","--login","-c", command]
   proc = subprocess.Popen(subprocess_command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                           cwd=cwd, env=env, shell=False,

ambari-common/src/main/python/resource_management/libraries/functions/decorator.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+"""
+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.
+
+Ambari Agent
+
+"""
+
+import time
+__all__ = ['retry', ]
+
+from resource_management.core.logger import Logger
+
+
+def retry(times=3, sleep_time=1, backoff_factor=1, err_class=Exception):
+  """
+  Retry decorator for improved robustness of functions.
+  :param times: Number of times to attempt to call the function.
+  :param sleep_time: Initial sleep time between attempts
+  :param backoff_factor: After every failed attempt, multiple the previous sleep time by this factor.
+  :param err_class: Exception class to handle
+  :return: Returns the output of the wrapped function.
+  """
+  def decorator(function):
+    def wrapper(*args, **kwargs):
+      _times = times
+      _sleep_time = sleep_time
+      _backoff_factor = backoff_factor
+      _err_class = err_class
+
+      while _times > 1:
+        _times -= 1
+        _sleep_time *= _backoff_factor
+        try:
+          return function(*args, **kwargs)
+        except _err_class, err:
+          Logger.info("Will retry %d time(s), caught exception: %s. Sleeping for %d sec(s)" % (_times, str(err), _sleep_time))
+          time.sleep(_sleep_time)
+
+      return function(*args, **kwargs)
+    return wrapper
+  return decorator
+

ambari-common/src/main/python/resource_management/libraries/functions/version.py

@@ -36,10 +36,14 @@ def _normalize(v, desired_segments=0):
 
 def format_hdp_stack_version(input):
   """
-  :param input: Input string, e.g. "2.2" or "GlusterFS", or "2.0.6.GlusterFS"
+  :param input: Input string, e.g. "2.2" or "GlusterFS", or "2.0.6.GlusterFS", or "2.2.0.1-885"
   :return: Returns a well-formatted HDP stack version of the form #.#.#.# as a string.
   """
   if input:
+    if "-" in input:
+      input_array = input.split("-")
+      input = input_array[0]
+
     input = re.sub(r'^\D+', '', input)
     input = re.sub(r'\D+$', '', input)
     input = input.strip('.')

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

@@ -256,6 +256,12 @@ class Script(object):
     """
     self.fail_with_error('stop method isn\'t implemented')
 
+  def pre_rolling_restart(self, env):
+    """
+    To be overridden by subclasses
+    """
+    pass
+
   def restart(self, env):
     """
     Default implementation of restart command is to call stop and start methods
@@ -272,9 +278,32 @@ class Script(object):
     if componentCategory and componentCategory.strip().lower() == 'CLIENT'.lower():
       self.install(env)
     else:
+      restart_type = ""
+      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 ""
+          if restart_type:
+            restart_type = restart_type.encode('ascii', 'ignore')
+
+      rolling_restart = restart_type.lower().startswith("rolling")
+
       self.stop(env)
+
+      if rolling_restart:
+        self.pre_rolling_restart(env)
+
       self.start(env)
 
+      if rolling_restart:
+        self.post_rolling_restart(env)
+
+  def post_rolling_restart(self, env):
+    """
+    To be overridden by subclasses
+    """
+    pass
+
   def configure(self, env):
     """
     To be overridden by subclasses

ambari-server/src/main/java/org/apache/ambari/server/controller/internal/UpgradeResourceProvider.java

@@ -28,6 +28,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.TreeMap;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -361,7 +362,8 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
     List<UpgradeGroupHolder> groups = helper.createUpgrade(cluster, pack);
     List<UpgradeGroupEntity> groupEntities = new ArrayList<UpgradeGroupEntity>();
 
-    RequestStageContainer req = createRequest((String) requestMap.get(UPGRADE_VERSION));
+    final String version = (String) requestMap.get(UPGRADE_VERSION);
+    RequestStageContainer req = createRequest(version);
 
     for (UpgradeGroupHolder group : groups) {
       UpgradeGroupEntity groupEntity = new UpgradeGroupEntity();
@@ -378,7 +380,7 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
         itemEntities.add(itemEntity);
 
         // upgrade items match a stage
-        createStage(cluster, req, itemEntity, wrapper);
+        createStage(cluster, req, version, itemEntity, wrapper);
       }
 
       itemEntities = injectVariables(configHelper, cluster, itemEntities);
@@ -415,18 +417,18 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
     return requestStages;
   }
 
-  private void createStage(Cluster cluster, RequestStageContainer request,
+  private void createStage(Cluster cluster, RequestStageContainer request, final String version,
       UpgradeItemEntity entity, StageWrapper wrapper) throws AmbariException {
 
     if (wrapper.hasCommand()) {
-      makeRestartStage(cluster, request, entity, wrapper);
+      makeRestartStage(cluster, request, version, entity, wrapper);
     } else {
-      makeActionStage(cluster, request, entity, wrapper);
+      makeActionStage(cluster, request, version, entity, wrapper);
     }
 
   }
 
-  private void makeActionStage(Cluster cluster, RequestStageContainer request,
+  private void makeActionStage(Cluster cluster, RequestStageContainer request, final String version,
       UpgradeItemEntity entity, StageWrapper wrapper) throws AmbariException {
 
     Map<String, String> hostLevelParams = new HashMap<String, String>();
@@ -475,7 +477,7 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
     request.addStages(Collections.singletonList(stage));
   }
 
-  private void makeRestartStage(Cluster cluster, RequestStageContainer request,
+  private void makeRestartStage(Cluster cluster, RequestStageContainer request, final String version,
       UpgradeItemEntity entity, StageWrapper wrapper) throws AmbariException {
 
     List<RequestResourceFilter> filters = new ArrayList<RequestResourceFilter>();
@@ -486,10 +488,14 @@ public class UpgradeResourceProvider extends AbstractControllerResourceProvider
           new ArrayList<String>(tw.getHosts())));
     }
 
+    Map<String, String> restartCommandParams = new HashMap<String, String>();
+    restartCommandParams.put("restart_type", "rolling_upgrade");
+    restartCommandParams.put("version", version);
+
     ActionExecutionContext actionContext = new ActionExecutionContext(
         cluster.getClusterName(), "RESTART",
         filters,
-        Collections.<String, String>emptyMap());
+        restartCommandParams);
     actionContext.setTimeout(Short.valueOf((short)-1));
 
     ExecuteCommandJson jsons = commandExecutionHelper.get().getCommandJson(

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HDFS/package/scripts/datanode.py

@@ -17,8 +17,9 @@ limitations under the License.
 
 """
 
-from resource_management import *
 from hdfs_datanode import datanode
+from resource_management import *
+from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
 from hdfs import hdfs
 
 
@@ -29,6 +30,15 @@ class DataNode(Script):
     self.install_packages(env, params.exclude_packages)
     env.set_params(params)
 
+  def pre_rolling_restart(self, env):
+    Logger.info("Executing Rolling Upgrade pre-restart")
+    import params
+    env.set_params(params)
+
+    version = default("/commandParams/version", None)
+    if version and compare_versions(format_hdp_stack_version(version), '2.2.0.0') >= 0:
+      Execute(format("hdp-select set hadoop-hdfs-datanode {version}"))
+
   def start(self, env):
     import params
 

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HDFS/package/scripts/namenode.py

@@ -45,6 +45,24 @@ class NameNode(Script):
     self.configure(env)
     namenode(action="start")
 
+  def pre_rolling_restart(self, env):
+    Logger.info("Executing Rolling Upgrade pre-restart")
+    import params
+    env.set_params(params)
+
+    version = default("/commandParams/version", None)
+    if version and compare_versions(format_hdp_stack_version(version), '2.2.0.0') >= 0:
+      Execute(format("hdp-select set hadoop-hdfs-namenode {version}"))
+
+  def post_rolling_restart(self, env):
+    Logger.info("Executing Rolling Upgrade post-restart")
+    import params
+    env.set_params(params)
+
+    Execute("hdfs dfsadmin -report -live",
+            user=params.hdfs_principal_name if params.security_enabled else params.hdfs_user
+    )
+
   def stop(self, env):
     import params
 

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HDFS/package/scripts/utils.py

@@ -97,15 +97,6 @@ def service(action=None, name=None, user=None, create_pid_dir=False,
           environment=hadoop_env_exports
   )
 
-  #After performing the desired action, perform additional tasks.
-  # Only needed during Rolling Upgrade
-  '''
-  if action == "start":
-    Execute("hdfs dfsadmin -report -live",
-            user=params.hdfs_principal_name if params.security_enabled else params.hdfs_user
-    )
-  '''
-
   if action == "stop":
     File(pid_file,
          action="delete",

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/ZOOKEEPER/package/scripts/zookeeper_server.py

@@ -20,26 +20,71 @@ Ambari Agent
 """
 
 import sys
+import re
 from resource_management import *
+from resource_management.libraries.functions.format import format
+from resource_management.libraries.functions import get_unique_id_and_date
+from resource_management.libraries.functions.decorator import retry
+from resource_management.libraries.functions.version import compare_versions, format_hdp_stack_version
+from resource_management.core.shell import call
+
 
 from zookeeper import zookeeper
 from zookeeper_service import zookeeper_service
 
+@retry(times=10, sleep_time=2, err_class=Fail)
+def call_and_match_output(command, regex_expression, err_message):
+  """
+  Call the command and performs a regex match on the output for the specified expression.
+  :param command: Command to call
+  :param regex_expression: Regex expression to search in the output
+  """
+  code, out = call(command, verbose=True)
+  if not (out and re.search(regex_expression, out)):
+    raise Fail(err_message)
+
+
 class ZookeeperServer(Script):
   def install(self, env):
     self.install_packages(env)
     self.configure(env)
+
   def configure(self, env):
     import params
     env.set_params(params)
     zookeeper(type='server')
 
+  def pre_rolling_restart(self, env):
+    Logger.info("Executing Rolling Upgrade pre-restart")
+    import params
+    env.set_params(params)
+
+    version = default("/commandParams/version", None)
+    if version and compare_versions(format_hdp_stack_version(version), '2.2.0.0') >= 0:
+      Execute(format("hdp-select set zookeeper-server {version}"))
+
   def start(self, env):
     import params
     env.set_params(params)
     self.configure(env)
     zookeeper_service(action = 'start')
 
+  def post_rolling_restart(self, env):
+    Logger.info("Executing Rolling Upgrade post-restart")
+    import params
+    env.set_params(params)
+
+    # Ensure that a quorum is still formed.
+    unique = get_unique_id_and_date()
+    create_command = format("echo 'create /{unique} mydata' | {zk_cli_shell}")
+    list_command = format("echo 'ls /' | {zk_cli_shell}")
+    delete_command = format("echo 'delete /{unique} ' | {zk_cli_shell}")
+
+    quorum_err_message = "Failed to establish zookeeper quorum"
+    call_and_match_output(create_command, 'Created', quorum_err_message)
+    call_and_match_output(list_command, r"\[.*?" + unique + ".*?\]", quorum_err_message)
+    call(delete_command)
+
   def stop(self, env):
     import params
     env.set_params(params)

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/ZOOKEEPER/package/scripts/zookeeper_service.py

@@ -41,20 +41,6 @@ def zookeeper_service(action='start'):
               user=params.smokeuser
       )
 
-    # Ensure that a quorum is still formed.
-    create_command = format("echo 'create /zk_test mydata' | {zk_cli_shell}")
-    list_command = format("echo 'ls /' | {zk_cli_shell}")
-    delete_command = format("echo 'delete /zk_test ' | {zk_cli_shell}")
-    # FIXME: this makes ZK service start to fail
-    """
-    Execute(create_command,
-            user=params.smokeuser)
-    Execute(list_command,
-            user=params.smokeuser)
-    Execute(delete_command,
-            user=params.smokeuser)
-    """
-
   elif action == 'stop':
     daemon_cmd = format("source {config_dir}/zookeeper-env.sh ; {cmd} stop")
     rm_pid = format("rm -f {zk_pid_file}")

ambari-server/src/main/resources/stacks/HDP/2.2/upgrades/upgrade-2.2.xml

@@ -200,14 +200,8 @@
             <command>su {{hadoop-env/hdfs_user}} -c 'hdfs dfsadmin -getDatanodeInfo {{hdfs-site/dfs.datanode.ipc.address}}'</command>
             <ignore>255</ignore>
           </task>
-          
-
-          <!-- TODO, move this to HDFS Datanode restart. -->
-          <task xsi:type="execute">
-            <command>hdp-select set hadoop-hdfs-datanode {{version}}</command>
-          </task>
-          
         </pre-upgrade>
+
         <upgrade>
           <task xsi:type="restart" />
         </upgrade>