SWARM-386

Author: Eric Diven

bin/install-docker.sh

@@ -18,21 +18,26 @@
 import logging
 import os
 from StringIO import StringIO
+from contextlib import closing
 from fabric.contrib import files
 from fabric.decorators import task, serial
 from fabric.operations import get
+from fabric.operations import put
 from fabric.state import env
 from fabric.utils import abort, warn
 import prestoadmin.deploy
 from prestoadmin.topology import requires_topology
 from prestoadmin.util import constants
+from prestoadmin.util.filesystem import ensure_parent_directories_exist
 
 
 CONFIG_PROPERTIES = "config.properties"
 LOG_PROPERTIES = "log.properties"
 JVM_CONFIG = "jvm.config"
 NODE_PROPERTIES = "node.properties"
 
+ALL_CONFIG = [CONFIG_PROPERTIES, LOG_PROPERTIES, JVM_CONFIG, NODE_PROPERTIES]
+
 _LOGGER = logging.getLogger(__name__)
 
 __all__ = ['deploy', 'show']
@@ -69,15 +74,51 @@ def deploy(rolename=None):
             abort("Invalid Argument. Possible values: coordinator, workers")
 
 
-def configuration_show(file_name, should_warn=True):
-    file_path = os.path.join(constants.REMOTE_CONF_DIR, file_name)
-    if not files.exists(file_path):
+def gather_directory(target_directory, allow_overwrite=False):
+    fetch_all(target_directory, allow_overwrite=allow_overwrite)
+
+
+def deploy_all(source_directory, should_warn=True):
+    host_config_dir = os.path.join(source_directory, env.host)
+    for file_name in ALL_CONFIG:
+        local_config_file = os.path.join(host_config_dir, file_name)
+        if not os.path.exists(local_config_file):
+            if should_warn:
+                warn("No configuration file found for %s at %s"
+                     % (env.host, local_config_file))
+            continue
+        remote_config_file = os.path.join(constants.REMOTE_CONF_DIR, file_name)
+        put(local_config_file, remote_config_file)
+
+
+def fetch_all(target_directory, allow_overwrite=False):
+    host_config_dir = os.path.join(target_directory, env.host)
+    for file_name in ALL_CONFIG:
+        local_config_file = os.path.join(host_config_dir, file_name)
+        ensure_parent_directories_exist(local_config_file)
+        if not allow_overwrite and os.path.exists(local_config_file):
+            abort("Refusing to overwrite %s" % (local_config_file))
+        configuration_fetch(file_name, local_config_file)
+
+
+def configuration_fetch(file_name, config_destination, should_warn=True):
+    remote_file_path = os.path.join(constants.REMOTE_CONF_DIR, file_name)
+    if not files.exists(remote_file_path):
         if should_warn:
             warn("No configuration file found for %s at %s"
-                 % (env.host, file_path))
+                 % (env.host, remote_file_path))
+        return None
     else:
-        file_content_buffer = StringIO()
-        get(file_path, file_content_buffer)
+        get(remote_file_path, config_destination)
+        return remote_file_path
+
+
+def configuration_show(file_name, should_warn=True):
+    with closing(StringIO()) as file_content_buffer:
+        file_path = configuration_fetch(file_name, file_content_buffer,
+                                        should_warn)
+        if file_path is None:
+            return
         config_values = file_content_buffer.getvalue()
         file_content_buffer.close()
         print ("\n%s: Configuration file at %s:" % (env.host, file_path))

prestoadmin/configure_cmds.py

@@ -18,14 +18,16 @@
 import logging
 import errno
 
-from fabric.api import task, env
+from fabric.api import task, env, abort
 from fabric.context_managers import hide
-from fabric.operations import sudo, os, put
+from fabric.contrib import files
+from fabric.operations import sudo, os, put, get
 import fabric.utils
 
 from prestoadmin.util import constants
 from prestoadmin.util.exception import ConfigFileNotFoundError, \
     ConfigurationError
+from prestoadmin.util.filesystem import ensure_directory_exists
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -40,6 +42,17 @@ def deploy_files(filenames, local_dir, remote_dir):
         put(os.path.join(local_dir, name), remote_dir, use_sudo=True)
 
 
+def gather_connectors(local_config_dir, allow_overwrite=False):
+    local_catalog_dir = os.path.join(local_config_dir, env.host, 'catalog')
+    if not allow_overwrite and os.path.exists(local_catalog_dir):
+        abort("Refusing to overwrite %s" % local_catalog_dir)
+    ensure_directory_exists(local_catalog_dir)
+    if files.exists(constants.REMOTE_CATALOG_DIR):
+        return get(constants.REMOTE_CATALOG_DIR, local_catalog_dir)
+    else:
+        return []
+
+
 def validate(filenames):
     for name in filenames:
         file_path = os.path.join(constants.CONNECTORS_DIR, name)

