AMBARI-5138. Ambari deploy and unit test fails on Python 2.6.9 (aonishuk)

ambari-agent/src/main/python/resource_management/core/exceptions.py

@@ -23,6 +23,8 @@ Ambari Agent
 class Fail(Exception):
   pass
 
+class ExecuteTimeoutException(Exception):
+  pass
 
 class InvalidArgument(Fail):
   pass

ambari-agent/src/main/python/resource_management/core/providers/system.py

@@ -27,9 +27,9 @@ import os
 import pwd
 import time
 import shutil
-from subprocess import TimeoutExpired
 from resource_management.core import shell
 from resource_management.core.base import Fail
+from resource_management.core import ExecuteTimeoutException
 from resource_management.core.providers import Provider
 from resource_management.core.logger import Logger
 
@@ -240,7 +240,7 @@ class ExecuteProvider(Provider):
         else:
           Logger.info("Retrying after %d seconds. Reason: %s" % (self.resource.try_sleep, str(ex)))
           time.sleep(self.resource.try_sleep)
-      except TimeoutExpired:
+      except ExecuteTimeoutException:
         err_msg = ("Execution of '%s' was killed due timeout after %d seconds") % (self.resource.command, self.resource.timeout)
         
         if self.resource.on_timeout:

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

@@ -22,10 +22,12 @@ Ambari Agent
 
 __all__ = ["checked_call", "call"]
 
-import subprocess
 import pipes
-from subprocess import TimeoutExpired
+import subprocess
+import threading
+from multiprocessing import Queue
 from exceptions import Fail
+from exceptions import ExecuteTimeoutException
 from resource_management.core.logger import Logger
 
 def checked_call(command, logoutput=False, 
@@ -35,8 +37,7 @@ def checked_call(command, logoutput=False,
 def call(command, logoutput=False, 
          cwd=None, env=None, preexec_fn=None, user=None, wait_for_finish=True, timeout=None):
   return _call(command, logoutput, False, cwd, env, preexec_fn, user, wait_for_finish, timeout)
-  
-
+            
 def _call(command, logoutput=False, throw_on_failure=True, 
          cwd=None, env=None, preexec_fn=None, user=None, wait_for_finish=True, timeout=None):
   """
@@ -65,13 +66,20 @@ def _call(command, logoutput=False, throw_on_failure=True,
   if not wait_for_finish:
     return None, None
   
-
-  try:
-    out = proc.communicate(timeout=timeout)[0].strip('\n')
-  except TimeoutExpired as ex:
-    proc.terminate()
-    raise ex
+  if timeout:
+    q = Queue()
+    t = threading.Timer( timeout, on_timeout, [proc, q] )
+    t.start()
     
+  out = proc.communicate()[0].strip('\n')
+  
+  if timeout:
+    if q.empty():
+      t.cancel()
+    # timeout occurred
+    else:
+      raise ExecuteTimeoutException()
+   
   code = proc.returncode
   
   if logoutput and out:
@@ -81,4 +89,12 @@ def _call(command, logoutput=False, throw_on_failure=True,
     err_msg = ("Execution of '%s' returned %d. %s") % (command[-1], code, out)
     raise Fail(err_msg)
   
-  return code, out
+  return code, out
+
+def on_timeout(proc, q):
+  q.put(True)
+  if proc.poll() == None:
+    try:
+      proc.terminate()
+    except:
+      pass