AMBARI-8834. Distribute Repository/Install: multiple improvements (dlysnichenko)

ambari-agent/src/main/python/ambari_agent/AmbariAgent.py

@@ -28,7 +28,7 @@ if os.environ.has_key("PYTHON_BIN"):
 else:
   AGENT_SCRIPT = "/usr/lib/python2.6/site-packages/ambari_agent/main.py"
 if os.environ.has_key("AMBARI_PID_DIR"):
-  AGENT_SCRIPT = os.path.join(os.environ["AMBARI_PID_DIR"],"ambari-agent.pid")
+  AGENT_PID_FILE = os.path.join(os.environ["AMBARI_PID_DIR"],"ambari-agent.pid")
 else:
   AGENT_PID_FILE = "/var/run/ambari-agent/ambari-agent.pid"
 # AGENT_AUTO_RESTART_EXIT_CODE = 77 is exit code which we return when restart_agent() is called

ambari-agent/src/main/python/ambari_agent/CustomServiceOrchestrator.py

@@ -22,7 +22,7 @@ import logging
 import os
 import json
 import sys
-import shell
+from ambari_commons import shell
 import threading
 
 from FileCache import FileCache

ambari-agent/src/main/python/ambari_agent/Facter.py

@@ -27,7 +27,7 @@ import shlex
 import socket
 import multiprocessing
 import subprocess
-from shell import shellRunner
+from ambari_commons.shell import shellRunner
 import time
 import uuid
 from ambari_commons import OSCheck, OSConst

ambari-agent/src/main/python/ambari_agent/Hardware.py

@@ -22,7 +22,7 @@ import os.path
 import logging
 import subprocess
 import platform
-from shell import shellRunner
+from ambari_commons.shell import shellRunner
 from Facter import Facter
 from ambari_commons.os_check import OSConst, OSCheck
 logger = logging.getLogger()

ambari-agent/src/main/python/ambari_agent/HostInfo.py

@@ -28,10 +28,10 @@ import threading
 import shlex
 import platform
 import hostname
-from PackagesAnalyzer import PackagesAnalyzer
 from HostCheckReportFileHandler import HostCheckReportFileHandler
 from Hardware import Hardware
 from ambari_commons import OSCheck, OSConst, Firewall
+from resource_management.libraries.functions import packages_analyzer
 import socket
 from ambari_commons.os_family_impl import OsFamilyImpl
 
@@ -176,7 +176,6 @@ class HostInfoLinux(HostInfo):
 
   def __init__(self, config=None):
     super(HostInfoLinux, self).__init__(config)
-    self.packages = PackagesAnalyzer()
 
   def osdiskAvailableSpace(self, path):
     diskInfo = {}
@@ -281,7 +280,7 @@ class HostInfoLinux(HostInfo):
     for repo in repos:
       addToRemoveList = True
       for ignoreRepo in ignoreList:
-        if self.packages.nameMatch(ignoreRepo, repo):
+        if packages_analyzer.nameMatch(ignoreRepo, repo):
           addToRemoveList = False
           continue
       if addToRemoveList:
@@ -372,17 +371,17 @@ class HostInfoLinux(HostInfo):
 
       installedPackages = []
       availablePackages = []
-      self.packages.allInstalledPackages(installedPackages)
-      self.packages.allAvailablePackages(availablePackages)
+      packages_analyzer.allInstalledPackages(installedPackages)
+      packages_analyzer.allAvailablePackages(availablePackages)
 
       repos = []
