AMBARI-8639. Distribute Repository/Install: parallel install not working (dlysnichenko)

ambari-agent/src/test/python/resource_management/TestPackageResource.py

@@ -88,6 +88,18 @@ class TestPackageResource(TestCase):
               )
     self.assertFalse(shell_mock.mock_calls)
 
+  @patch.object(shell, "call", new = MagicMock(return_value=(0, None)))
+  @patch.object(shell, "checked_call")
+  @patch.object(System, "os_family", new = 'redhat')
+  def test_action_install_use_repos_rhel(self, shell_mock):
+    with Environment('/') as env:
+      Package("some_package", use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885']
+              )
+    self.assertEquals(shell_mock.call_args[0][0],
+                      ['/usr/bin/yum', '-d', '0', '-e', '0', '-y', 'install',
+                       '--disablerepo=*',
+                       '--enablerepo=HDP-UTILS-2.2.0.1-885,HDP-2.2.0.1-885', 'some_package'])
+
   @patch.object(shell, "call", new = MagicMock(return_value=(0, None)))
   @patch.object(shell, "checked_call")
   @patch.object(System, "os_family", new = 'suse')

ambari-common/src/main/python/resource_management/core/providers/package/__init__.py

@@ -37,11 +37,11 @@ class PackageProvider(Provider):
 
   def action_install(self):
     package_name = self.get_package_name_with_version()
-    self.install_package(package_name)
+    self.install_package(package_name, self.resource.use_repos)
 
   def action_upgrade(self):
     package_name = self.get_package_name_with_version()
-    self.upgrade_package(package_name)
+    self.upgrade_package(package_name, self.resource.use_repos)
 
   def action_remove(self):
     package_name = self.get_package_name_with_version()

ambari-common/src/main/python/resource_management/core/providers/package/apt.py

@@ -40,7 +40,7 @@ def replace_underscores(function_to_decorate):
 class AptProvider(PackageProvider):
 
   @replace_underscores
-  def install_package(self, name):
+  def install_package(self, name, use_repos=[]):
     if not self._check_existence(name):
       cmd = INSTALL_CMD + [name]
       Logger.info("Installing package %s ('%s')" % (name, string_cmd_from_args_list(cmd)))
@@ -61,8 +61,8 @@ class AptProvider(PackageProvider):
       Logger.info("Skipping installing existent package %s" % (name))
 
   @replace_underscores
-  def upgrade_package(self, name):
-    return self.install_package(name)
+  def upgrade_package(self, name, use_repos=[]):
+    return self.install_package(name, use_repos)
 
   @replace_underscores
   def remove_package(self, name):

ambari-common/src/main/python/resource_management/core/providers/package/yumrpm.py

@@ -30,16 +30,20 @@ REMOVE_CMD = ['/usr/bin/yum', '-d', '0', '-e', '0', '-y', 'erase']
 CHECK_CMD = "installed_pkgs=`rpm -qa %s` ; [ ! -z \"$installed_pkgs\" ]"
 
 class YumProvider(PackageProvider):
-  def install_package(self, name):
-    if not self._check_existence(name):
-      cmd = INSTALL_CMD + [name]
+  def install_package(self, name, use_repos=[]):
+    if not self._check_existence(name) or use_repos:
+      cmd = INSTALL_CMD
+      if use_repos:
+        enable_repo_option = '--enablerepo=' + ",".join(use_repos)
+        cmd = cmd + ['--disablerepo=*', enable_repo_option]
+      cmd = cmd + [name]
       Logger.info("Installing package %s ('%s')" % (name, string_cmd_from_args_list(cmd)))
       shell.checked_call(cmd, sudo=True)
     else:
       Logger.info("Skipping installing existent package %s" % (name))
 
-  def upgrade_package(self, name):
-    return self.install_package(name)
+  def upgrade_package(self, name, use_repos=[]):
+    return self.install_package(name, use_repos)
 
   def remove_package(self, name):
     if self._check_existence(name):

ambari-common/src/main/python/resource_management/core/providers/package/zypper.py

@@ -30,7 +30,7 @@ REMOVE_CMD = ['/usr/bin/zypper', '--quiet', 'remove', '--no-confirm']
 CHECK_CMD = "installed_pkgs=`rpm -qa %s` ; [ ! -z \"$installed_pkgs\" ]"
 
 class ZypperProvider(PackageProvider):
-  def install_package(self, name):
+  def install_package(self, name, use_repos=[]):
     if not self._check_existence(name):
       cmd = INSTALL_CMD + [name]
       Logger.info("Installing package %s ('%s')" % (name, string_cmd_from_args_list(cmd)))
