AMBARI-8068. Ambari Server HA Phase 1: Provide backup and restore capability (via ncole)

ambari-agent/conf/unix/ambari-agent

@@ -175,7 +175,7 @@ case "$1" in
           tput bold
           echo "$AMBARI_AGENT currently not running"
           tput sgr0
-          echo "Usage: /usr/sbin/ambari-agent {start|stop|restart|status}"
+          echo "Usage: /usr/sbin/ambari-agent {start|stop|restart|status|reset <server_hostname>}"
           retcode=3
         fi
         ;;
@@ -212,9 +212,33 @@ case "$1" in
         $0 start "$@"
         retcode=$?
         ;;
+  reset)
+          if [ "$#" -ne 2 ]; then
+            echo "You must supply the hostname of the Ambari Server with the restore option. (e.g. ambari-agent reset c6401.ambari.apache.org)"
+            exit 1
+          fi
+          if [ -f $PIDFILE ]; then
+            echo "$AMBARI_AGENT is running. You must stop it before using reset."
+            exit 1
+          fi
+          echo -e "Resetting $AMBARI_AGENT"
+          $PYTHON $AGENT_SCRIPT reset $2
+          retcode=$?
+
+          if [ $retcode -eq 0 ]; then
+            tput bold
+            echo "$AMBARI_AGENT has been reset successfully. Changed Ambari Server hostname to $2 and certificates were cleared."
+            tput sgr0
+          else
+            tput bold
+            echo "$AMBARI_AGENT could not be reset."
+            tput sgr0
+          fi
+          ;;
+
   *)
         tput bold
-        echo "Usage: /usr/sbin/ambari-agent {start|stop|restart|status}"
+        echo "Usage: /usr/sbin/ambari-agent {start|stop|restart|status|reset <server_hostname>}"
         tput sgr0
         retcode=1
 esac

ambari-agent/etc/init.d/ambari-agent

@@ -34,8 +34,11 @@ case "$1" in
         $0 stop
         $0 start
         ;;
+  reset)
+        /usr/sbin/ambari-agent $@
+        ;;
   *)
-        echo "Usage: $0 {start|stop|status|restart}"
+        echo "Usage: $0 {start|stop|status|restart|reset <server_hostname>}"
         exit 1
 esac
 

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

@@ -180,6 +180,32 @@ def stop_agent():
       os.kill(pid, signal.SIGKILL)
     os._exit(1)
 
+def reset_agent(options):
+  try:
+    # update agent config file
+    agent_config = ConfigParser.ConfigParser()
+    agent_config.read(configFile)
+    server_host = agent_config.get('server', 'hostname')
+    new_host = options[2]
+    if new_host is not None and server_host != new_host:
+      print "Updating server host from " + server_host + " to " + new_host
+      agent_config.set('server', 'hostname', new_host)
+      with (open(configFile, "wb")) as new_agent_config:
+        agent_config.write(new_agent_config)
+
+    # clear agent certs
+    agent_keysdir = agent_config.get('security', 'keysdir')
+    print "Removing Agent certificates..."
+    for root, dirs, files in os.walk(agent_keysdir, topdown=False):
+      for name in files:
+        os.remove(os.path.join(root, name))
+      for name in dirs:
+        os.rmdir(os.path.join(root, name))
+  except Exception, err:
+    print("A problem occurred while trying to reset the agent: " + str(err))
+    os._exit(1)
+
+  os._exit(0)
 
 def main():
   global config
@@ -201,6 +227,9 @@ def main():
   if (len(sys.argv) > 1) and sys.argv[1] == 'stop':
     stop_agent()
 
+  if (len(sys.argv) > 2) and sys.argv[1] == 'reset':
+    reset_agent(sys.argv)
+
   # Check for ambari configuration file.
   config = resolve_ambari_config()
 

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

@@ -25,8 +25,9 @@ import signal
 import os
 import socket
 import tempfile
+import ConfigParser
 
-from mock.mock import MagicMock, patch, ANY
+from mock.mock import MagicMock, patch, ANY, Mock
 
 with patch("platform.linux_distribution", return_value = ('Suse','11','Final')):
   from ambari_agent import NetUtil, security
@@ -225,6 +226,44 @@ class TestMain(unittest.TestCase):
     ProcessHelper.pidfile = oldpid
     os.remove(tmpoutfile)
 