-      self.packages.getInstalledRepos(self.PACKAGES, installedPackages + availablePackages,
+      packages_analyzer.getInstalledRepos(self.PACKAGES, installedPackages + availablePackages,
                                       self.IGNORE_PACKAGES_FROM_REPOS, repos)
-      packagesInstalled = self.packages.getInstalledPkgsByRepo(repos, self.IGNORE_PACKAGES, installedPackages)
-      additionalPkgsInstalled = self.packages.getInstalledPkgsByNames(
+      packagesInstalled = packages_analyzer.getInstalledPkgsByRepo(repos, self.IGNORE_PACKAGES, installedPackages)
+      additionalPkgsInstalled = packages_analyzer.getInstalledPkgsByNames(
         self.ADDITIONAL_PACKAGES, installedPackages)
       allPackages = list(set(packagesInstalled + additionalPkgsInstalled))
-      dict['installedPackages'] = self.packages.getPackageDetails(installedPackages, allPackages)
+      dict['installedPackages'] = packages_analyzer.getPackageDetails(installedPackages, allPackages)
 
       repos = self.getReposToRemove(repos, self.IGNORE_REPOS)
       dict['existingRepos'] = repos

ambari-agent/src/main/python/ambari_agent/PackagesAnalyzer.py

@@ -1,234 +0,0 @@
-#!/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.
-'''
-
-import os
-import logging
-import shell
-import subprocess
-from threading import Thread
-import threading
-from ambari_commons import OSCheck, OSConst, Firewall
-
-LIST_INSTALLED_PACKAGES_UBUNTU = "for i in $(dpkg -l |grep ^ii |awk -F' ' '{print $2}'); do      apt-cache showpkg \"$i\"|head -3|grep -v '^Versions'| tr -d '()' | awk '{ print $1\" \"$2 }'|sed -e 's/^Package: //;' | paste -d ' ' - -;  done"
-LIST_AVAILABLE_PACKAGES_UBUNTU = "packages=`for  i in $(ls -1 /var/lib/apt/lists  | grep -v \"ubuntu.com\") ; do grep ^Package: /var/lib/apt/lists/$i |  awk '{print $2}' ; done` ; for i in $packages; do      apt-cache showpkg \"$i\"|head -3|grep -v '^Versions'| tr -d '()' | awk '{ print $1\" \"$2 }'|sed -e 's/^Package: //;' | paste -d ' ' - -;  done"
-
-logger = logging.getLogger()
-
-class PackagesAnalyzer:
-
-  # default timeout for async invoked processes
-  TIMEOUT_SECONDS = 40
-  event = threading.Event()
-
-  def launch_subprocess(self, command):
-    isShell = not isinstance(command, (list, tuple))
-    return subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=isShell, close_fds=True)
-
-  def watchdog_func(self, command):
-    self.event.wait(self.TIMEOUT_SECONDS)
-    if command.returncode is None:
-      logger.error("Task timed out and will be killed")
-      shell.kill_process_with_children(command.pid)
-    pass
-
-  def subprocessWithTimeout(self, command):
-    osStat = self.launch_subprocess(command)
-    logger.debug("Launching watchdog thread")
-    self.event.clear()
-    thread = Thread(target=self.watchdog_func, args=(osStat, ))
-    thread.start()
-
-    out, err = osStat.communicate()
-    result = {}
-    result['out'] = out
-    result['err'] = err
-    result['retCode'] = osStat.returncode
-
-    self.event.set()
-    thread.join()
-    return result
-
-  # Get all installed package whose name starts with the
-  # strings contained in pkgName
-  def installedPkgsByName(self, allInstalledPackages,
-                          pkgName, installedPkgs):
-    for item in allInstalledPackages:
-      if item[0].find(pkgName) == 0:
-        installedPkgs.append(item[0])
-
-  # All installed packages in systems supporting yum
-  def allInstalledPackages(self, allInstalledPackages):
-    osType = OSCheck.get_os_family()
-
-    if osType == OSConst.SUSE_FAMILY:
-      return self.lookUpZypperPackages(
-        ["zypper", "search", "--installed-only", "--details"],
-        allInstalledPackages)
-    elif osType == OSConst.REDHAT_FAMILY:
-      return self.lookUpYumPackages(
-        ["yum", "list", "installed"],
-        'Installed Packages',
-        allInstalledPackages)
-    elif osType == OSConst.UBUNTU_FAMILY:
-       return self.lookUpAptPackages(
-        LIST_INSTALLED_PACKAGES_UBUNTU,
-        allInstalledPackages)
-
-  def allAvailablePackages(self, allAvailablePackages):
-    osType = OSCheck.get_os_family()
-
-    if osType == OSConst.SUSE_FAMILY:
-      return self.lookUpZypperPackages(
-        ["zypper", "search", "--uninstalled-only", "--details"],
-        allAvailablePackages)
-    elif osType == OSConst.REDHAT_FAMILY:
-      return self.lookUpYumPackages(
-        ["yum", "list", "available"],
-        'Available Packages',
-        allAvailablePackages)
-    elif osType == OSConst.UBUNTU_FAMILY:
-       return self.lookUpAptPackages(
-        LIST_AVAILABLE_PACKAGES_UBUNTU,
-        allAvailablePackages)
-
-  def lookUpAptPackages(self, command, allPackages):
-    try:
-      result = self.subprocessWithTimeout(command)
-      if 0 == result['retCode']:
-        for x in result['out'].split('\n'):
-          if x.strip():
-            allPackages.append(x.split(' '))
-
-    except:
-      pass
-
-  def lookUpYumPackages(self, command, skipTill, allPackages):
-    try:
-      result = self.subprocessWithTimeout(command)
-      if 0 == result['retCode']:
-        lines = result['out'].split('\n')
-        lines = [line.strip() for line in lines]
-        items = []
-        skipIndex = 3
-        for index in range(len(lines)):
-          if skipTill in lines[index]:
-            skipIndex = index + 1
-            break
-
-        for line in lines[skipIndex:]:
-          items = items + line.strip(' \t\n\r').split()
-
-        for i in range(0, len(items), 3):
-          if items[i + 2].find('@') == 0:
-            items[i + 2] = items[i + 2][1:]
-          allPackages.append(items[i:i + 3])
-    except:
-      pass
-
-  def lookUpZypperPackages(self, command, allPackages):
-    try:
-      result = self.subprocessWithTimeout(command)
-      if 0 == result['retCode']:
-        lines = result['out'].split('\n')
-        lines = [line.strip() for line in lines]
-        items = []
-        for index in range(len(lines)):
-          if "--+--" in lines[index]:
-            skipIndex = index + 1
-            break
-
-        for line in lines[skipIndex:]:
-          items = line.strip(' \t\n\r').split('|')
-          allPackages.append([items[1].strip(), items[3].strip(), items[5].strip()])
-    except:
-      pass
-
-  def nameMatch(self, lookupName, actualName):
-    tokens = actualName.strip().split()
-    for token in tokens:
-      if token.lower().find(lookupName.lower()) == 0:
-        return True
-    return False
-
-  # Gets all installed repos by name based on repos that provide any package
-  # contained in hintPackages
-  # Repos starting with value in ignoreRepos will not be returned
-  def getInstalledRepos(self, hintPackages, allPackages, ignoreRepos, repoList):
-    allRepos = []
-    for hintPackage in hintPackages:
-      for item in allPackages:
-        if 0 == item[0].find(hintPackage):
-          if not item[2] in allRepos:
-            allRepos.append(item[2])
-        elif hintPackage[0] == '*':
-          if item[0].find(hintPackage[1:]) > 0:
-            if not item[2] in allRepos:
-              allRepos.append(item[2])
-
-    for repo in allRepos:
-      ignore = False
-      for ignoredRepo in ignoreRepos:
-        if self.nameMatch(ignoredRepo, repo):
-          ignore = True
-      if not ignore:
-        repoList.append(repo)
-
-  # Get all the installed packages from the repos listed in repos
-  def getInstalledPkgsByRepo(self, repos, ignorePackages, installedPackages):
-    packagesFromRepo = []
-    packagesToRemove = []
-    for repo in repos:
-      subResult = []
-      for item in installedPackages:
-        if repo == item[2]:
-          subResult.append(item[0])
-      packagesFromRepo = list(set(packagesFromRepo + subResult))
-
-    for package in packagesFromRepo:
-      keepPackage = True
-      for ignorePackage in ignorePackages:
-        if self.nameMatch(ignorePackage, package):
-          keepPackage = False
-          break
-      if keepPackage:
-        packagesToRemove.append(package)
-    return packagesToRemove
-
-  # Gets all installed packages that start with names in pkgNames
-  def getInstalledPkgsByNames(self, pkgNames, installedPackages):
-    packages = []
-    for pkgName in pkgNames:
-      subResult = []
-      self.installedPkgsByName(installedPackages, pkgName, subResult)
-      packages = list(set(packages + subResult))
-    return packages
-
-  # Gets the name, version, and repoName for the packages
-  def getPackageDetails(self, installedPackages, foundPackages):
-    packageDetails = []
-    for package in foundPackages:
-      pkgDetail = {}
-      for installedPackage in installedPackages:
-        if package == installedPackage[0]:
-          pkgDetail['name'] = installedPackage[0]
-          pkgDetail['version'] = installedPackage[1]
-          pkgDetail['repoName'] = installedPackage[2]
-      packageDetails.append(pkgDetail)
-    return packageDetails

ambari-agent/src/main/python/ambari_agent/ProcessHelper.py

@@ -22,7 +22,7 @@ import os
 import logging
 import traceback
 import sys
-from shell import getTempFiles
+from ambari_commons.shell import getTempFiles
 
 logger = logging.getLogger()
 

ambari-agent/src/main/python/ambari_agent/PythonExecutor.py

@@ -29,7 +29,8 @@ import time
 from BackgroundCommandExecutionHandle import BackgroundCommandExecutionHandle
 from ambari_commons.os_check import OSConst, OSCheck
 from Grep import Grep
-import shell, sys
+import sys
+from ambari_commons import shell
 
 
 logger = logging.getLogger()

ambari-agent/src/main/python/ambari_agent/StatusCheck.py

@@ -22,7 +22,7 @@ import logging
 import os
 import re
 import string
-from shell import shellRunner
+from ambari_commons.shell import shellRunner
 
 
 logger = logging.getLogger()

ambari-agent/src/main/python/ambari_agent/main.py

@@ -37,8 +37,8 @@ import hostname
 from DataCleaner import DataCleaner
 import socket
 from ambari_commons import OSConst, OSCheck
-from shell import shellRunner
-from ambari_agent import shell
+from ambari_commons.shell import shellRunner
+from ambari_commons import shell
 import HeartbeatHandlers
 from HeartbeatHandlers import bind_signal_handlers
 logger = logging.getLogger()

ambari-agent/src/test/python/ambari_agent/TestCustomServiceOrchestrator.py

@@ -22,7 +22,7 @@ from multiprocessing.pool import ThreadPool
 import os
 
 import pprint
-import shell
+from ambari_commons import shell
 
 from unittest import TestCase
 import threading
@@ -245,7 +245,7 @@ class TestCustomServiceOrchestrator(TestCase):
 
     pass
 
-  @patch("shell.kill_process_with_children")
+  @patch("ambari_commons.shell.kill_process_with_children")
   @patch.object(CustomServiceOrchestrator, "resolve_script_path")
   @patch.object(CustomServiceOrchestrator, "resolve_hook_script_path")
   @patch.object(FileCache, "get_host_scripts_base_dir")
@@ -325,7 +325,7 @@ class TestCustomServiceOrchestrator(TestCase):
 
   from ambari_agent.StackVersionsFileHandler import StackVersionsFileHandler
 
-  @patch("shell.kill_process_with_children")
+  @patch("ambari_commons.shell.kill_process_with_children")
   @patch.object(FileCache, "__init__")
   @patch.object(CustomServiceOrchestrator, "resolve_script_path")
   @patch.object(CustomServiceOrchestrator, "resolve_hook_script_path")

ambari-agent/src/test/python/ambari_agent/TestHostInfo.py

@@ -40,26 +40,22 @@ else:
 
 with patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value)):
   from ambari_agent.HostCheckReportFileHandler import HostCheckReportFileHandler
-  from ambari_agent.PackagesAnalyzer import PackagesAnalyzer
   from ambari_agent.HostInfo import HostInfo, HostInfoLinux
   from ambari_agent.Hardware import Hardware
   from ambari_agent.AmbariConfig import AmbariConfig
   from resource_management.core.system import System
-  from ambari_commons import OSCheck, Firewall, FirewallChecks ,OSConst
+  from ambari_commons import OSCheck, Firewall, FirewallChecks, OSConst
+  from resource_management.libraries.functions import packages_analyzer
   import ambari_commons
 
-
-
-
 @patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_distro_value))
 class TestHostInfo(TestCase):
 
   @only_for_platform(PLATFORM_LINUX)
   @patch.object(OSCheck, 'get_os_family')
-  @patch.object(PackagesAnalyzer, 'subprocessWithTimeout')
+  @patch('resource_management.libraries.functions.packages_analyzer.subprocessWithTimeout')
   def test_analyze_zypper_out(self, spwt_mock, get_os_family_mock):
     get_os_family_mock.return_value = 'suse'
-    packageAnalyzer = PackagesAnalyzer()
     stringToRead = """Refreshing service 'susecloud'.
            Loading repository data...
            Reading installed packages...
@@ -80,7 +76,7 @@ class TestHostInfo(TestCase):
 
     spwt_mock.return_value = result
     installedPackages = []
-    packageAnalyzer.allInstalledPackages(installedPackages)
+    packages_analyzer.allInstalledPackages(installedPackages)
     self.assertEqual(7, len(installedPackages))
     self.assertTrue(installedPackages[1][0], "gweb")
     self.assertTrue(installedPackages[3][2], "HDP")
@@ -102,7 +98,6 @@ class TestHostInfo(TestCase):
 
   @only_for_platform(PLATFORM_LINUX)
   def test_perform_package_analysis(self):
-    packageAnalyzer = PackagesAnalyzer()
     installedPackages = [
       ["hadoop-a", "2.3", "HDP"], ["zk", "3.1", "HDP"], ["webhcat", "3.1", "HDP"],
       ["hadoop-b", "2.3", "HDP-epel"], ["epel", "3.1", "HDP-epel"], ["epel-2", "3.1", "HDP-epel"],
@@ -123,19 +118,19 @@ class TestHostInfo(TestCase):
     additionalPackages = ["ganglia", "rrd"]
 
     repos = []
-    packageAnalyzer.getInstalledRepos(packagesToLook, installedPackages + availablePackages, reposToIgnore, repos)
+    packages_analyzer.getInstalledRepos(packagesToLook, installedPackages + availablePackages, reposToIgnore, repos)
     self.assertEqual(3, len(repos))
     expected = ["HDP", "HDP-epel", "DEF.3"]
     for repo in expected:
       self.assertTrue(repo in repos)
 
-    packagesInstalled = packageAnalyzer.getInstalledPkgsByRepo(repos, ["epel"], installedPackages)
+    packagesInstalled = packages_analyzer.getInstalledPkgsByRepo(repos, ["epel"], installedPackages)
     self.assertEqual(5, len(packagesInstalled))
     expected = ["hadoop-a", "zk", "webhcat", "hadoop-b", "def-def.x86"]
     for repo in expected:
       self.assertTrue(repo in packagesInstalled)
 
-    additionalPkgsInstalled = packageAnalyzer.getInstalledPkgsByNames(
+    additionalPkgsInstalled = packages_analyzer.getInstalledPkgsByNames(
         additionalPackages, installedPackages)
     self.assertEqual(2, len(additionalPkgsInstalled))
     expected = ["ganglia", "rrd"]
@@ -150,10 +145,9 @@ class TestHostInfo(TestCase):
 
   @only_for_platform(PLATFORM_LINUX)
   @patch.object(OSCheck, 'get_os_family')
-  @patch.object(PackagesAnalyzer, 'subprocessWithTimeout')
+  @patch('resource_management.libraries.functions.packages_analyzer.subprocessWithTimeout')
   def test_analyze_yum_output(self, subprocessWithTimeout_mock, get_os_family_mock):
     get_os_family_mock.return_value = 'redhat'
-    packageAnalyzer = PackagesAnalyzer()
     stringToRead = """Loaded plugins: amazon-id, product-id, rhui-lb, security, subscription-manager
                       Updating certificate-based repositories.
                       Installed Packages
@@ -177,7 +171,7 @@ class TestHostInfo(TestCase):
 
     subprocessWithTimeout_mock.return_value = result
     installedPackages = []
-    packageAnalyzer.allInstalledPackages(installedPackages)
+    packages_analyzer.allInstalledPackages(installedPackages)
     self.assertEqual(9, len(installedPackages))
     for package in installedPackages:
       self.assertTrue(package[0] in ["AMBARI.dev.noarch", "PyXML.x86_64", "oracle-server-db.x86",
@@ -188,7 +182,7 @@ class TestHostInfo(TestCase):
       self.assertTrue(package[2] in ["installed", "koji-override-0", "HDP-1.3.0",
                                  "koji-override-0/$releasever", "AMBARI.dev-1.x", "Oracle-11g", "HDP-epel"])
 
-    packages = packageAnalyzer.getInstalledPkgsByNames(["AMBARI", "Red_Hat_Enterprise", "hesiod", "hive"],
+    packages = packages_analyzer.getInstalledPkgsByNames(["AMBARI", "Red_Hat_Enterprise", "hesiod", "hive"],
                                                        installedPackages)
     self.assertEqual(4, len(packages))
     expected = ["AMBARI.dev.noarch", "Red_Hat_Enterprise_Linux-Release_Notes-6-en-US.noarch",
@@ -196,7 +190,7 @@ class TestHostInfo(TestCase):
     for package in expected:
       self.assertTrue(package in packages)
 
-    detailedPackages = packageAnalyzer.getPackageDetails(installedPackages, packages)
+    detailedPackages = packages_analyzer.getPackageDetails(installedPackages, packages)
     self.assertEqual(4, len(detailedPackages))
     for package in detailedPackages:
       self.assertTrue(package['version'] in ["1.x-1.el6", "3-7.el6", "3.1.0-19.el6",
@@ -207,12 +201,10 @@ class TestHostInfo(TestCase):
 
   @only_for_platform(PLATFORM_LINUX)
   @patch.object(OSCheck, 'get_os_family')
-  @patch.object(PackagesAnalyzer, 'subprocessWithTimeout')
+  @patch('resource_management.libraries.functions.packages_analyzer.subprocessWithTimeout')
   def test_analyze_yum_output_err(self, subprocessWithTimeout_mock, get_os_family_mock):
     get_os_family_mock.return_value = OSConst.REDHAT_FAMILY
 
-    packageAnalyzer = PackagesAnalyzer()
-
     result = {}
     result['out'] = ""
     result['err'] = ""
@@ -220,7 +212,7 @@ class TestHostInfo(TestCase):
 
     subprocessWithTimeout_mock.return_value = result
     installedPackages = []
-    packageAnalyzer.allInstalledPackages(installedPackages)
+    packages_analyzer.allInstalledPackages(installedPackages)
     self.assertEqual(installedPackages, [])
 
 
@@ -265,12 +257,12 @@ class TestHostInfo(TestCase):
   @patch.object(OSCheck, "get_os_type")
   @patch('os.umask')
   @patch.object(HostCheckReportFileHandler, 'writeHostCheckFile')
-  @patch.object(PackagesAnalyzer, 'allAvailablePackages')
-  @patch.object(PackagesAnalyzer, 'allInstalledPackages')
-  @patch.object(PackagesAnalyzer, 'getPackageDetails')
-  @patch.object(PackagesAnalyzer, 'getInstalledPkgsByNames')
-  @patch.object(PackagesAnalyzer, 'getInstalledPkgsByRepo')
-  @patch.object(PackagesAnalyzer, 'getInstalledRepos')
+  @patch('resource_management.libraries.functions.packages_analyzer.allAvailablePackages')
+  @patch('resource_management.libraries.functions.packages_analyzer.allInstalledPackages')
+  @patch('resource_management.libraries.functions.packages_analyzer.getPackageDetails')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledPkgsByNames')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledPkgsByRepo')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledRepos')
   @patch.object(HostInfoLinux, 'checkUsers')
   @patch.object(HostInfoLinux, 'checkLiveServices')
   @patch.object(HostInfoLinux, 'javaProcs')
@@ -308,12 +300,12 @@ class TestHostInfo(TestCase):
   @patch.object(OSCheck, "get_os_type")
   @patch('os.umask')
   @patch.object(HostCheckReportFileHandler, 'writeHostCheckFile')
-  @patch.object(PackagesAnalyzer, 'allAvailablePackages')
-  @patch.object(PackagesAnalyzer, 'allInstalledPackages')
-  @patch.object(PackagesAnalyzer, 'getPackageDetails')
-  @patch.object(PackagesAnalyzer, 'getInstalledPkgsByNames')
-  @patch.object(PackagesAnalyzer, 'getInstalledPkgsByRepo')
-  @patch.object(PackagesAnalyzer, 'getInstalledRepos')
+  @patch('resource_management.libraries.functions.packages_analyzer.allAvailablePackages')
+  @patch('resource_management.libraries.functions.packages_analyzer.allInstalledPackages')
+  @patch('resource_management.libraries.functions.packages_analyzer.getPackageDetails')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledPkgsByNames')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledPkgsByRepo')
+  @patch('resource_management.libraries.functions.packages_analyzer.getInstalledRepos')
   @patch.object(HostInfoLinux, 'checkUsers')
   @patch.object(HostInfoLinux, 'checkLiveServices')
   @patch.object(HostInfoLinux, 'javaProcs')

ambari-agent/src/test/python/ambari_agent/TestMain.py

@@ -46,7 +46,7 @@ with patch.object(OSCheck, "os_distribution", new = MagicMock(return_value = os_
   import ambari_agent.HeartbeatHandlers as HeartbeatHandlers
   from ambari_commons.os_check import OSConst, OSCheck
 
-  from ambari_agent.shell import shellRunner
+  from ambari_commons.shell import shellRunner
 
 class TestMain(unittest.TestCase):
 

ambari-agent/src/test/python/ambari_agent/TestPythonExecutor.py

@@ -30,10 +30,9 @@ from PythonExecutor import PythonExecutor
 from AmbariConfig import AmbariConfig
 from mock.mock import MagicMock, patch
 
-
 class TestPythonExecutor(TestCase):
 
-  @patch("shell.kill_process_with_children")
+  @patch("ambari_commons.shell.kill_process_with_children")
   def test_watchdog_1(self, kill_process_with_children_mock):
     """
     Tests whether watchdog works

ambari-agent/src/test/python/ambari_agent/TestShell.py

@@ -24,8 +24,8 @@ import unittest
 import tempfile
 from mock.mock import patch, MagicMock, call
 from ambari_agent.AmbariConfig import AmbariConfig
-from ambari_agent import shell
-from shell import shellRunner
+from ambari_commons import shell
+from ambari_commons.shell import shellRunner
 from sys import platform as _platform
 from only_for_platform import only_for_platform, PLATFORM_LINUX
 import subprocess, time

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

@@ -138,10 +138,10 @@ class TestRepositoryResource(TestCase):
                      repo_template = "dummy.j2",
                      components = ['a','b','c']
           )
-      
-      template_item = file_mock.call_args_list[0]
-      template_name = template_item[0][0]
-      template_content = template_item[1]['content'].get_content()
+
+      call_content = file_mock.call_args_list[0]
+      template_name = call_content[0][0]
+      template_content = call_content[1]['content']
       
       self.assertEquals(template_name, '/tmp/1.txt')
       self.assertEquals(template_content, 'deb http://download.base_url.org/rpm/ a b c\n')
@@ -173,10 +173,10 @@ class TestRepositoryResource(TestCase):
                      repo_template = "dummy.j2",
                      components = ['a','b','c']
           )
-      
-      template_item = file_mock.call_args_list[0]
-      template_name = template_item[0][0]
-      template_content = template_item[1]['content'].get_content()
+
+      call_content = file_mock.call_args_list[0]
+      template_name = call_content[0][0]
+      template_content = call_content[1]['content']
       
       self.assertEquals(template_name, '/tmp/1.txt')
       self.assertEquals(template_content, 'deb http://download.base_url.org/rpm/ a b c\n')

ambari-agent/src/main/python/ambari_agent/shell.py → ambari-common/src/main/python/ambari_commons/shell.py

@@ -27,7 +27,6 @@ import sys
 import threading
 import time
 import traceback
-import AmbariConfig
 import pprint
 import platform
 

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

@@ -19,6 +19,10 @@ Ambari Agent
 
 """
 
+import os
+import tempfile
+import shutil
+
 from resource_management.core.providers.package import PackageProvider
 from resource_management.core import shell
 from resource_management.core.shell import string_cmd_from_args_list
@@ -37,19 +41,39 @@ REPO_UPDATE_CMD = ['/usr/bin/apt-get', 'update','-qq']
 
 CHECK_CMD = "dpkg --get-selections | grep -v deinstall | awk '{print $1}' | grep ^%s$"
 
+EMPTY_FILE = "/dev/null"
+APT_SOURCES_LIST_DIR = "/etc/apt/sources.list.d"
+
 def replace_underscores(function_to_decorate):
   def wrapper(*args):
     self = args[0]
     name = args[1].replace("_", "-")
-    return function_to_decorate(self, name)
+    return function_to_decorate(self, name, *args[2:])
   return wrapper
 
 class AptProvider(PackageProvider):
 
   @replace_underscores
   def install_package(self, name, use_repos=[]):
-    if not self._check_existence(name):
-      cmd = INSTALL_CMD[self.get_logoutput()]  + [name]
+    if not self._check_existence(name) or use_repos:
+      cmd = INSTALL_CMD[self.get_logoutput()]
+      copied_sources_files = []
+      is_tmp_dir_created = False
+      if use_repos:
+        is_tmp_dir_created = True
+        apt_sources_list_tmp_dir = tempfile.mkdtemp(suffix="-ambari-apt-sources-d")
+        Logger.info("Temporal sources directory was created: %s" % apt_sources_list_tmp_dir)
+        if 'base' not in use_repos:
+          cmd = cmd + ['-o', 'Dir::Etc::SourceList=%s' % EMPTY_FILE]
+        for repo in use_repos:
+          if repo != 'base':
+            new_sources_file = os.path.join(apt_sources_list_tmp_dir, repo + '.list')
+            Logger.info("Temporal sources file will be copied: %s" % new_sources_file)
+            shutil.copy(os.path.join(APT_SOURCES_LIST_DIR, repo + '.list'), new_sources_file)
+            copied_sources_files.append(new_sources_file)
+        cmd = cmd + ['-o', 'Dir::Etc::SourceParts=%s' % apt_sources_list_tmp_dir]
+
+      cmd = cmd + [name]
       Logger.info("Installing package %s ('%s')" % (name, string_cmd_from_args_list(cmd)))
       code, out = shell.call(cmd, sudo=True, env=INSTALL_CMD_ENV, logoutput=self.get_logoutput())
       
@@ -64,6 +88,13 @@ class AptProvider(PackageProvider):
           
         Logger.info("Retrying to install package %s" % (name))
         shell.checked_call(cmd, sudo=True, logoutput=self.get_logoutput())
+
+      if is_tmp_dir_created:
+        for temporal_sources_file in copied_sources_files:
+          Logger.info("Removing temporal sources file: %s" % temporal_sources_file)
+          os.remove(temporal_sources_file)
+        Logger.info("Removing temporal sources directory: %s" % apt_sources_list_tmp_dir)
+        os.rmdir(apt_sources_list_tmp_dir)
     else:
       Logger.info("Skipping installing existent package %s" % (name))
 

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

@@ -24,6 +24,7 @@ from resource_management.core.providers.package import PackageProvider
 from resource_management.core import shell
 from resource_management.core.shell import string_cmd_from_args_list
 from resource_management.core.logger import Logger
+import os
 
 INSTALL_CMD = {
   True: ['/usr/bin/yum', '-y', 'install'],
@@ -62,5 +63,7 @@ class YumProvider(PackageProvider):
       Logger.info("Skipping removing non-existent package %s" % (name))
 
   def _check_existence(self, name):
+    if '.' in name:  # To work with names like 'zookeeper_2_2_1_0_2072.noarch'
+      name = os.path.splitext(name)[0]
     code, out = shell.call(CHECK_CMD % name)
     return not bool(code)

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

@@ -34,11 +34,36 @@ REMOVE_CMD = {
   False: ['/usr/bin/zypper', '--quiet', 'remove', '--no-confirm'],
 }
 CHECK_CMD = "installed_pkgs=`rpm -qa %s` ; [ ! -z \"$installed_pkgs\" ]"
+LIST_ACTIVE_REPOS_CMD = ['/usr/bin/zypper', 'repos']
+
+def get_active_base_repos():
+  (code, output) = shell.call(LIST_ACTIVE_REPOS_CMD)
+  enabled_repos = []
+  if not code:
+    for line in output.split('\n')[2:]:
+      line_list = line.split('|')
+      if line_list[3].strip() == 'Yes' and line_list[2].strip().startswith("SUSE-"):
+        enabled_repos.append(line_list[1].strip())
+      if line_list[2].strip() == 'OpenSuse':
+        return [line_list[1].strip()]
+  return enabled_repos
+
 
 class ZypperProvider(PackageProvider):
   def install_package(self, name, use_repos=[]):
-    if not self._check_existence(name):
-      cmd = INSTALL_CMD[self.get_logoutput()] + [name]
+    if not self._check_existence(name) or use_repos:
+      cmd = INSTALL_CMD[self.get_logoutput()]
+      if use_repos:
+        active_base_repos = get_active_base_repos()
+        if 'base' in use_repos:
+          use_repos = filter(lambda x: x != 'base', use_repos)
+          use_repos.extend(active_base_repos)
+        use_repos_options = []
+        for repo in use_repos:
+          use_repos_options = use_repos_options + ['--repo', repo]
+        cmd = cmd + use_repos_options
+
+      cmd = cmd + [name]
       Logger.info("Installing package %s ('%s')" % (name, string_cmd_from_args_list(cmd)))
       shell.checked_call(cmd, sudo=True, logoutput=self.get_logoutput())
     else:

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

@@ -0,0 +1,263 @@
+#!/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.
+"""
+
+import sys
+import logging
+from ambari_commons import shell
+import subprocess
+from threading import Thread
+import threading
+from ambari_commons import OSCheck, OSConst, Firewall
+
+__all__ = ["installedPkgsByName", "allInstalledPackages", "allAvailablePackages", "nameMatch",
+           "getInstalledRepos", "getInstalledPkgsByRepo", "getInstalledPkgsByNames", "getPackageDetails"]
+
+LIST_INSTALLED_PACKAGES_UBUNTU = "for i in $(dpkg -l |grep ^ii |awk -F' ' '{print $2}'); do      apt-cache showpkg \"$i\"|head -3|grep -v '^Versions'| tr -d '()' | awk '{ print $1\" \"$2 }'|sed -e 's/^Package: //;' | paste -d ' ' - -;  done"
+LIST_AVAILABLE_PACKAGES_UBUNTU = "packages=`for  i in $(ls -1 /var/lib/apt/lists  | grep -v \"ubuntu.com\") ; do grep ^Package: /var/lib/apt/lists/$i |  awk '{print $2}' ; done` ; for i in $packages; do      apt-cache showpkg \"$i\"|head -3|grep -v '^Versions'| tr -d '()' | awk '{ print $1\" \"$2 }'|sed -e 's/^Package: //;' | paste -d ' ' - -;  done"
+
+logger = logging.getLogger()
+
+# default timeout for async invoked processes
+TIMEOUT_SECONDS = 40
+
+
+def _launch_subprocess(command):
+  isShell = not isinstance(command, (list, tuple))
+  return subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=isShell, close_fds=True)
+
+
+def subprocessWithTimeout(command):
+  event = threading.Event()
+
+  def watchdog_func(command):
+    event.wait(TIMEOUT_SECONDS)
+    if command.returncode is None:
+      logger.error("Task timed out and will be killed")
+      shell.kill_process_with_children(command.pid)
+    pass
+
+  osStat = _launch_subprocess(command)
+  logger.debug("Launching watchdog thread")
+
+  event.clear()
+
+  thread = Thread(target=watchdog_func, args=(osStat, ))
+  thread.start()
+
+  out, err = osStat.communicate()
+  result = {}
+  result['out'] = out
+  result['err'] = err
+  result['retCode'] = osStat.returncode
+
+  event.set()
+  thread.join()
+  return result
+
+
+def installedPkgsByName(allInstalledPackages,
+                        pkgName, installedPkgs):
+  """
+  Get all installed package whose name starts with the
+  strings contained in pkgName
+  """
+  for item in allInstalledPackages:
+    if item[0].find(pkgName) == 0:
+      installedPkgs.append(item[0])
+
+
+def allInstalledPackages(allInstalledPackages):
+  """
+  All installed packages in system
+  """
+  osType = OSCheck.get_os_family()
+
+  if osType == OSConst.SUSE_FAMILY:
+    return _lookUpZypperPackages(
+      ["zypper", "search", "--installed-only", "--details"],
+      allInstalledPackages)
+  elif osType == OSConst.REDHAT_FAMILY:
+    return _lookUpYumPackages(
+      ["yum", "list", "installed"],
+      'Installed Packages',
+      allInstalledPackages)
+  elif osType == OSConst.UBUNTU_FAMILY:
+     return _lookUpAptPackages(
+      LIST_INSTALLED_PACKAGES_UBUNTU,
+      allInstalledPackages)
+
+
+def allAvailablePackages(allAvailablePackages):
+  osType = OSCheck.get_os_family()
+
+  if osType == OSConst.SUSE_FAMILY:
+    return _lookUpZypperPackages(
+      ["zypper", "search", "--uninstalled-only", "--details"],
+      allAvailablePackages)
+  elif osType == OSConst.REDHAT_FAMILY:
+    return _lookUpYumPackages(
+      ["yum", "list", "available"],
+      'Available Packages',
+      allAvailablePackages)
+  elif osType == OSConst.UBUNTU_FAMILY:
+     return _lookUpAptPackages(
+      LIST_AVAILABLE_PACKAGES_UBUNTU,
+      allAvailablePackages)
+
+
+def _lookUpAptPackages(command, allPackages):
+  try:
+    result = subprocessWithTimeout(command)
+    if 0 == result['retCode']:
+      for x in result['out'].split('\n'):
+        if x.strip():
+          allPackages.append(x.split(' '))
+
+  except:
+    logger.error("Unexpected error:", sys.exc_info()[0])
+
+
+def _lookUpYumPackages(command, skipTill, allPackages):
+  try:
+    result = subprocessWithTimeout(command)
+    if 0 == result['retCode']:
+      lines = result['out'].split('\n')
+      lines = [line.strip() for line in lines]
+      items = []
+      skipIndex = 3
+      for index in range(len(lines)):
+        if skipTill in lines[index]:
+          skipIndex = index + 1
+          break
+
+      for line in lines[skipIndex:]:
+        items = items + line.strip(' \t\n\r').split()
+
+      for i in range(0, len(items), 3):
+        if items[i + 2].find('@') == 0:
+          items[i + 2] = items[i + 2][1:]
+        allPackages.append(items[i:i + 3])
+  except:
+    logger.error("Unexpected error:", sys.exc_info()[0])
+
+
+def _lookUpZypperPackages(command, allPackages):
+  try:
+    result = subprocessWithTimeout(command)
+    if 0 == result['retCode']:
+      lines = result['out'].split('\n')
+      lines = [line.strip() for line in lines]
+      items = []
+      for index in range(len(lines)):
+        if "--+--" in lines[index]:
+          skipIndex = index + 1
+          break
+
+      for line in lines[skipIndex:]:
+        items = line.strip(' \t\n\r').split('|')
+        allPackages.append([items[1].strip(), items[3].strip(), items[5].strip()])
+  except:
+    logger.error("Unexpected error:", sys.exc_info()[0])
+
+
+def nameMatch(lookupName, actualName):
+  tokens = actualName.strip().split()
+  for token in tokens:
+    if token.lower().find(lookupName.lower()) == 0:
+      return True
+  return False
+
+
+def getInstalledRepos(hintPackages, allPackages, ignoreRepos, repoList):
+  """
+  Gets all installed repos by name based on repos that provide any package
+  contained in hintPackages
+  Repos starting with value in ignoreRepos will not be returned
+  """
+  allRepos = []
+  for hintPackage in hintPackages:
+    for item in allPackages:
+      if 0 == item[0].find(hintPackage):
+        if not item[2] in allRepos:
+          allRepos.append(item[2])
+      elif hintPackage[0] == '*':
+        if item[0].find(hintPackage[1:]) > 0:
+          if not item[2] in allRepos:
+            allRepos.append(item[2])
+
+  for repo in allRepos:
+    ignore = False
+    for ignoredRepo in ignoreRepos:
+      if nameMatch(ignoredRepo, repo):
+        ignore = True
+    if not ignore:
+      repoList.append(repo)
+
+
+def getInstalledPkgsByRepo(repos, ignorePackages, installedPackages):
+  """
+  Get all the installed packages from the repos listed in repos
+  """
+  packagesFromRepo = []
+  packagesToRemove = []
+  for repo in repos:
+    subResult = []
+    for item in installedPackages:
+      if repo == item[2]:
+        subResult.append(item[0])
+    packagesFromRepo = list(set(packagesFromRepo + subResult))
+
+  for package in packagesFromRepo:
+    keepPackage = True
+    for ignorePackage in ignorePackages:
+      if nameMatch(ignorePackage, package):
+        keepPackage = False
+        break
+    if keepPackage:
+      packagesToRemove.append(package)
+  return packagesToRemove
+
+
+def getInstalledPkgsByNames(pkgNames, installedPackages):
+  """
+  Gets all installed packages that start with names in pkgNames
+  """
+  packages = []
+  for pkgName in pkgNames:
+    subResult = []
+    installedPkgsByName(installedPackages, pkgName, subResult)
+    packages = list(set(packages + subResult))
+  return packages
+
+
+def getPackageDetails(installedPackages, foundPackages):
+  """
+  Gets the name, version, and repoName for the packages
+  """
+  packageDetails = []
+  for package in foundPackages:
+    pkgDetail = {}
+    for installedPackage in installedPackages:
+      if package == installedPackage[0]:
+        pkgDetail['name'] = installedPackage[0]
+        pkgDetail['version'] = installedPackage[1]
+        pkgDetail['repoName'] = installedPackage[2]
+    packageDetails.append(pkgDetail)
+  return packageDetails

ambari-common/src/main/python/resource_management/libraries/providers/repository.py

@@ -32,9 +32,14 @@ class RhelSuseRepositoryProvider(Provider):
       repo_file_name = self.resource.repo_file_name
       repo_dir = repos_dirs[env.system.os_family]
       repo_template = self.resource.repo_template
-      File(format("{repo_dir}/{repo_file_name}.repo"),
-        content = Template(repo_template, repo_id=self.resource.repo_id, repo_file_name=self.resource.repo_file_name, base_url=self.resource.base_url, mirror_list=self.resource.mirror_list)
-      )
+      new_content = Template(repo_template, repo_id=self.resource.repo_id, repo_file_name=self.resource.repo_file_name,
+                             base_url=self.resource.base_url, mirror_list=self.resource.mirror_list)
+      repo_file_path = format("{repo_dir}/{repo_file_name}.repo")
+      if self.resource.append_to_file and os.path.isfile(repo_file_path):
+        with open(repo_file_path, 'a') as repo_file:
+          repo_file.write('\n' + new_content.get_content())
+      else:
+        File(repo_file_path, content=new_content)
   
   def action_remove(self):
     with Environment.get_instance_copy() as env:
@@ -61,14 +66,19 @@ class UbuntuRepositoryProvider(Provider):
   def action_create(self):
     with Environment.get_instance_copy() as env:
       with tempfile.NamedTemporaryFile() as tmpf:
-        File(tmpf.name,
-          content = Template(self.resource.repo_template,
-              package_type=self.package_type, base_url=self.resource.base_url, components=' '.join(self.resource.components))
-        )
-        
         repo_file_name = format("{repo_file_name}.list",repo_file_name = self.resource.repo_file_name)
         repo_file_path = format("{repo_dir}/{repo_file_name}", repo_dir = self.repo_dir)
-        
+
+        new_content = Template(self.resource.repo_template, package_type=self.package_type,
+                                      base_url=self.resource.base_url,
+                                      components=' '.join(self.resource.components)).get_content()
+        old_content = ''
+        if self.resource.append_to_file and os.path.isfile(repo_file_path):
+          with open(repo_file_path) as repo_file:
+            old_content = repo_file.read() + '\n'
+
+        File(tmpf.name, content=old_content+new_content)
+
         if not os.path.isfile(repo_file_path) or not filecmp.cmp(tmpf.name, repo_file_path):
           File(repo_file_path,
                content = StaticFile(tmpf.name)

ambari-common/src/main/python/resource_management/libraries/resources/repository.py

@@ -31,6 +31,7 @@ class Repository(Resource):
   mirror_list = ResourceArgument()
   repo_file_name = ResourceArgument()
   repo_template = ResourceArgument()
+  append_to_file = ResourceArgument(default=False)
   components = ForcedListArgument(default=[]) # ubuntu specific
 
   actions = Resource.actions + ["create","remove"]

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

@@ -26,6 +26,7 @@ import traceback
 from resource_management import *
 from resource_management.libraries.functions.list_ambari_managed_repos import *
 from ambari_commons.os_check import OSCheck, OSConst
+from resource_management.libraries.functions import packages_analyzer
 
 
 class InstallPackages(Script):
@@ -37,6 +38,7 @@ class InstallPackages(Script):
   """
 
   UBUNTU_REPO_COMPONENTS_POSTFIX = ["main"]
+  REPO_FILE_NAME_PREFIX = 'HDP-'
 
   def actionexecute(self, env):
     delayed_fail = False
@@ -59,10 +61,14 @@ class InstallPackages(Script):
     # Install/update repositories
     installed_repositories = []
     current_repositories = ['base']  # Some our packages are installed from the base repo
+    current_repo_files = set(['base'])
     try:
+      append_to_file = False
       for url_info in base_urls:
-        repo_name = self.install_repository(url_info, repository_version)
+        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)
+        append_to_file = True
 
       installed_repositories = list_ambari_managed_repos()
     except Exception, err:
@@ -72,15 +78,29 @@ class InstallPackages(Script):
 
     # Install packages
     if not delayed_fail:
+      packages_were_checked = False
       try:
+        packages_installed_before = []
+        packages_analyzer.allInstalledPackages(packages_installed_before)
+        packages_installed_before = [package[0] for package in packages_installed_before]
+        packages_were_checked = True
         for package in package_list:
-          Package(package['name'], use_repos=current_repositories)
+          Package(package['name'], use_repos=list(current_repo_files) if OSCheck.is_ubuntu_family() else current_repositories)
         package_install_result = True
       except Exception, err:
         print "Can not install packages."
         print traceback.format_exc()
         delayed_fail = True
-        # TODO : remove already installed packages in case of fail
+
+        # Remove already installed packages in case of fail
+        if packages_were_checked and packages_installed_before:
+          packages_installed_after = []
+          packages_analyzer.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]
+          for package in new_packages_installed:
+            Package(package, action="remove")
 
     # Build structured output
     structured_output = {
@@ -94,8 +114,7 @@ class InstallPackages(Script):
     if delayed_fail:
       raise Fail("Failed to distribute repositories/install packages")
 
-
-  def install_repository(self, url_info, repository_version):
+  def install_repository(self, url_info, repository_version, append_to_file):
     template = "repo_suse_rhel.j2" if OSCheck.is_redhat_family() or OSCheck.is_suse_family() else "repo_ubuntu.j2"
 
     repo = {
@@ -113,16 +132,18 @@ class InstallPackages(Script):
       repo['mirrorsList'] = url_info['mirrorsList']
 
     ubuntu_components = [url_info['repositoryId']] + self.UBUNTU_REPO_COMPONENTS_POSTFIX
+    file_name = self.REPO_FILE_NAME_PREFIX + repository_version
 
     Repository(repo['repoName'],
       action = "create",
       base_url = repo['baseurl'],
       mirror_list = repo['mirrorsList'],
-      repo_file_name = repo['repoName'],
+      repo_file_name = file_name,
       repo_template = template,
+      append_to_file = append_to_file,
       components = ubuntu_components,  # ubuntu specific
     )
-    return repo['repoName']
+    return repo['repoName'], file_name
 
 if __name__ == "__main__":
   InstallPackages().execute()

ambari-server/src/main/resources/custom_actions/templates/repo_suse_rhel.j2

@@ -1,7 +1,8 @@
 [{{repo_id}}]
-name={{repo_file_name}}
+name={{repo_id}}
 {% if mirror_list %}mirrorlist={{mirror_list}}{% else %}baseurl={{base_url}}{% endif %}
 
 path=/
 enabled=1
 gpgcheck=0
+

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

@@ -26,11 +26,19 @@ from mock.mock import MagicMock
 from stacks.utils.RMFTestCase import *
 from install_packages import InstallPackages
 from mock.mock import patch, MagicMock
-
+from resource_management.core.base import Resource
+from resource_management.core.resources.packaging import Package
+from resource_management.core.exceptions import Fail
 
 class TestInstallPackages(RMFTestCase):
 
+  def _add_packages(arg):
+    arg.append(["pkg1", "1.0", "repo"])
+    arg.append(["pkg2", "2.0", "repo2"])
+
   @patch("resource_management.libraries.script.Script.put_structured_out")
+  @patch("resource_management.libraries.functions.packages_analyzer.allInstalledPackages",
+         new=MagicMock(side_effect = _add_packages))
   def test_normal_flow(self, put_structured_out):
     self.executeScript("scripts/install_packages.py",
                        classname="InstallPackages",
@@ -49,8 +57,9 @@ class TestInstallPackages(RMFTestCase):
                               action=['create'],
                               components=[u'HDP-UTILS-1.1.0.20', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-UTILS-2.2.0.1-885',
+                              repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
+                              append_to_file=False,
     )
     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',
@@ -59,6 +68,7 @@ class TestInstallPackages(RMFTestCase):
                               repo_template='repo_suse_rhel.j2',
                               repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
+                              append_to_file=True,
     )
     self.assertResourceCalled('Package', 'hadoop_2_2_*', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
     self.assertResourceCalled('Package', 'snappy', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
@@ -73,6 +83,8 @@ class TestInstallPackages(RMFTestCase):
   @patch("resource_management.libraries.functions.list_ambari_managed_repos.list_ambari_managed_repos",
          new=MagicMock(return_value=["HDP-UTILS-2.2.0.1-885"]))
   @patch("resource_management.libraries.script.Script.put_structured_out")
+  @patch("resource_management.libraries.functions.packages_analyzer.allInstalledPackages",
+         new=MagicMock(side_effect = _add_packages))
   def test_exclude_existing_repo(self, put_structured_out):
     self.executeScript("scripts/install_packages.py",
                        classname="InstallPackages",
@@ -91,8 +103,9 @@ class TestInstallPackages(RMFTestCase):
                               action=['create'],
                               components=[u'HDP-UTILS-1.1.0.20', 'main'],
                               repo_template='repo_suse_rhel.j2',
-                              repo_file_name='HDP-UTILS-2.2.0.1-885',
+                              repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
+                              append_to_file=False,
     )
     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',
@@ -101,6 +114,7 @@ class TestInstallPackages(RMFTestCase):
                               repo_template='repo_suse_rhel.j2',
                               repo_file_name='HDP-2.2.0.1-885',
                               mirror_list=None,
+                              append_to_file=True,
     )
     self.assertResourceCalled('Package', 'hadoop_2_2_*', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
     self.assertResourceCalled('Package', 'snappy', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
@@ -112,3 +126,62 @@ class TestInstallPackages(RMFTestCase):
     self.assertNoMoreResources()
 
 
+  _install_failed = False
+
+  def _add_packages_with_fail(arg):
+    arg.append(["pkg1", "1.0", "repo"])
+    arg.append(["pkg2", "2.0", "repo2"])
+    if TestInstallPackages._install_failed:
+      arg.append(["hadoop_2_2_fake_pkg", "1.0", "repo"])
+      arg.append(["snappy_fake_pkg", "3.0", "repo2"])
+
+  @staticmethod
+  def _new_with_exception(cls, name, env=None, provider=None, **kwargs):
+    if (name != "snappy-devel"):
+      return Resource.__new__(cls, name, env, provider, **kwargs)
+    else:
+      TestInstallPackages._install_failed = True
+      raise Exception()
+
+  @patch("resource_management.libraries.script.Script.put_structured_out")
+  @patch("resource_management.libraries.functions.packages_analyzer.allInstalledPackages",
+         new=MagicMock(side_effect = _add_packages_with_fail))
+  @patch("resource_management.core.resources.packaging.Package.__new__",
+         new=_new_with_exception)
+  def test_fail(self, put_structured_out):
+    self.assertRaises(Fail, self.executeScript, "scripts/install_packages.py",
+                      classname="InstallPackages",
+                      command="actionexecute",
+                      config_file="install_packages_config.json",
+                      target=RMFTestCase.TARGET_CUSTOM_ACTIONS,
+                      os_type=('Suse', '11', 'Final'))
+
+    self.assertTrue(put_structured_out.called)
+    self.assertEquals(put_structured_out.call_args[0][0],
+                      {'package_installation_result': 'FAIL',
+                       'installed_repository_version': u'2.2.0.1-885',
+                       'ambari_repositories': []})
+    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-2.2.0.1-885',
+                              mirror_list=None,
+                              append_to_file=False,
+                              )
+    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.0.1-885',
+                              mirror_list=None,
+                              append_to_file=True,
+                              )
+    self.assertResourceCalled('Package', 'hadoop_2_2_*', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'snappy', use_repos=['base', 'HDP-UTILS-2.2.0.1-885', 'HDP-2.2.0.1-885'])
+    self.assertResourceCalled('Package', 'hadoop_2_2_fake_pkg', action=["remove"])
+    self.assertResourceCalled('Package', 'snappy_fake_pkg', action=["remove"])
+    self.assertNoMoreResources()
+