@@ -38,8 +38,8 @@ class ZypperProvider(PackageProvider):
     else:
       Logger.info("Skipping installing existent package %s" % (name))
 
-  def upgrade_package(self, name):
-    return self.install_package(name)
+  def upgrade_package(self, name, use_repos=[]):
+    return self.install_package(name, use_repos)
   
   def remove_package(self, name):
     if self._check_existence(name):

ambari-common/src/main/python/resource_management/core/resources/packaging.py

@@ -29,6 +29,10 @@ class Package(Resource):
   action = ForcedListArgument(default="install")
   package_name = ResourceArgument(default=lambda obj: obj.name)
   location = ResourceArgument(default=lambda obj: obj.package_name)
+
+  # Allow using only specific list of repositories when performing action
+  use_repos = ResourceArgument(default=[])
+
   version = ResourceArgument()
   actions = ["install", "upgrade", "remove"]
   build_vars = ForcedListArgument(default=[])

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

@@ -220,7 +220,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou
           NoSuchParentResourceException {
     Iterator<Map<String, Object>> iterator = request.getProperties().iterator();
     String clName;
-    String desiredRepoVersion;
+    final String desiredRepoVersion;
     String stackName;
     String stackVersion;
     if (request.getProperties().size() != 1) {
@@ -334,6 +334,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou
       final String repoList = gson.toJson(repoInfo);
 
       Map<String, String> params = new HashMap<String, String>() {{
+        put("repository_version", desiredRepoVersion);
         put("base_urls", repoList);
         put("package_list", packageList);
       }};
@@ -346,7 +347,7 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou
               cluster.getClusterName(), INSTALL_PACKAGES_ACTION,
               Collections.singletonList(filter),
               params);
-      actionContext.setTimeout((short) 60);
+      actionContext.setTimeout((short) 600);
 
       try {
         actionExecutionHelper.get().addExecutionCommandsToStage(actionContext, stage);
@@ -356,20 +357,23 @@ public class ClusterStackVersionResourceProvider extends AbstractControllerResou
     }
 
     try {
-      req.persist();
-
-      try {  // Create/persist cluster stack version
-        cluster.createClusterVersion(stackId, desiredRepoVersion, managementController.getAuthName(), RepositoryVersionState.INSTALLED);
-        ClusterVersionEntity newCSVer = clusterVersionDAO.findByClusterAndStackAndVersion(clName, stackId, desiredRepoVersion);
-        cluster.initHostVersions(newCSVer);
-      } catch (AmbariException e) {
-        throw new SystemException(
-                String.format(
-                        "Can not create cluster stack version %s for cluster %s",
-                        desiredRepoVersion, clName),
-                e);
+      ClusterVersionEntity existingCSVer = clusterVersionDAO.findByClusterAndStackAndVersion(clName, stackId, desiredRepoVersion);
+      if (existingCSVer == null) {
+        try {  // Create/persist new cluster stack version
+          cluster.createClusterVersion(stackId, desiredRepoVersion, managementController.getAuthName(), RepositoryVersionState.INSTALLED);
+          ClusterVersionEntity newCSVer = clusterVersionDAO.findByClusterAndStackAndVersion(clName, stackId, desiredRepoVersion);
+          cluster.initHostVersions(newCSVer);
+        } catch (AmbariException e) {
+          throw new SystemException(
+                  String.format(
+                          "Can not create cluster stack version %s for cluster %s",
+                          desiredRepoVersion, clName),
+                  e);
+        }
       }
 
+      req.persist();
+
     } catch (AmbariException e) {
       throw new SystemException("Can not persist request", e);
     }

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

@@ -235,7 +235,7 @@ public class HostStackVersionResourceProvider extends AbstractControllerResource
           NoSuchParentResourceException {
     Iterator<Map<String,Object>> iterator = request.getProperties().iterator();
     String hostName;
-    String desiredRepoVersion;
+    final String desiredRepoVersion;
     String stackName;
     String stackVersion;
     if (request.getProperties().size() != 1) {
@@ -362,6 +362,7 @@ public class HostStackVersionResourceProvider extends AbstractControllerResource
     final String repoList = gson.toJson(repoInfo);
 
     Map<String, String> params = new HashMap<String, String>(){{
+      put("repository_version", desiredRepoVersion);
       put("base_urls", repoList);
       put("package_list", packageList);
     }};
@@ -374,7 +375,7 @@ public class HostStackVersionResourceProvider extends AbstractControllerResource
             cluster.getClusterName(), INSTALL_PACKAGES_ACTION,
             Collections.singletonList(filter),
             params);
-    actionContext.setTimeout((short) 60);
+    actionContext.setTimeout((short) 600);
 
     String caption = String.format(INSTALL_PACKAGES_FULL_NAME + " on host %s", hostName);
     RequestStageContainer req = createRequest(caption);

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

@@ -47,18 +47,22 @@ class InstallPackages(Script):
 
     # Select dict that contains parameters
     try:
+      repository_version = config['roleParams']['repository_version']
       base_urls = json.loads(config['roleParams']['base_urls'])
       package_list = json.loads(config['roleParams']['package_list'])
     except KeyError:
       # Last try
+      repository_version = config['commandParams']['repository_version']
       base_urls = json.loads(config['commandParams']['base_urls'])
       package_list = json.loads(config['commandParams']['package_list'])
 
     # Install/update repositories
     installed_repositories = []
+    current_repositories = ['base']  # Some our packages are installed from the base repo
     try:
       for url_info in base_urls:
-        self.install_repository(url_info)
+        repo_name = self.install_repository(url_info, repository_version)
+        current_repositories.append(repo_name)
 
       installed_repositories = list_ambari_managed_repos()
     except Exception, err:
@@ -70,7 +74,7 @@ class InstallPackages(Script):
     if not delayed_fail:
       try:
         for package in package_list:
-          Package(package['name'])
+          Package(package['name'], use_repos=current_repositories)
         package_install_result = True
       except Exception, err:
         print "Can not install packages."
@@ -81,6 +85,7 @@ class InstallPackages(Script):
     # Build structured output
     structured_output = {
       'ambari_repositories': installed_repositories,
+      'installed_repository_version': repository_version,
       'package_installation_result': 'SUCCESS' if package_install_result else 'FAIL'
     }
     self.put_structured_out(structured_output)
@@ -90,11 +95,11 @@ class InstallPackages(Script):
       raise Fail("Failed to distribute repositories/install packages")
 
 
-  def install_repository(self, url_info):
+  def install_repository(self, url_info, repository_version):
     template = "repo_suse_rhel.j2" if OSCheck.is_redhat_family() or OSCheck.is_suse_family() else "repo_ubuntu.j2"
 
     repo = {
-      'repoName': url_info['repositoryId']
+      'repoName': "{0}-{1}".format(url_info['name'], repository_version)
     }
 
     if not 'baseUrl' in url_info:
@@ -107,7 +112,7 @@ class InstallPackages(Script):
     else:
       repo['mirrorsList'] = url_info['mirrorsList']
 
-    ubuntu_components = [repo['repoName']] + self.UBUNTU_REPO_COMPONENTS_POSTFIX
+    ubuntu_components = [url_info['repositoryId']] + self.UBUNTU_REPO_COMPONENTS_POSTFIX
 
     Repository(repo['repoName'],
       action = "create",
@@ -117,7 +122,7 @@ class InstallPackages(Script):
       repo_template = template,
       components = ubuntu_components,  # ubuntu specific
     )
-
+    return repo['repoName']
 
 if __name__ == "__main__":
   InstallPackages().execute()

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

@@ -42,35 +42,36 @@ class TestInstallPackages(RMFTestCase):
     self.assertTrue(put_structured_out.called)
     self.assertEquals(put_structured_out.call_args[0][0],
                       {'package_installation_result': 'SUCCESS',
+                       'installed_repository_version': u'2.2.0.1-885',
                        'ambari_repositories': []})
-    self.assertResourceCalled('Repository', 'HDP-UTILS-1.1.0.20',
+    self.assertResourceCalled('Repository', 'HDP-UTILS-2.2.0.1-885',
                               base_url='http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0',
                               action=['create'],
                               components=[u'HDP-UTILS-1.1.0.20', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-UTILS-1.1.0.20',
+                              repo_file_name='HDP-UTILS-2.2.0.1-885',
                               mirror_list=None,
     )
-    self.assertResourceCalled('Repository', 'HDP-2.2',
+    self.assertResourceCalled('Repository', 'HDP-2.2.0.1-885',
                               base_url='http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0',
                               action=['create'],
                               components=[u'HDP-2.2', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-2.2',
+                              repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
     )
-    self.assertResourceCalled('Package', 'hadoop_2_2_*',)
-    self.assertResourceCalled('Package', 'snappy',)
-    self.assertResourceCalled('Package', 'snappy-devel',)
-    self.assertResourceCalled('Package', 'lzo',)
-    self.assertResourceCalled('Package', 'hadooplzo_2_2_*',)
-    self.assertResourceCalled('Package', 'hadoop_2_2_*-libhdfs',)
-    self.assertResourceCalled('Package', 'ambari-log4j',)
+    self.assertResourceCalled('Package', 'hadoop_2_2_*', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'snappy', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'snappy-devel', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'lzo', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'hadooplzo_2_2_*', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'hadoop_2_2_*-libhdfs', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'ambari-log4j', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
     self.assertNoMoreResources()
 
 
   @patch("resource_management.libraries.functions.list_ambari_managed_repos.list_ambari_managed_repos",
-         new=MagicMock(return_value=["HDP-UTILS-1.0.0.20"]))
+         new=MagicMock(return_value=["HDP-UTILS-2.2.0.1-885"]))
   @patch("resource_management.libraries.script.Script.put_structured_out")
   def test_exclude_existing_repo(self, put_structured_out):
     self.executeScript("scripts/install_packages.py",
@@ -83,30 +84,31 @@ class TestInstallPackages(RMFTestCase):
     self.assertTrue(put_structured_out.called)
     self.assertEquals(put_structured_out.call_args[0][0],
                       {'package_installation_result': 'SUCCESS',
-                       'ambari_repositories': ["HDP-UTILS-1.0.0.20"]})
-    self.assertResourceCalled('Repository', 'HDP-UTILS-1.1.0.20',
+                       'installed_repository_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='http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0',
                               action=['create'],
                               components=[u'HDP-UTILS-1.1.0.20', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-UTILS-1.1.0.20',
+                              repo_file_name='HDP-UTILS-2.2.0.1-885',
                               mirror_list=None,
     )
-    self.assertResourceCalled('Repository', 'HDP-2.2',
+    self.assertResourceCalled('Repository', 'HDP-2.2.0.1-885',
                               base_url='http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0',
                               action=['create'],
                               components=[u'HDP-2.2', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-2.2',
+                              repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
     )
-    self.assertResourceCalled('Package', 'hadoop_2_2_*',)
-    self.assertResourceCalled('Package', 'snappy',)
-    self.assertResourceCalled('Package', 'snappy-devel',)
-    self.assertResourceCalled('Package', 'lzo',)
-    self.assertResourceCalled('Package', 'hadooplzo_2_2_*',)
-    self.assertResourceCalled('Package', 'hadoop_2_2_*-libhdfs',)
-    self.assertResourceCalled('Package', 'ambari-log4j',)
+    self.assertResourceCalled('Package', 'hadoop_2_2_*', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'snappy', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'snappy-devel', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'lzo', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'hadooplzo_2_2_*', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'hadoop_2_2_*-libhdfs', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'ambari-log4j', use_repos=['HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
     self.assertNoMoreResources()
 
 

ambari-server/src/test/python/custom_actions/configs/install_packages_config.json

@@ -22,6 +22,7 @@
     }, 
     "commandType": "EXECUTION_COMMAND", 
     "roleParams": {
+        "repository_version": "2.2.0.1-885",
         "base_urls": "[{\"name\":\"HDP-UTILS\",\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0\",\"repositoryId\":\"HDP-UTILS-1.1.0.20\"},{\"name\":\"HDP\",\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0\",\"repositoryId\":\"HDP-2.2\"}]",
         "package_list": "[{\"name\":\"hadoop_2_2_*\"},{\"name\":\"snappy\"},{\"name\":\"snappy-devel\"},{\"name\":\"lzo\"},{\"name\":\"hadooplzo_2_2_*\"},{\"name\":\"hadoop_2_2_*-libhdfs\"},{\"name\":\"ambari-log4j\"}]"
     }, 
@@ -33,7 +34,8 @@
     "configurations": {}, 
     "commandParams": {
         "command_timeout": "60", 
-        "script_type": "PYTHON", 
+        "script_type": "PYTHON",
+        "repository_version": "2.2.0.1-885",
         "base_urls": "[{\"name\":\"HDP-UTILS\",\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0\",\"repositoryId\":\"HDP-UTILS-1.1.0.20\"},{\"name\":\"HDP\",\"baseUrl\":\"http://s3.amazonaws.com/dev.hortonworks.com/HDP/centos5/2.x/updates/2.2.0.0\",\"repositoryId\":\"HDP-2.2\"}]",
         "package_list": "[{\"name\":\"hadoop_2_2_*\"},{\"name\":\"snappy\"},{\"name\":\"snappy-devel\"},{\"name\":\"lzo\"},{\"name\":\"hadooplzo_2_2_*\"},{\"name\":\"hadoop_2_2_*-libhdfs\"},{\"name\":\"ambari-log4j\"}]",
         "script": "install_packages.py"