AMBARI-6956. Rebalance HDFS should fail if NameNode is down. (mahadev)

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

@@ -595,15 +595,10 @@ class TestActionQueue(TestCase):
     
     report = actionQueue.result()
     self.assertEqual(len(report['reports']),1)
-    
-
       
-  @patch.object(StackVersionsFileHandler, "read_stack_version")
   @patch.object(CustomServiceOrchestrator, "resolve_script_path")
-  @patch.object(FileCache, "__init__")
-  def test_execute_python_executor(self, read_stack_version_mock, FileCache_mock, resolve_script_path_mock):
-    FileCache_mock.return_value = None
-    
+  @patch.object(StackVersionsFileHandler, "read_stack_version")
+  def test_execute_python_executor(self, read_stack_version_mock, resolve_script_path_mock):
     
     dummy_controller = MagicMock()
     cfg = AmbariConfig().getConfig()

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/HDFS/package/scripts/balancer-emulator/hdfs-command.py

@@ -29,10 +29,10 @@ def write_function(path, handle, interval):
           handle.flush()
           time.sleep(interval)
           
-thread = Thread(target =  write_function, args = ('balancer.log', sys.stdout, 1))
+thread = Thread(target =  write_function, args = ('balancer.log', sys.stdout, 1.5))
 thread.start()
 
-threaderr = Thread(target =  write_function, args = ('balancer-err.log', sys.stderr, 0.3))
+threaderr = Thread(target =  write_function, args = ('balancer-err.log', sys.stderr, 1.5 * 0.023))
 threaderr.start()
 
 thread.join()  

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

@@ -101,11 +101,13 @@ class NameNode(Script):
     _print("Executing command %s\n" % command)
     
     parser = hdfs_rebalance.HdfsParser()
-    proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-                          shell=False,
-                          close_fds=True,
-                          cwd=basedir
-                          )
+    proc = subprocess.Popen(
+                            command, 
+                            stdout=subprocess.PIPE, 
+                            shell=False,
+                            close_fds=True,
+                            cwd=basedir
+                           )
     for line in iter(proc.stdout.readline, ''):
       _print('[balancer] %s %s' % (str(datetime.now()), line ))
       pl = parser.parseLine(line)
@@ -118,7 +120,11 @@ class NameNode(Script):
         _print('[balancer] %s %s' % (str(datetime.now()), 'Process is finished' ))
         self.put_structured_out({'completePercent' : 1})
         break
-      
+    
+    proc.stdout.close()
+    proc.wait()
+    if proc.returncode != None and proc.returncode != 0:
+      raise Fail('Hdfs rebalance process exited with error. See the log output')
       
 def _print(line):
   sys.stdout.write(line)

ambari-server/src/test/python/stacks/2.0.6/HDFS/test_namenode.py

@@ -21,6 +21,9 @@ from ambari_commons import OSCheck
 from ambari_commons import OSCheck
 from mock.mock import MagicMock, patch
 from stacks.utils.RMFTestCase import *
+import resource_management
+
+import subprocess
 
 
 @patch.object(OSCheck,"get_os_type", new = MagicMock(return_value='suse'))
@@ -451,10 +454,42 @@ class TestNamenode(RMFTestCase):
                               recursive = True,
                               mode = 0755,
                               )
-#   def test_rebalance_hdfs(self): ##Does not work because of exectuteScript Framework does not works with strcuturedoutput
-#     self.executeScript("2.0.6/services/HDFS/package/scripts/namenode.py",
-#                        classname = "NameNode",
-#                        command = "rebalancehdfs",
-#                        config_file="rebalancehdfs_default.json"
-#     )
+  
+  @patch("resource_management.libraries.script.Script.put_structured_out")
+  def test_rebalance_hdfs(self, pso):
+    Popen_Mock.return_value = 1
+    with patch("subprocess.Popen", new_callable=Popen_Mock):
+      ll = subprocess.Popen()
+      self.assertTrue(isinstance(ll.stdout.readline(),str))
+      try:
+        self.executeScript("2.0.6/services/HDFS/package/scripts/namenode.py",
+                           classname = "NameNode",
+                           command = "rebalancehdfs",
+                           config_file="rebalancehdfs_default.json"
+        )
+        self.fail("Exception was not thrown")
+      except  resource_management.core.exceptions.Fail:
+        pass ##expected
+       
+      pso.reset_mock()
+      Popen_Mock.return_value = 0
+      ll = subprocess.Popen()
+      self.assertTrue(isinstance(ll.stdout.readline(),str))
+      self.executeScript("2.0.6/services/HDFS/package/scripts/namenode.py",
+                         classname = "NameNode",
+                         command = "rebalancehdfs",
+                         config_file="rebalancehdfs_default.json"
+      )
+      self.assertEqual(pso.call_count, 2, "Output was not parsed properly")
 
+class Popen_Mock:
+  return_value = 1
+  lines = ['Time Stamp               Iteration#  Bytes Already Moved  Bytes Left To Move  Bytes Being Moved\n',
+       'Jul 28, 2014 5:01:49 PM           0                  0 B             5.74 GB            9.79 GB\n',
+       'Jul 28, 2014 5:03:00 PM           1                  0 B             5.58 GB            9.79 GB\n',
+       '']
+  def __call__(self, *args,**kwargs):
+    popen = MagicMock()
+    popen.returncode = Popen_Mock.return_value
+    popen.stdout.readline = MagicMock(side_effect = Popen_Mock.lines)
+    return popen

ambari-server/src/test/python/stacks/2.0.6/configs/rebalancehdfs_default.json

@@ -30,7 +30,7 @@
         "script": "scripts/namenode.py", 
         "hooks_folder": "HDP/2.0.6/hooks", 
         "command_timeout": "600", 
-        "namenode": "{\"threshold\":\"DEBUG\"}", 
+        "namenode": "{\"threshold\":\"-1\"}", 
         "script_type": "PYTHON"
     }, 
     "taskId": 104,