prestoadmin/connector.py

@@ -61,9 +61,17 @@ def check_if_valid_rpm(local_path):
 
 
 def deploy_install(local_path):
+    deploy_action(local_path, rpm_install)
+
+
+def deploy_upgrade(local_path):
+    deploy_action(local_path, rpm_upgrade)
+
+
+def deploy_action(local_path, rpm_action):
     check_if_valid_rpm(local_path)
     deploy(local_path)
-    rpm_install(os.path.basename(local_path))
+    rpm_action(os.path.basename(local_path))
 
 
 def deploy(local_path=None):
@@ -86,6 +94,23 @@ def rpm_install(rpm_name):
         nodeps = '--nodeps '
 
     ret = sudo('rpm -i %s%s' %
-               (nodeps, constants.REMOTE_PACKAGES_PATH + "/" + rpm_name))
+               (nodeps, os.path.join(constants.REMOTE_PACKAGES_PATH,
+                                     rpm_name)))
     if ret.succeeded:
         print("Package installed successfully on: " + env.host)
+
+
+def rpm_upgrade(rpm_name):
+    _LOGGER.info("Upgrading the rpm")
+    nodeps = ''
+    if env.nodeps:
+        nodeps = '--nodeps '
+
+    package_path = os.path.join(constants.REMOTE_PACKAGES_PATH, rpm_name)
+    package_name = sudo('rpm -qp --queryformat \'%%{NAME}\' %s'
+                        % package_path, quiet=True)
+
+    ret_uninstall = sudo('rpm -e %s%s' % (nodeps, package_name))
+    ret_install = sudo('rpm -i %s%s' % (nodeps, package_path))
+    if ret_uninstall.succeeded and ret_install.succeeded:
+        print("Package upgraded successfully on: " + env.host)

prestoadmin/package.py

@@ -37,9 +37,12 @@
 from prestoadmin.util.exception import ConfigFileNotFoundError
 from prestoadmin.util.fabricapi import get_host_list, get_coordinator_role
 from prestoadmin.util.service_util import lookup_port
+
+from tempfile import mkdtemp
 import util.filesystem
 
-__all__ = ['install', 'uninstall', 'start', 'stop', 'restart', 'status']
+__all__ = ['install', 'uninstall', 'upgrade', 'start', 'stop', 'restart',
+           'status']
 
 INIT_SCRIPTS = '/etc/init.d/presto'
 RETRY_TIMEOUT = 120
@@ -125,6 +128,50 @@ def uninstall():
         print('Package uninstalled successfully on: ' + env.host)
 
 
+@task
+@requires_topology
+def upgrade(local_package_path, local_config_dir=None):
+    """
+    Copy and upgrade a new presto-server rpm to all of the nodes in the
+    cluster. Retains existing node configuration.
+
+    The existing topology information is read from the config.json file.
+    Unlike install, there is no provision to supply topology information
+    interactively.
+
+    The existing cluster configuration is collected from the nodes on the
+    cluster and stored on the host running presto-admin. After the
+    presto-server packages have been upgraded, presto-admin pushes the
+    collected configuration back out to the hosts on the cluster.
+
+    Note that the configuration files in /etc/opt/prestoadmin are not updated
+    during upgrade.
+
+    :param local_package_path - Absolute path to the presto rpm to be
+                                installed
+    :param local_config_dir -   (optional) Directory to store the cluster
+                                configuration in. If not specified, a temp
+                                directory is used.
+    """
+    stop()
+
+    if not local_config_dir:
+        local_config_dir = mkdtemp()
+        print('Saving cluster configuration to %s' % local_config_dir)
+
+    configure_cmds.gather_directory(local_config_dir)
+    filenames = connector.gather_connectors(local_config_dir)
+
+    package.deploy_upgrade(local_package_path)
+
+    configure_cmds.deploy_all(local_config_dir)
+    connector.deploy_files(
+        filenames,
+        os.path.join(local_config_dir, env.host, 'catalog'),
+        constants.REMOTE_CATALOG_DIR
+    )
+
+
 def service(control=None):
     if check_presto_version() != '':
         return False

prestoadmin/server.py

@@ -30,6 +30,14 @@ def ensure_parent_directories_exist(path):
             raise e
 
 
+def ensure_directory_exists(path):
+    try:
+        os.makedirs(path)
+    except OSError as e:
+        if e.errno != errno.EEXIST:
+            raise e
+
+
 def write_to_file_if_not_exists(content, path):
     flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY
 

