AMBARI-12083. Installing Repo Packages needs to be more robust to handle the actual_version installed when the script is killed because of a timeout (alejandro)

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/datanode.py

@@ -52,8 +52,8 @@ class DataNode(Script):
     # pre-upgrade steps shutdown the datanode, so there's no need to call
     # action=stop
     if rolling_restart:
-      force_stop = datanode_upgrade.pre_upgrade_shutdown()
-      if force_stop:
+      stopped = datanode_upgrade.pre_upgrade_shutdown()
+      if not stopped:
         datanode(action="stop")
     else:
       datanode(action="stop")

ambari-server/src/main/resources/common-services/HDFS/2.1.0.2.0/package/scripts/datanode_upgrade.py

@@ -49,7 +49,7 @@ def pre_upgrade_shutdown():
   else:
     # Due to bug HDFS-7533, DataNode may not always shutdown during rolling upgrade, and it is necessary to kill it.
     if output is not None and re.search("Shutdown already in progress", output):
-      Logger.error("Due to a known issue in DataNode, the command {0} did not work and will shutdown the datanode forcefully.")
+      Logger.error("Due to a known issue in DataNode, the command {0} did not work, so will need to shutdown the datanode forcefully.".format(command))
       return False
   return True
 

ambari-server/src/main/resources/custom_actions/scripts/install_packages.py