+  @patch("os.rmdir")
+  @patch("os.path.join")
+  @patch('__builtin__.open')
+  @patch.object(ConfigParser, "ConfigParser")
+  @patch("os._exit")
+  @patch("os.walk")
+  @patch("os.remove")
+  def test_reset(self, os_remove_mock, os_walk_mock, os_exit_mock, config_parser_mock, open_mock, os_path_join_mock, os_rmdir_mock):
+    # Agent config update
+    config_mock = MagicMock()
+    os_walk_mock.return_value = [('/', ('',), ('file1.txt', 'file2.txt'))]
+    config_parser_mock.return_value= config_mock
+    config_mock.get('server', 'hostname').return_value = "old_host"
+    main.reset_agent(["test", "reset", "new_hostname"])
+    self.assertEqual(config_mock.get.call_count, 3)
+    self.assertEqual(config_mock.set.call_count, 1)
+    self.assertEqual(os_remove_mock.call_count, 2)
+
+  @patch("os.rmdir")
+  @patch("os.path.join")
+  @patch('__builtin__.open')
+  @patch.object(ConfigParser, "ConfigParser")
+  @patch("os._exit")
+  @patch("os.walk")
+  @patch("os.remove")
+  def test_reset_invalid_path(self, os_remove_mock, os_walk_mock, os_exit_mock, config_parser_mock, open_mock, os_path_join_mock, os_rmdir_mock):
+    # Agent config file cannot be accessed
+    config_mock = MagicMock()
+    os_walk_mock.return_value = [('/', ('',), ('file1.txt', 'file2.txt'))]
+    config_parser_mock.return_value= config_mock
+    config_mock.get('server', 'hostname').return_value = "old_host"
+    open_mock.side_effect = Exception("Invalid Path!")
+    try:
+      main.reset_agent(["test", "reset", "new_hostname"])
+      self.fail("Should have thrown exception!")
+    except:
+      self.assertTrue(True)
+
 
   @patch.object(socket, "gethostbyname")
   @patch.object(main, "setup_logging")

ambari-server/sbin/ambari-server

@@ -127,9 +127,17 @@ case "$1" in
         echo -e "Refreshing stack hashes..."
         $PYTHON /usr/sbin/ambari-server.py $@
         ;;
+  backup)
+        echo -e "Backing up Ambari File System state... *this will not backup the server database*"
+        $PYTHON /usr/sbin/ambari-server.py $@
+        ;;
+  restore)
+        echo -e "Restoring Ambari File System state"
+        $PYTHON /usr/sbin/ambari-server.py $@
+        ;;
   *)
         echo "Usage: /usr/sbin/ambari-server
-        {start|stop|restart|setup|upgrade|status|upgradestack|setup-ldap|sync-ldap|setup-security|refresh-stack-hash} [options]
+        {start|stop|restart|setup|upgrade|status|upgradestack|setup-ldap|sync-ldap|setup-security|refresh-stack-hash|backup|restore} [options]
         Use usr/sbin/ambari-server <action> --help to get details on options available.
         Or, simply invoke ambari-server.py --help to print the options."
         exit 1

ambari-server/src/main/python/ambari-server.py

@@ -44,7 +44,7 @@ import json
 import base64
 from threading import Thread
 from ambari_commons import OSCheck, OSConst, Firewall
-from ambari_server import utils
+from ambari_server import utils, BackupRestore
 
 # debug settings
 VERBOSE = False
@@ -78,6 +78,8 @@ SETUP_NAGIOS_HTTPS_ACTION = "setup-nagios-https"
 ENCRYPT_PASSWORDS_ACTION = "encrypt-passwords"
 SETUP_SECURITY_ACTION = "setup-security"
 REFRESH_STACK_HASH_ACTION = "refresh-stack-hash"
+BACKUP_ACTION = "backup"
+RESTORE_ACTION = "restore"
 
 ACTION_REQUIRE_RESTART = [RESET_ACTION, UPGRADE_ACTION, UPGRADE_STACK_ACTION,
                           SETUP_SECURITY_ACTION, LDAP_SETUP_ACTION]
@@ -4387,6 +4389,25 @@ def refresh_stack_hash():
                                                 resources_location, str(ex))
     raise FatalException(-1, msg)
 
+def backup(path):
+  print "Backup requested."
+  if path is None:
+    backup_command = [BackupRestore, 'backup']
+  else:
+    backup_command = [BackupRestore, 'backup', path]
+
+
+  BackupRestore.main(backup_command)
+
+def restore(path):
+  print "Restore requested."
+  if path is None:
+    restore_command = [BackupRestore, 'restore']
+  else:
+    restore_command = [BackupRestore, 'restore', path]
+
+
+  BackupRestore.main(restore_command)
 
 #
 # Main.