prestoadmin/util/filesystem.py

@@ -169,6 +169,9 @@ def stop_host(self, container_name):
         self.client.stop(container_name)
         self.client.wait(container_name)
 
+    def start_host(self, container_name):
+        self.client.start(container_name)
+
     def get_down_hostname(self, host_name):
         return host_name
 

tests/docker_cluster.py

@@ -515,11 +515,8 @@ def retry(self, method_to_check):
                 result = method_to_check()
                 # No exception thrown, success
                 return result
-            except (AssertionError, PrestoError):
+            except (AssertionError, PrestoError, OSError):
                 pass
-            except OSError as e:
-                if not e.errno == 7:
-                    raise
             sleep(RETRY_INTERVAL)
             time_spent_waiting += RETRY_INTERVAL
         return method_to_check()

tests/product/base_product_case.py

@@ -0,0 +1,95 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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
+
+from nose.plugins.attrib import attr
+
+from tests.product.base_product_case import BaseProductTestCase, \
+    LOCAL_RESOURCES_DIR, docker_only
+
+
+class TestServerUpgrade(BaseProductTestCase):
+
+    def setUp(self):
+        super(TestServerUpgrade, self).setUp()
+        self.setup_cluster('presto')
+
+    def start_and_assert_started(self):
+        cmd_output = self.run_prestoadmin('server start')
+        process_per_host = self.get_process_per_host(cmd_output)
+        self.assert_started(process_per_host)
+
+    def assert_upgraded_to_dummy_rpm(self, hosts):
+        for container in hosts:
+            # Still should have the same configs
+            self.assert_installed(container)
+            self.assert_has_default_config(container)
+            self.assert_has_default_connector(container)
+
+            # However, dummy_rpm.rpm removes /usr/lib/presto/lib and
+            # /usr/lib/presto/lib/plugin
+            self.assert_path_removed(container, '/usr/lib/presto/lib')
+            self.assert_path_removed(container, '/usr/lib/presto/lib/plugin')
+
+            # And adds /usr/lib/presto/README.txt
+            self.assert_path_exists(container, '/usr/lib/presto/README.txt')
+
+            # And modifies the text of the readme in
+            # /usr/shared/doc/presto/README.txt
+            self.assert_file_content_regex(
+                container,
+                '/usr/shared/doc/presto/README.txt',
+                r'.*New line of text here.$'
+            )
+
+    def copy_upgrade_rpm_to_cluster(self):
+        self.cluster.copy_to_host(
+            os.path.join(LOCAL_RESOURCES_DIR, 'dummy-rpm.rpm'),
+            self.cluster.master)
+        path_on_cluster = os.path.join(
+            self.cluster.mount_dir, 'dummy-rpm.rpm')
+        return path_on_cluster
+
+    @attr('smoketest')
+    def test_upgrade(self):
+        self.start_and_assert_started()
+
+        self.run_prestoadmin('configuration deploy')
+        for container in self.cluster.all_hosts():
+            self.assert_installed(container)
+            self.assert_has_default_config(container)
+            self.assert_has_default_connector(container)
+
+        path_on_cluster = self.copy_upgrade_rpm_to_cluster()
+        self.upgrade_and_assert_success(path_on_cluster)
+
+    def upgrade_and_assert_success(self, path_on_cluster):
+        self.run_prestoadmin('server upgrade ' + path_on_cluster)
+        self.assert_upgraded_to_dummy_rpm(self.cluster.all_hosts())
+
+    @docker_only
+    def test_rolling_upgrade(self):
+        # Test that if a node is down, and then you upgrade again, it works
+        self.run_prestoadmin('configuration deploy')
+
+        self.cluster.stop_host(self.cluster.slaves[0])
+        path_on_cluster = self.copy_upgrade_rpm_to_cluster()
+        self.run_prestoadmin('server upgrade ' + path_on_cluster)
+        running_hosts = self.cluster.all_hosts()[:]
+        running_hosts.remove(self.cluster.slaves[0])
+        self.assert_upgraded_to_dummy_rpm(running_hosts)
+
+        self.cluster.start_host(self.cluster.slaves[0])
+        self.retry(lambda: self.upgrade_and_assert_success(path_on_cluster))

tests/product/resources/dummy-rpm.rpm

@@ -54,5 +54,6 @@ Commands:
     server status
     server stop
     server uninstall
+    server upgrade
     topology show
 

tests/product/test_server_upgrade.py

@@ -28,5 +28,6 @@ Commands:
     server status
     server stop
     server uninstall
+    server upgrade
     topology show