@@ -19,17 +19,22 @@ limitations under the License.
 Ambari Agent
 
 """
+import os
+import signal
 
-import ambari_simplejson as json # simplejson is much faster comparing to Python 2.6 json module and has the same functions set.
 import sys
 import re
-import traceback
+import os.path
+
+import ambari_simplejson as json # simplejson is much faster comparing to Python 2.6 json module and has the same functions set.
+
 from resource_management import *
 from resource_management.libraries.functions.list_ambari_managed_repos import list_ambari_managed_repos
 from ambari_commons.os_check import OSCheck, OSConst
 from resource_management.libraries.functions.packages_analyzer import allInstalledPackages
 from resource_management.core.shell import call
-from resource_management.libraries.functions.default import default
+
+from resource_management.core.logger import Logger
 
 
 class InstallPackages(Script):
@@ -42,112 +47,238 @@ class InstallPackages(Script):
 
   UBUNTU_REPO_COMPONENTS_POSTFIX = ["main"]
   REPO_FILE_NAME_PREFIX = 'HDP-'
+  
+  # Mapping file used to store repository versions without a build number, and the actual version it corresponded to.
+  # E.g., HDP 2.2.0.0 => HDP 2.2.0.0-2041
+  REPO_VERSION_HISTORY_FILE = "/var/lib/ambari-agent/data/repo_version_history.json"
 
   def actionexecute(self, env):
     num_errors = 0
-    package_install_result = False
 
     # Parse parameters
     config = Script.get_config()
 
+    # Handle a SIGTERM and SIGINT gracefully
+    signal.signal(signal.SIGTERM, self.abort_handler)
+    signal.signal(signal.SIGINT, self.abort_handler)
+
     # Select dict that contains parameters
     try:
-      repository_version = config['roleParams']['repository_version']
+      self.repository_version = config['roleParams']['repository_version']
       base_urls = json.loads(config['roleParams']['base_urls'])
       package_list = json.loads(config['roleParams']['package_list'])
       stack_id = config['roleParams']['stack_id']
     except KeyError:
       # Last try
-      repository_version = config['commandParams']['repository_version']
+      self.repository_version = config['commandParams']['repository_version']
       base_urls = json.loads(config['commandParams']['base_urls'])
       package_list = json.loads(config['commandParams']['package_list'])
       stack_id = config['commandParams']['stack_id']
 
+    self.repository_version = self.repository_version.strip()
+
     # Install/update repositories
     installed_repositories = []
-    current_repositories = ['base']
+    self.current_repositories = ['base']
     # Some packages are installed from core repos
     if OSCheck.is_redhat_family():
       # rhui* is used for rhel, and base is for centos
-      current_repositories.append('rhui*')
-    current_repo_files = set(['base'])
-    old_versions = self.hdp_versions()
-    
+      self.current_repositories.append('rhui*')
+    self.current_repo_files = set(['base'])
+
+    Logger.info("Will install packages for repository version {0}".format(self.repository_version))
     try:
       append_to_file = False
       for url_info in base_urls:
-        repo_name, repo_file = self.install_repository(url_info, repository_version, append_to_file)
-        current_repositories.append(repo_name)
-        current_repo_files.add(repo_file)
+        repo_name, repo_file = self.install_repository(url_info, append_to_file)
+        self.current_repositories.append(repo_name)
+        self.current_repo_files.add(repo_file)
         append_to_file = True
 
       installed_repositories = list_ambari_managed_repos()
     except Exception, err:
-      print "Cannot distribute repositories."
-      print traceback.format_exc()
+      Logger.logger.exception("Cannot distribute repositories. Error: {0}".format(str(err)))
       num_errors += 1
 
     # Build structured output with initial values
-    structured_output = {
+    self.structured_output = {
       'ambari_repositories': installed_repositories,
-      'installed_repository_version': repository_version,
-      'stack_id': stack_id
+      'installed_repository_version': self.repository_version,
+      'stack_id': stack_id,
+      'package_installation_result': 'FAIL'
     }
+    self.put_structured_out(self.structured_output)
 
-    # Install packages
-    if not num_errors:
-      packages_were_checked = False
-      try:
-        packages_installed_before = []
-        allInstalledPackages(packages_installed_before)
-        packages_installed_before = [package[0] for package in packages_installed_before]
-        packages_were_checked = True
-        for package in package_list:
-          name = self.format_package_name(package['name'], repository_version)
-          Package(name, use_repos=list(current_repo_files) if OSCheck.is_ubuntu_family() else current_repositories)
-        package_install_result = True
-      except Exception, err:
-        print "Cannot install packages."
-        print traceback.format_exc()
-        num_errors += 1
+    if num_errors > 0:
+      raise Fail("Failed to distribute repositories/install packages")
 
-        # Remove already installed packages in case of fail
-        if packages_were_checked and packages_installed_before:
-          packages_installed_after = []
-          allInstalledPackages(packages_installed_after)
-          packages_installed_after = [package[0] for package in packages_installed_after]
-          packages_installed_before = set(packages_installed_before)
-          new_packages_installed = [package for package in packages_installed_after if package not in packages_installed_before]
-
-          if OSCheck.is_ubuntu_family():
-            package_version_string = repository_version.replace('.', '-')
-          else:
-            package_version_string = repository_version.replace('-', '_')
-            package_version_string = package_version_string.replace('.', '_')
-          for package in new_packages_installed:
-            if package_version_string and (package_version_string in package):
-              Package(package, action="remove")
-
-    # Add more values to structured_out
-    structured_output['package_installation_result'] = 'SUCCESS' if package_install_result else 'FAIL'
-
-    # Even if it failed or did a partial install, report the new version if possible.
-    new_versions = self.hdp_versions()
-    deltas = set(new_versions) - set(old_versions)
-    if 1 == len(deltas):
-      structured_output['actual_version'] = next(iter(deltas))
-
-    self.put_structured_out(structured_output)
+    # If the repo doesn't contain a build number, then will need to calculate it after the packages are installed.
+    self.actual_version = None
+    if self.repository_version:
+      m = re.search("[\d\.]+-\d+", self.repository_version)
+      if m:
+        # Contains a build number
+        self.actual_version = self.repository_version
+        self.structured_output['actual_version'] = self.actual_version
+        self.put_structured_out(self.structured_output)
+
+    # Initial list of versions, used to compute the new version installed
+    self.old_versions = []
+    if self.actual_version is None:
+      Logger.info("Repository version {0} doesn't contain a build number. Will have to calculate the actual version.".format(self.repository_version))
+      self.old_versions = self.hdp_versions()
+
+    try:
+      # It's possible for the process to receive a SIGTERM while installing the packages
+      ret_code = self.install_packages(package_list)
+      if ret_code == 0:
+        self.structured_output['package_installation_result'] = 'SUCCESS'
+        self.put_structured_out(self.structured_output)
+      else:
+        num_errors += 1
+    except Exception, err:
+      Logger.logger.exception("Could not install packages. Error: {0}".format(str(err)))
 
     # Provide correct exit code
     if num_errors > 0:
       raise Fail("Failed to distribute repositories/install packages")
 
-  def install_repository(self, url_info, repository_version, append_to_file):
+  def get_actual_version_from_file(self):
+    """
+    Search the repository version history file for a line that contains repository_version,actual_version
+    Notice that the parts are delimited by a comma.
+    :return: Return the actual_version if found, otherwise, return None.
+    """
+    actual_version = None
+    if os.path.isfile(self.REPO_VERSION_HISTORY_FILE):
+      with open(self.REPO_VERSION_HISTORY_FILE, "r") as f:
+        for line in f.readlines():
+          line_parts = line.split(",")
+          if line_parts and len(line_parts) == 2 and line_parts[0] == self.repository_version:
+            item = line_parts[1].strip()
+            if item != "":
+              actual_version = item
+              break
+    return actual_version
+
+  def write_actual_version_to_file(self, actual_version):
+    """
+    Save the tuple of repository_version,actual_version to the repo version history file if the repository_version
+    doesn't already exist
+    :param actual_version: Repo version with the build number
+    :returns Return True if appended the values to the file, otherwise, return False.
+    """
+    wrote_value = False
+    if self.repository_version is None or actual_version is None:
+      return
+
+    if self.repository_version == "" or actual_version == "":
+      return
+
+    value = self.repository_version + "," + actual_version
+    key_exists = False
+    try:
+      if os.path.isfile(self.REPO_VERSION_HISTORY_FILE):
+        with open(self.REPO_VERSION_HISTORY_FILE, "r") as f:
+          for line in f.readlines():
+            line_parts = line.split(",")
+            if line_parts and len(line_parts) == 2 and line_parts[0] == self.repository_version:
+              key_exists = True
+              break
+
+      if not key_exists:
+        with open(self.REPO_VERSION_HISTORY_FILE, "a") as f:
+          f.write(self.repository_version + "," + actual_version + "\n")
+          wrote_value = True
+      if wrote_value:
+        Logger.info("Appended value \"{0}\" to file {1} to track this as a new version.".format(value, self.REPO_VERSION_HISTORY_FILE))
+    except Exception, err:
+      Logger.error("Failed to write to file {0} the value: {1}. Error: {2}".format(self.REPO_VERSION_HISTORY_FILE, value, str(err)))
+
+    return wrote_value
+
+  def compute_actual_version(self):
+    """
+    After packages are installed, determine what the new actual version is, in order to save it.
+    """
+
+    # If needed to calculate the actual_version, add it to the structured out file.
+    if self.actual_version is None:
+      Logger.info("Attempting to determine actual version with build number.")
+      Logger.info("Old versions: {0}".format(self.old_versions))
+
+      new_versions = self.hdp_versions()
+      Logger.info("New versions: {0}".format(new_versions))
+
+      deltas = set(new_versions) - set(self.old_versions)
+      Logger.info("Deltas: {0}".format(deltas))
+
+      if 1 == len(deltas):
+        self.actual_version = next(iter(deltas)).strip()
+        self.structured_output['actual_version'] = self.actual_version
+        self.put_structured_out(self.structured_output)
+        self.write_actual_version_to_file(self.actual_version)
+      else:
+        Logger.info("Cannot determine a new actual version installed by using the delta method. "
+                    "This is expected during the first install attempt since not all packages will yield a new version in \"hdp-select versions\".")
+        # If the first install attempt does a partial install and is unable to report this to the server,
+        # then a subsequent attempt will report an empty delta. For this reason, it is important to search the
+        # repo version history file to determine if we previously did write an actual_version.
+        self.actual_version = self.get_actual_version_from_file()
+        if self.actual_version is not None:
+          self.actual_version = self.actual_version.strip()
+          self.structured_output['actual_version'] = self.actual_version
+          self.put_structured_out(self.structured_output)
+
+  def install_packages(self, package_list):
+    """
+    Actually install the packages using the package manager.
+    :param package_list: List of package names to install
+    :return: Returns 0 if no errors were found, and 1 otherwise.
+    """
+    ret_code = 0
+    # Install packages
+    packages_were_checked = False
+    try:
+      packages_installed_before = []
+      allInstalledPackages(packages_installed_before)
+      packages_installed_before = [package[0] for package in packages_installed_before]
+      packages_were_checked = True
+      for package in package_list:
+        name = self.format_package_name(package['name'], self.repository_version)
+        Package(name, use_repos=list(self.current_repo_files) if OSCheck.is_ubuntu_family() else self.current_repositories)
+    except Exception, err:
+      ret_code = 1
+      Logger.logger.exception("Package Manager failed to install packages. Error: {0}".format(str(err)))
+
+      # Remove already installed packages in case of fail
+      if packages_were_checked and packages_installed_before:
+        packages_installed_after = []
+        allInstalledPackages(packages_installed_after)
+        packages_installed_after = [package[0] for package in packages_installed_after]
+        packages_installed_before = set(packages_installed_before)
+        new_packages_installed = [package for package in packages_installed_after if package not in packages_installed_before]
+
+        if OSCheck.is_ubuntu_family():
+          package_version_string = self.repository_version.replace('.', '-')
+        else:
+          package_version_string = self.repository_version.replace('-', '_')
+          package_version_string = package_version_string.replace('.', '_')
+        for package in new_packages_installed:
+          if package_version_string and (package_version_string in package):
+            Package(package, action="remove")
+    else:
+      # Compute the actual version in order to save it in structured out
+      self.compute_actual_version()
+
+    pass
+    return ret_code
+
+  def install_repository(self, url_info, append_to_file):
     template = "repo_suse_rhel.j2" if OSCheck.is_redhat_family() or OSCheck.is_suse_family() else "repo_ubuntu.j2"
 
     repo = {
-      'repoName': "{0}-{1}".format(url_info['name'], repository_version)
+      'repoName': "{0}-{1}".format(url_info['name'], self.repository_version)
     }
 
     if not 'baseUrl' in url_info:
@@ -161,7 +292,7 @@ class InstallPackages(Script):
       repo['mirrorsList'] = url_info['mirrorsList']
 
     ubuntu_components = [url_info['name']] + self.UBUNTU_REPO_COMPONENTS_POSTFIX
-    file_name = self.REPO_FILE_NAME_PREFIX + repository_version
+    file_name = self.REPO_FILE_NAME_PREFIX + self.repository_version
 
     Repository(repo['repoName'],
       action = "create",
@@ -198,6 +329,10 @@ class InstallPackages(Script):
     else:
       return []
 
+  def abort_handler(self, signum, frame):
+    Logger.error("Caught signal {0}, will handle it gracefully. Compute the actual version if possible before exiting.".format(signum))
+    self.compute_actual_version()
+
 
 if __name__ == "__main__":
   InstallPackages().execute()

ambari-server/src/test/python/custom_actions/TestInstallPackages.py

@@ -69,6 +69,7 @@ class TestInstallPackages(RMFTestCase):
                       {'package_installation_result': 'SUCCESS',
                        'installed_repository_version': u'2.2.0.1-885',
                        'stack_id': 'HDP-2.2',
+                       'actual_version': u'2.2.0.1-885',
                        'ambari_repositories': []})
     self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url=u'http://repo1/HDP/centos5/2.x/updates/2.2.0.0',
@@ -117,6 +118,7 @@ class TestInstallPackages(RMFTestCase):
                       {'package_installation_result': 'SUCCESS',
                        'installed_repository_version': u'2.2.0.1-885',
                        'stack_id': 'HDP-2.2',
+                       'actual_version': u'2.2.0.1-885',
                        'ambari_repositories': []})
     self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url=u'http://repo1/HDP/centos5/2.x/updates/2.2.0.0',
@@ -167,6 +169,7 @@ class TestInstallPackages(RMFTestCase):
                       {'package_installation_result': 'SUCCESS',
                        'installed_repository_version': u'2.2.0.1-885',
                        'stack_id': 'HDP-2.2',
+                       'actual_version': u'2.2.0.1-885',
                        'ambari_repositories': ["HDP-UTILS-2.2.0.1-885"]})
     self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url=u'http://repo1/HDP/centos5/2.x/updates/2.2.0.0',
@@ -240,7 +243,11 @@ class TestInstallPackages(RMFTestCase):
 
     self.assertTrue(put_structured_out_mock.called)
     self.assertEquals(put_structured_out_mock.call_args[0][0],
-                      {'stack_id': 'HDP-2.2', 'installed_repository_version': u'2.2.0.1-885', 'ambari_repositories': [], 'package_installation_result': 'FAIL'})
+                      {'stack_id': 'HDP-2.2',
+                      'actual_version': u'2.2.0.1-885',
+                      'installed_repository_version': u'2.2.0.1-885',
+                      'ambari_repositories': [],
+                      'package_installation_result': 'FAIL'})
     self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url=u'http://repo1/HDP/centos5/2.x/updates/2.2.0.0',
                               action=['create'],
@@ -284,6 +291,7 @@ class TestInstallPackages(RMFTestCase):
                       {'package_installation_result': 'SUCCESS',
                        'installed_repository_version': u'2.2.0.1-885',
                        'stack_id': 'HDP-2.2',
+                       'actual_version': u'2.2.0.1-885',
                        'ambari_repositories': []})
     self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url=u'http://repo1/HDP/centos5/2.x/updates/2.2.0.0',