@@ -4547,6 +4568,8 @@ def main():
 
   if action == UPGRADE_STACK_ACTION:
     possible_args_numbers = [2,4] # OR
+  elif action == BACKUP_ACTION or action == RESTORE_ACTION:
+    possible_args_numbers = [1,2]
   else:
     possible_args_numbers = [1]
 
@@ -4593,6 +4616,18 @@ def main():
       need_restart = setup_security(options)
     elif action == REFRESH_STACK_HASH_ACTION:
       refresh_stack_hash()
+    elif action == BACKUP_ACTION:
+      if len(args) == 2:
+        path = args[1]
+      else:
+        path = None
+      backup(path)
+    elif action == RESTORE_ACTION:
+      if len(args) == 2:
+        path = args[1]
+      else:
+        path = None
+      restore(path)
     else:
       parser.error("Invalid action")
 

ambari-server/src/main/python/ambari_server/BackupRestore.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env ambari-python-wrap
+
+'''
+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 zipfile
+import os
+
+# Default values are hardcoded here
+BACKUP_PROCESS = 'backup'
+RESTORE_PROCESS = 'restore'
+SUPPORTED_PROCESSES = [BACKUP_PROCESS, RESTORE_PROCESS]
+
+# The list of files where the ambari server state is kept on the filesystem
+AMBARI_FILESYSTEM_STATE = ["/etc/ambari-server/conf",
+                           "/var/lib/ambari-server/resources",
+                           "/var/run/ambari-server/bootstrap/",
+                           "/var/run/ambari-server/stack-recommendations"]
+
+# What to use when no path/archive is specified
+DEFAULT_ARCHIVE = "/var/lib/ambari-server/Ambari_State_Backup.zip"
+
+# Responsible for managing the Backup/Restore process
+class BackupRestore:
+  def __init__(self, state_file_list, zipname, zip_folder_path):
+    """
+    Zip file creator
+    :param state_file_list: the list of files where the Ambari State is kept on the filesystem
+    :param zipname: the name of the archive to use
+    :param zip_folder_path: the path of the archive
+    :return:
+    """
+    self.state_file_list = state_file_list
+    self.zipname = zipname
+    self.zip_folder_path = zip_folder_path
+
+  def perform_backup(self):
+    """
+    Used to perform the actual backup, by creating the zip archive
+    :return:
+    """
+    try:
+      print("Creating zip file...")
+      zipf = zipfile.ZipFile(self.zip_folder_path + self.zipname, 'w')
+      zipdir(zipf, self.state_file_list, self.zipname)
+    except Exception, e:
+      sys.exit("Could not create zip file. Details: " + str(e))
+
+    print("Zip file created at " + self.zip_folder_path + self.zipname)
+
+  def perform_restore(self):
+    """
+    Used to perform the restore process
+    :return:
+    """
+    try:
+      print("Extracting the archive " + self.zip_folder_path + self.zipname)
+      unzip(self.zip_folder_path + self.zipname, '/')
+    except Exception, e:
+      sys.exit("Could not extract the zipfile " + self.zip_folder_path + self.zipname
+               + " Details: " + str(e))
+
+
+def unzip(source_filename, dest_dir):
+  """
+  Zip archive extractor
+  :param source_filename: the absolute path of the file to unzip
+  :param dest_dir: the destination of the zip content
+  :return:
+  """
+  zf = zipfile.ZipFile(source_filename)
+  try:
+    zf.extractall(dest_dir)
+  except Exception, e:
+    print("A problem occurred while unzipping. Details: " + str(e))
+    raise e
+  finally:
+    zf.close()
+
+
+def zipdir(zipf, state_file_list, zipname):
+  """
+  Used to archive the specified directory
+  :param zipf: the zipfile
+  :param state_file_list: the file list to archive
+  :param zipname: the name of the zip
+  :return:
+  """
+  try:
+    for path in state_file_list:
+      for root, dirs, files in os.walk(path):
+        for file in files:
+          if not file == zipname:
+            zipf.write(os.path.join(root, file))
+  except Exception, e:
+    print("A problem occurred while unzipping. Details: " + str(e))
+    raise e
+  finally:
+    zipf.close()
+
+def print_usage():
+  """
+  Usage instructions
+  :return:
+  """
+  print("Usage: python BackupRestore.py <processType> [zip-folder-path|zip-file-path]\n\n"
+        + "    processType - backup : backs up the filesystem state of the Ambari server into a zip file\n"
+        + "    processType - restore : restores the filesystem state of the Ambari server\n"
+        + "    [zip-folder-path] used with backup specifies the path of the folder where the zip file to be created\n"
+        + "    [zip-folder-path] used with restore specifies the path of the Ambari folder where the zip file to restore from is located\n")
+
+
+def validate_folders(folders):
+  """
+  Used to validate folder existence on the machine
+  :param folders: folder list containing paths to validate
+  :return:
+  """
+  for folder in folders:
+    if not os.path.isdir(folder):
+      sys.exit("Error while validating folders. Folder " + folder + " does not exist.")
+
+def retrieve_path_and_zipname(archive_absolute_path):
+  target = {'path': None , 'zipname': None}
+  try:
+    elements = archive_absolute_path.split("/")
+    if elements is not None and len(elements)>0:
+      target['zipname'] = elements[len(elements)-1]
+      target['path'] = archive_absolute_path.replace(elements[len(elements)-1], "")
+  except Exception, e:
+    sys.exit("Could not retrieve path and zipname from the absolute path " + archive_absolute_path + ". Please check arguments."
+             + " Details: " + str(e))
+
+  return target
+
+def main(argv=None):
+  # Arg checks
+  if len(argv) != 3 and len(argv) != 2:
+    print_usage()
+    sys.exit("Invalid usage.")
+  else:
+    process_type = argv[1]
+    if not (SUPPORTED_PROCESSES.__contains__(process_type)):
+      sys.exit("Unsupported process type: " + process_type)
+    # if no archive is specified
+    if len(argv) == 2:
+      print "No path specified. Will use " + DEFAULT_ARCHIVE
+      location_data = retrieve_path_and_zipname(DEFAULT_ARCHIVE)
+    else:
+      location_data = retrieve_path_and_zipname(argv[2])
+
+    validate_folders([location_data['path']])
+    zip_file_path = location_data['path']
+    ambari_backup_zip_filename = location_data['zipname']
+
+  backup_restore = BackupRestore(AMBARI_FILESYSTEM_STATE, ambari_backup_zip_filename, zip_file_path)
+
+  print(process_type.title() + " process initiated.")
+  if process_type == BACKUP_PROCESS:
+    validate_folders(AMBARI_FILESYSTEM_STATE)
+    backup_restore.perform_backup()
+    print(BACKUP_PROCESS.title() + " complete.")
+  if process_type == RESTORE_PROCESS:
+    backup_restore.perform_restore()
+    print(RESTORE_PROCESS.title() + " complete.")
+
+
+if __name__ == '__main__':
+  main(sys.argv)
+

ambari-server/src/test/python/TestAmbariServer.py

@@ -32,6 +32,7 @@ import platform
 import shutil
 from pwd import getpwnam
 from ambari_server.resourceFilesKeeper import ResourceFilesKeeper, KeeperException
+from ambari_server import BackupRestore
  
 # We have to use this import HACK because the filename contains a dash
 from ambari_commons import Firewall, OSCheck, OSConst, FirewallChecks
@@ -400,6 +401,61 @@ class TestAmbariServer(TestCase):
 
     self.assertTrue(ambari_server.SERVER_DEBUG_MODE)
 
+  @patch.object(ambari_server, 'setup')
+  @patch.object(ambari_server, 'start')
+  @patch.object(ambari_server, 'stop')
+  @patch.object(ambari_server, 'reset')
+  @patch.object(ambari_server, 'backup')
+  @patch.object(ambari_server, 'restore')
+  @patch('optparse.OptionParser')
+  def test_main_test_backup(self, OptionParserMock, restore_mock, backup_mock, reset_method, stop_method,
+                           start_method, setup_method):
+    opm = OptionParserMock.return_value
+    options = MagicMock()
+    args = ["backup"]
+    opm.parse_args.return_value = (options, args)
+
+    options.dbms = None
+    options.sid_or_sname = "sname"
+    ambari_server.main()
+
+    self.assertTrue(backup_mock.called)
+    self.assertFalse(restore_mock.called)
+    self.assertFalse(setup_method.called)
+    self.assertFalse(start_method.called)
+    self.assertFalse(stop_method.called)
+    self.assertFalse(reset_method.called)
+
+    self.assertFalse(False, ambari_server.VERBOSE)
+    self.assertFalse(False, ambari_server.SILENT)
+
+  @patch.object(ambari_server, 'setup')
+  @patch.object(ambari_server, 'start')
+  @patch.object(ambari_server, 'stop')
+  @patch.object(ambari_server, 'reset')
+  @patch.object(ambari_server, 'backup')
+  @patch.object(ambari_server, 'restore')
+  @patch('optparse.OptionParser')
+  def test_main_test_restore(self, OptionParserMock, restore_mock, backup_mock, reset_method, stop_method,
+                            start_method, setup_method):
+    opm = OptionParserMock.return_value
+    options = MagicMock()
+    args = ["restore"]
+    opm.parse_args.return_value = (options, args)
+
+    options.dbms = None
+    options.sid_or_sname = "sname"
+    ambari_server.main()
+
+    self.assertTrue(restore_mock.called)
+    self.assertFalse(backup_mock.called)
+    self.assertFalse(setup_method.called)
+    self.assertFalse(start_method.called)
+    self.assertFalse(stop_method.called)
+    self.assertFalse(reset_method.called)
+
+    self.assertFalse(False, ambari_server.VERBOSE)
+    self.assertFalse(False, ambari_server.SILENT)
 
   @patch.object(ambari_server, 'setup')
   @patch.object(ambari_server, 'start')
@@ -2941,6 +2997,25 @@ MIIFHjCCAwYCCQDpHKOBI+Lt0zANBgkqhkiG9w0BAQUFADBRMQswCQYDVQQGEwJV
     self.assertTrue(killMock.called)
     self.assertTrue(removeMock.called)
 
+  @patch.object(BackupRestore, "main")
+  def test_backup(self, bkrestore_mock):
+    ambari_server.backup("/some/path/file.zip")
+    self.assertTrue(bkrestore_mock.called)
+
+  @patch.object(BackupRestore, "main")
+  def test_backup_no_path(self, bkrestore_mock):
+    ambari_server.backup(None)
+    self.assertTrue(bkrestore_mock.called)
+
+  @patch.object(BackupRestore, "main")
+  def test_restore(self, bkrestore_mock):
+    ambari_server.restore("/some/path/file.zip")
+    self.assertTrue(bkrestore_mock.called)
+
+  @patch.object(BackupRestore, "main")
+  def test_restore_no_path(self, bkrestore_mock):
+    ambari_server.restore(None)
+    self.assertTrue(bkrestore_mock.called)
 
   @patch.object(ambari_server, "is_root")
   @patch.object(ambari_server, "check_database_name_property")

ambari-server/src/test/python/TestBackupRestore.py

@@ -0,0 +1,93 @@
+'''
+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 unittest
+
+import mock
+from mock.mock import MagicMock, patch
+from ambari_server import BackupRestore
+
+
+class TestBKRestore(unittest.TestCase):
+
+  @patch('os.walk')
+  @patch('zipfile.ZipFile')
+  @patch.object(BackupRestore, "zipdir")
+  def test_perform_backup(self, zipdir_mock, zipfile_mock, oswalk_mock):
+    #set up the mocks
+    files_mock = MagicMock()
+    files_mock.__iter__.return_value = ['file1', 'file2']
+    path_mock =  MagicMock()
+    zipname_mock = MagicMock()
+
+
+    bkRestore = BackupRestore.BackupRestore(files_mock, zipname_mock, path_mock)
+    bkRestore.perform_backup()
+
+    self.assertEqual(zipfile_mock.call_count, 1)
+    self.assertEqual(zipdir_mock.call_count, 1)
+
+    zipfile_mock.side_effect = Exception('Invalid path!')
+    try:
+      bkRestore.perform_backup()
+      self.fail("should throw exception")
+    except:
+      self.assertTrue(True)
+
+
+  @patch('zipfile.ZipFile')
+  @patch.object(BackupRestore, "unzip")
+  def test_perform_restore(self, unzip_mock, zipfile_mock):
+    #set up the mocks
+    path_mock =  MagicMock()
+    zipname_mock = MagicMock()
+    files_mock = MagicMock()
+    files_mock.__iter__.return_value = ['file1', 'file2']
+
+
+    bkRestore = BackupRestore.BackupRestore(files_mock, zipname_mock, path_mock)
+    bkRestore.perform_restore()
+
+    self.assertEqual(unzip_mock.call_count, 1)
+
+    unzip_mock.side_effect = Exception('Invalid path!')
+    try:
+      bkRestore.perform_restore()
+      self.fail("should throw exception")
+    except:
+      self.assertTrue(True)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+