Add support for HP Vertica datastore in Trove

A specification for this change was submitted for review in
https://review.openstack.org/#/c/151126/

The following features have been implemented in this patchset:
- Launch/Reboot/Terminate
- Included unit tests.

Workflow for building instance is as follows:
- Guest instance is booted using taskmanager.
- Once the guest instance is active in nova,
  it receives "prepare" message.
- Mount the data disk on device_path.
- Check if vertica packages have been installed,
  install_if_needed().
- Run Vertica pre-install test, prepare_for_install_vertica().
- Run install_vertica command to install and
  configure Vertica, install_vertica().
- Create a database named db_srvr, create_db().

New Files:
- A new directory, vertica, has been created for manager code
  under guestagent/datastore/experimental.
- test_vertica_manager.py contains the unit-tests
  for vertica-manager.

Change-Id: I30bc4fd597d30c817bf0a8adc1109ca1f6495096
Implements: blueprint vertica-db-support

trove/common/cfg.py

@@ -326,7 +326,8 @@ common_opts = [
                          'couchbase': 'fa62fe68-74d9-4779-a24e-36f19602c415',
                          'mongodb': 'c8c907af-7375-456f-b929-b637ff9209ee',
                          'postgresql': 'ac277e0d-4f21-40aa-b347-1ea31e571720',
-                         'couchdb': 'f0a9ab7b-66f7-4352-93d7-071521d44c7c'},
+                         'couchdb': 'f0a9ab7b-66f7-4352-93d7-071521d44c7c',
+                         'vertica': 'a8d805ae-a3b2-c4fd-gb23-b62cee5201ae'},
                 help='Unique ID to tag notification events.'),
     cfg.StrOpt('nova_proxy_admin_user', default='',
                help="Admin username used to connect to Nova.", secret=True),
@@ -807,6 +808,42 @@ couchdb_opts = [
                 'instance-create as the "password" field.'),
 ]
 
+# Vertica
+vertica_group = cfg.OptGroup(
+    'vertica', title='Vertica options',
+    help="Oslo option group designed for Vertica datastore")
+vertica_opts = [
+    cfg.ListOpt('tcp_ports', default=["5433"],
+                help='List of TCP ports and/or port ranges to open '
+                     'in the security group (only applicable '
+                     'if trove_security_groups_support is True).'),
+    cfg.ListOpt('udp_ports', default=["5433"],
+                help='List of UDP ports and/or port ranges to open '
+                     'in the security group (only applicable '
+                     'if trove_security_groups_support is True).'),
+    cfg.StrOpt('backup_strategy', default=None,
+               help='Default strategy to perform backups.'),
+    cfg.DictOpt('backup_incremental_strategy', default={},
+                help='Incremental Backup Runner based on the default '
+                'strategy. For strategies that do not implement an '
+                'incremental, the runner will use the default full backup.'),
+    cfg.StrOpt('replication_strategy', default=None,
+               help='Default strategy for replication.'),
+    cfg.StrOpt('mount_point', default='/var/lib/vertica',
+               help="Filesystem path for mounting "
+               "volumes if volume support is enabled."),
+    cfg.BoolOpt('volume_support', default=True,
+                help='Whether to provision a Cinder volume for datadir.'),
+    cfg.StrOpt('device_path', default='/dev/vdb',
+               help='Device path for volume if volume support is enabled.'),
+    cfg.StrOpt('backup_namespace', default=None,
+               help='Namespace to load backup strategies from.'),
+    cfg.StrOpt('restore_namespace', default=None,
+               help='Namespace to load restore strategies from.'),
+    cfg.IntOpt('readahead_size', default=2048,
+               help='Size(MB) to be set as readahead_size for data volume'),
+]
+
 # RPC version groups
 upgrade_levels = cfg.OptGroup(
     'upgrade_levels',
@@ -841,6 +878,7 @@ CONF.register_group(couchbase_group)
 CONF.register_group(mongodb_group)
 CONF.register_group(postgresql_group)
 CONF.register_group(couchdb_group)
+CONF.register_group(vertica_group)
 
 CONF.register_opts(mysql_opts, mysql_group)
 CONF.register_opts(percona_opts, percona_group)
@@ -850,6 +888,7 @@ CONF.register_opts(couchbase_opts, couchbase_group)
 CONF.register_opts(mongodb_opts, mongodb_group)
 CONF.register_opts(postgresql_opts, postgresql_group)
 CONF.register_opts(couchdb_opts, couchdb_group)
+CONF.register_opts(vertica_opts, vertica_group)
 
 CONF.register_opts(rpcapi_cap_opts, upgrade_levels)
 

trove/guestagent/datastore/experimental/vertica/manager.py

@@ -0,0 +1,192 @@
+#Copyright [2015] Hewlett-Packard Development Company, L.P.
+#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 trove.common import cfg
+from trove.common import exception
+from trove.common import instance as rd_ins
+from trove.guestagent import volume
+from trove.guestagent import dbaas
+from trove.guestagent.datastore.experimental.vertica.service import (
+    VerticaAppStatus)
+from trove.guestagent.datastore.experimental.vertica.service import VerticaApp
+from trove.openstack.common import log as logging
+from trove.common.i18n import _
+from trove.openstack.common import periodic_task
+
+LOG = logging.getLogger(__name__)
+CONF = cfg.CONF
+MANAGER = 'vertica' if not CONF.datastore_manager else CONF.datastore_manager
+
+
+class Manager(periodic_task.PeriodicTasks):
+
+    def __init__(self):
+        self.appStatus = VerticaAppStatus()
+        self.app = VerticaApp(self.appStatus)
+
+    @periodic_task.periodic_task(ticks_between_runs=3)
+    def update_status(self, context):
+        """Update the status of the Vertica service."""
+        self.appStatus.update()
+
+    def prepare(self, context, packages, databases, memory_mb, users,
+                device_path=None, mount_point=None, backup_info=None,
+                config_contents=None, root_password=None, overrides=None,
+                cluster_config=None, path_exists_function=os.path.exists):
+        """Makes ready DBAAS on a Guest container."""
+        try:
+            LOG.info(_("Setting instance status to BUILDING."))
+            self.appStatus.begin_install()
+            if device_path:
+                device = volume.VolumeDevice(device_path)
+                # unmount if device is already mounted
+                device.unmount_device(device_path)
+                device.format()
+                if path_exists_function(mount_point):
+                    #rsync any existing data
+                    device.migrate_data(mount_point)
+                    # mount the volume
+                    device.mount(mount_point)
+                    LOG.debug("Mounted the volume.")
+            self.app.install_if_needed(packages)
+            self.app.prepare_for_install_vertica()
+            self.app.install_vertica()
+            self.app.create_db()
+            self.app.complete_install_or_restart()
+            LOG.info(_('Completed setup of Vertica database instance.'))
+        except Exception:
+            LOG.exception(_('Cannot prepare Vertica database instance.'))
+            self.appStatus.set_status(rd_ins.ServiceStatuses.FAILED)
+            raise
+
+    def restart(self, context):
+        LOG.debug("Restarting the database.")
+        self.app.restart()
+        LOG.debug("Restarted the database.")
+
+    def get_filesystem_stats(self, context, fs_path):
+        """Gets the filesystem stats for the path given."""
+        LOG.debug("Finding the file-systems stats.")
+        mount_point = CONF.get(MANAGER).mount_point
+        return dbaas.get_filesystem_volume_stats(mount_point)
+
+    def stop_db(self, context, do_not_start_on_reboot=False):
+        LOG.debug("Stopping the database.")
+        self.app.stop_db(do_not_start_on_reboot=do_not_start_on_reboot)
+        LOG.debug("Stopped the database.")
+
+    def mount_volume(self, context, device_path=None, mount_point=None):
+        LOG.debug("Mounting the volume.")
+        device = volume.VolumeDevice(device_path)
+        device.mount(mount_point, write_to_fstab=False)
+        LOG.debug("Mounted the volume.")
+
+    def unmount_volume(self, context, device_path=None, mount_point=None):
+        LOG.debug("Unmounting the volume.")
+        device = volume.VolumeDevice(device_path)
+        device.unmount(mount_point)
+        LOG.debug("Unmounted the volume.")
+
+    def resize_fs(self, context, device_path=None, mount_point=None):
+        LOG.debug("Resizing the filesystem.")
+        device = volume.VolumeDevice(device_path)
+        device.resize_fs(mount_point)
+        LOG.debug("Resized the filesystem.")
+
+    def reset_configuration(self, context, configuration):
+        LOG.debug("Resetting Vertica configuration.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='reset_configuration', datastore=MANAGER)
+
+    def change_passwords(self, context, users):
+        LOG.debug("Changing password.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='change_passwords', datastore=MANAGER)
+
+    def update_attributes(self, context, username, hostname, user_attrs):
+        LOG.debug("Updating database attributes.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='update_attributes', datastore=MANAGER)
+
+    def create_database(self, context, databases):
+        LOG.debug("Creating database.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='create_database', datastore=MANAGER)
+
+    def create_user(self, context, users):
+        LOG.debug("Creating user.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='create_user', datastore=MANAGER)
+
+    def delete_database(self, context, database):
+        LOG.debug("Deleting database.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='delete_database', datastore=MANAGER)
+
+    def delete_user(self, context, user):
+        LOG.debug("Deleting user.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='delete_user', datastore=MANAGER)
+
+    def get_user(self, context, username, hostname):
+        LOG.debug("Getting user.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='get_user', datastore=MANAGER)
+
+    def grant_access(self, context, username, hostname, databases):
+        LOG.debug("Granting acccess.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='grant_access', datastore=MANAGER)
+
+    def revoke_access(self, context, username, hostname, database):
+        LOG.debug("Revoking access.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='revoke_access', datastore=MANAGER)
+
+    def list_access(self, context, username, hostname):
+        LOG.debug("Listing access.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='list_access', datastore=MANAGER)
+
+    def list_databases(self, context, limit=None, marker=None,
+                       include_marker=False):
+        LOG.debug("Listing databases.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='list_databases', datastore=MANAGER)
+
+    def list_users(self, context, limit=None, marker=None,
+                   include_marker=False):
+        LOG.debug("Listing users.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='list_users', datastore=MANAGER)
+
+    def enable_root(self, context):
+        LOG.debug("Enabling root.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='enable_root', datastore=MANAGER)
+
+    def is_root_enabled(self, context):
+        LOG.debug("Checking if root is enabled.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='is_root_enabled', datastore=MANAGER)
+
+    def create_backup(self, context, backup_info):
+        LOG.debug("Creating backup.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='create_backup', datastore=MANAGER)
+
+    def start_db_with_conf_changes(self, context, config_contents):
+        LOG.debug("Starting with configuration changes.")
+        raise exception.DatastoreOperationNotSupported(
+            operation='start_db_with_conf_changes', datastore=MANAGER)

trove/guestagent/datastore/experimental/vertica/service.py

@@ -0,0 +1,221 @@
+#Copyright [2015] Hewlett-Packard Development Company, L.P.
+#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 ConfigParser
+import os
+import tempfile
+
+from trove.common import cfg
+from trove.common import exception
+from trove.common import utils as utils
+from trove.common import instance as rd_instance
+from trove.guestagent.common import operating_system
+from trove.guestagent.datastore import service
+from trove.guestagent import pkg
+from trove.guestagent import volume
+from trove.guestagent.datastore.experimental.vertica import system
+from trove.openstack.common import log as logging
+from trove.common.i18n import _
+
+LOG = logging.getLogger(__name__)
+CONF = cfg.CONF
+packager = pkg.Package()
+DB_NAME = 'db_srvr'
+MOUNT_POINT = CONF.vertica.mount_point
+
+
+class VerticaAppStatus(service.BaseDbStatus):
+
+    def _get_actual_db_status(self):
+        """Get the status of dbaas and report it back."""
+        try:
+            out, err = system.shell_execute(system.STATUS_ACTIVE_DB,
+                                            "dbadmin")
+            if out.strip() == DB_NAME:
+                #UP status is confirmed
+                LOG.info(_("Service Status is RUNNING."))
+                return rd_instance.ServiceStatuses.RUNNING
+            elif out.strip() == "":
+                #nothing returned, means no db running lets verify
+                out, err = system.shell_execute(system.STATUS_DB_DOWN,
+                                                "dbadmin")
+                if out.strip() == DB_NAME:
+                    #DOWN status is confirmed
+                    LOG.info(_("Service Status is SHUTDOWN."))
+                    return rd_instance.ServiceStatuses.SHUTDOWN
+                else:
+                    return rd_instance.ServiceStatuses.UNKNOWN
+        except exception.ProcessExecutionError:
+            LOG.exception(_("Failed to get database status."))
+            return rd_instance.ServiceStatuses.CRASHED
+
+
+class VerticaApp(object):
+    """Prepares DBaaS on a Guest container."""
+
+    def __init__(self, status):
+        self.state_change_wait_time = CONF.state_change_wait_time
+        self.status = status
+
+    def _enable_db_on_boot(self):
+        command = (system.SET_RESTART_POLICY % (DB_NAME, "always"))
+        try:
+            system.shell_execute(command, "dbadmin")
+        except exception.ProcessExecutionError:
+            LOG.exception(_("Failed to enable db on boot."))
+            raise
+
+    def _disable_db_on_boot(self):
+        command = (system.SET_RESTART_POLICY % (DB_NAME, "never"))
+        try:
+            system.shell_execute(command, "dbadmin")
+        except exception.ProcessExecutionError:
+            LOG.exception(_("Failed to disable db on boot."))
+            raise
+
+    def stop_db(self, update_db=False, do_not_start_on_reboot=False):
+        """Stop the database."""
+        LOG.info(_("Stopping Vertica."))
+        if do_not_start_on_reboot:
+            self._disable_db_on_boot()
+        # Using Vertica adminTools to stop db.
+        db_password = self._get_database_password()
+        stop_db_command = (system.STOP_DB % (DB_NAME, db_password))
+        system.shell_execute(stop_db_command, "dbadmin")
+        if not self.status.wait_for_real_status_to_change_to(
+                rd_instance.ServiceStatuses.SHUTDOWN,
+                self.state_change_wait_time, update_db):
+            LOG.error(_("Could not stop Vertica."))
+            self.status.end_install_or_restart()
+            raise RuntimeError("Could not stop Vertica!")
+
+    def start_db(self, update_db=False):
+        """Start the database."""
+        LOG.info(_("Starting Vertica."))
+        self._enable_db_on_boot()
+        # Using Vertica adminTools to start db.
+        db_password = self._get_database_password()
+        start_db_command = (system.START_DB % (DB_NAME, db_password))
+        system.shell_execute(start_db_command, "dbadmin")
+        if not self.status.wait_for_real_status_to_change_to(
+                rd_instance.ServiceStatuses.RUNNING,
+                self.state_change_wait_time, update_db):
+            LOG.error(_("Start up of Vertica failed."))
+            self.status.end_install_or_restart()
+            raise RuntimeError("Could not start Vertica!")
+
+    def restart(self):
+        """Restart the database."""
+        try:
+            self.status.begin_restart()
+            self.stop_db()
+            self.start_db()
+        finally:
+            self.status.end_install_or_restart()
+
+    def create_db(self, members=operating_system.get_ip_address()):
+        """Prepare the guest machine with a Vertica db creation."""
+        LOG.info(_("Creating database on Vertica host."))
+        try:
+            # Create db after install
+            db_password = self._get_database_password()
+            create_db_command = (system.CREATE_DB % (members, DB_NAME,
+                                                     MOUNT_POINT, MOUNT_POINT,
+                                                     db_password))
+            system.shell_execute(create_db_command, "dbadmin")
+        except Exception:
+            LOG.exception(_("Vertica database create failed."))
+        LOG.info(_("Vertica database create completed."))
+
+    def install_vertica(self, members=operating_system.get_ip_address()):
+        """Prepare the guest machine with a Vertica db creation."""
+        LOG.info(_("Installing Vertica Server."))
+        try:
+            # Create db after install
+            install_vertica_cmd = (system.INSTALL_VERTICA % (members,
+                                                             MOUNT_POINT))
+            system.shell_execute(install_vertica_cmd)
+        except exception.ProcessExecutionError:
+            LOG.exception(_("install_vertica failed."))
+        self._generate_database_password()
+        LOG.info(_("install_vertica completed."))
+
+    def complete_install_or_restart(self):
+        self.status.end_install_or_restart()
+
+    def _generate_database_password(self):
+        """Generate and write the password to vertica.cnf file."""
+        config = ConfigParser.ConfigParser()
+        config.add_section('credentials')
+        config.set('credentials', 'dbadmin_password',
+                   utils.generate_random_password())
+        self.write_config(config)
+
+    def write_config(self, config,
+                     unlink_function=os.unlink,
+                     temp_function=tempfile.NamedTemporaryFile):
+        """Write the configuration contents to vertica.cnf file."""
+        LOG.debug('Defining config holder at %s.' % system.VERTICA_CONF)
+        tempfile = temp_function(delete=False)
+        try:
+            config.write(tempfile)
+            tempfile.close()
+            command = (("install -o root -g root -m 644 %(source)s %(target)s"
+                        ) % {'source': tempfile.name,
+                             'target': system.VERTICA_CONF})
+            system.shell_execute(command)
+            unlink_function(tempfile.name)
+        except Exception:
+            unlink_function(tempfile.name)
+            raise
+
+    def read_config(self):
+        """Reads and returns the Vertica config."""
+        try:
+            config = ConfigParser.ConfigParser()
+            config.read(system.VERTICA_CONF)
+            return config
+        except Exception:
+            LOG.exception(_("Failed to read config %s.") % system.VERTICA_CONF)
+            raise RuntimeError
+
+    def _get_database_password(self):
+        """Read the password from vertica.cnf file and return it."""
+        return self.read_config().get('credentials', 'dbadmin_password')
+
+    def install_if_needed(self, packages):
+        """Install Vertica package if needed."""
+        LOG.info(_("Preparing Guest as Vertica Server."))
+        if not packager.pkg_is_installed(packages):
+            LOG.debug("Installing Vertica Package.")
+            packager.pkg_install(packages, None, system.INSTALL_TIMEOUT)
+
+    def _set_readahead_for_disks(self):
+        """This method sets readhead size for disks as needed by Vertica."""
+        device = volume.VolumeDevice(CONF.device_path)
+        device.set_readahead_size(CONF.vertica.readahead_size)
+        LOG.debug("Set readhead size as required by Vertica.")
+
+    def prepare_for_install_vertica(self):
+        """This method executes preparatory methods before
+        executing install_vertica.
+        """
+        command = ("VERT_DBA_USR=dbadmin VERT_DBA_HOME=/home/dbadmin "
+                   "VERT_DBA_GRP=verticadba /opt/vertica/oss/python/bin/python"
+                   " -m vertica.local_coerce")
+        try:
+            self._set_readahead_for_disks()
+            system.shell_execute(command)
+        except exception.ProcessExecutionError:
+            LOG.exception(_("Failed to prepare for install_vertica."))
+            raise

trove/guestagent/datastore/experimental/vertica/system.py

@@ -0,0 +1,39 @@
+#Copyright [2015] Hewlett-Packard Development Company, L.P.
+#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.
+
+from trove.common import utils
+
+CREATE_DB = ("/opt/vertica/bin/adminTools -t create_db -s"
+             " %s -d %s -c %s -D %s -p '%s'")
+INSTALL_VERTICA = ("/opt/vertica/sbin/install_vertica -s %s"
+                   " -d %s -X -N -S default -r"
+                   " /vertica.deb -L CE -Y --no-system-checks")
+STOP_DB = "/opt/vertica/bin/adminTools -t stop_db -F -d %s -p '%s'"
+START_DB = "/opt/vertica/bin/adminTools -t start_db -d %s -p '%s'"
+STATUS_ACTIVE_DB = "/opt/vertica/bin/adminTools -t show_active_db"
+STATUS_DB_DOWN = "/opt/vertica/bin/adminTools -t db_status -s DOWN"
+SET_RESTART_POLICY = ("/opt/vertica/bin/adminTools -t set_restart_policy "
+                      "-d %s -p '%s'")
+VERTICA_CONF = "/etc/vertica.cnf"
+INSTALL_TIMEOUT = 1000
+
+
+def shell_execute(command, command_executor="root"):
+    #This method encapsulates utils.execute for 2 purpose:
+    #1. Helps in safe testing.
+    #2. Helps in executing commands as other user, using their environment.
+
+    #Note: This method uses su because using sudo -i -u <user> <command>
+    #does not works with vertica installer
+    #and it has problems while executing remote commmands.
+    return utils.execute("sudo", "su", "-", command_executor, "-c", command)

trove/guestagent/dbaas.py

@@ -49,6 +49,8 @@ defaults = {
     'trove.guestagent.datastore.experimental.postgresql.manager.Manager',
     'couchdb':
     'trove.guestagent.datastore.experimental.couchdb.manager.Manager',
+    'vertica':
+    'trove.guestagent.datastore.experimental.vertica.manager.Manager',
 }
 CONF = cfg.CONF
 

trove/guestagent/volume.py

@@ -156,6 +156,20 @@ class VolumeDevice(object):
             raise GuestError(_("Could not obtain a list of mount points for "
                                "device: %s") % device_path)
 
+    def set_readahead_size(self, readahead_size,
+                           execute_function=utils.execute):
+        """Set the readahead size of disk."""
+        self._check_device_exists()
+        try:
+            execute_function("sudo", "blockdev", "--setra",
+                             readahead_size, self.device_path)
+        except ProcessExecutionError:
+            LOG.exception(_("Error setting readhead size to %(size)s "
+                            "for device %(device)s.") %
+                          {'size': readahead_size, 'device': self.device_path})
+            raise GuestError(_("Error setting readhead size: %s.") %
+                             self.device_path)
+
 
 class VolumeMountPoint(object):
 

trove/tests/unittests/guestagent/test_dbaas.py

@@ -12,6 +12,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import ConfigParser
 import os
 import tempfile
 from uuid import uuid4
@@ -56,7 +57,13 @@ from trove.guestagent.datastore.experimental.mongodb import (
     service as mongo_service)
 from trove.guestagent.datastore.experimental.mongodb import (
     system as mongo_system)
+from trove.guestagent.datastore.experimental.vertica.service import VerticaApp
+from trove.guestagent.datastore.experimental.vertica.service import (
+    VerticaAppStatus)
+from trove.guestagent.datastore.experimental.vertica import (
+    system as vertica_system)
 from trove.guestagent.db import models
+from trove.guestagent.volume import VolumeDevice
 from trove.instance.models import InstanceServiceStatus
 from trove.tests.unittests.util import util
 
@@ -1028,6 +1035,9 @@ class ServiceRegistryTest(testtools.TestCase):
         self.assertEqual(test_dict.get('couchdb'),
                          'trove.guestagent.datastore.experimental.couchdb.'
                          'manager.Manager')
+        self.assertEqual('trove.guestagent.datastore.experimental.vertica.'
+                         'manager.Manager',
+                         test_dict.get('vertica'))
 
     def test_datastore_registry_with_blank_dict(self):
         datastore_registry_ext_test = dict()
@@ -1055,6 +1065,9 @@ class ServiceRegistryTest(testtools.TestCase):
         self.assertEqual(test_dict.get('couchdb'),
                          'trove.guestagent.datastore.experimental.couchdb.'
                          'manager.Manager')
+        self.assertEqual('trove.guestagent.datastore.experimental.vertica.'
+                         'manager.Manager',
+                         test_dict.get('vertica'))
 
 
 class KeepAliveConnectionTest(testtools.TestCase):
@@ -2029,3 +2042,238 @@ class MongoDBAppTest(testtools.TestCase):
         self.mongoDbApp.install_if_needed(['package'])
         packager_mock.pkg_install.assert_any_call(ANY, {}, ANY)
         self.assert_reported_status(rd_instance.ServiceStatuses.NEW)
+
+
+class VerticaAppStatusTest(testtools.TestCase):
+
+    def setUp(self):
+        super(VerticaAppStatusTest, self).setUp()
+        util.init_db()
+        self.FAKE_ID = str(uuid4())
+        InstanceServiceStatus.create(instance_id=self.FAKE_ID,
+                                     status=rd_instance.ServiceStatuses.NEW)
+        self.appStatus = FakeAppStatus(self.FAKE_ID,
+                                       rd_instance.ServiceStatuses.NEW)
+
+    def tearDown(self):
+
+        super(VerticaAppStatusTest, self).tearDown()
+        InstanceServiceStatus.find_by(instance_id=self.FAKE_ID).delete()
+
+    def test_get_actual_db_status(self):
+        self.verticaAppStatus = VerticaAppStatus()
+        with patch.object(vertica_system, 'shell_execute',
+                          MagicMock(return_value=['db_srvr', None])):
+            status = self.verticaAppStatus._get_actual_db_status()
+        self.assertEqual(rd_instance.ServiceStatuses.RUNNING, status)
+
+    def test_get_actual_db_status_shutdown(self):
+        self.verticaAppStatus = VerticaAppStatus()
+        with patch.object(vertica_system, 'shell_execute',
+                          MagicMock(side_effect=[['', None],
+                                                 ['db_srvr', None]])):
+            status = self.verticaAppStatus._get_actual_db_status()
+        self.assertEqual(rd_instance.ServiceStatuses.SHUTDOWN, status)
+
+    def test_get_actual_db_status_error_crashed(self):
+        self.verticaAppStatus = VerticaAppStatus()
+        with patch.object(vertica_system, 'shell_execute',
+                          MagicMock(side_effect=ProcessExecutionError('problem'
+                                                                      ))):
+            status = self.verticaAppStatus._get_actual_db_status()
+        self.assertEqual(rd_instance.ServiceStatuses.CRASHED, status)
+
+    def test_get_actual_db_status_error_unknown(self):
+        self.verticaAppStatus = VerticaAppStatus()
+        with patch.object(vertica_system, 'shell_execute',
+                          MagicMock(return_value=['', None])):
+            status = self.verticaAppStatus._get_actual_db_status()
+        self.assertEqual(rd_instance.ServiceStatuses.UNKNOWN, status)
+
+
+class VerticaAppTest(testtools.TestCase):
+
+    def setUp(self):
+        super(VerticaAppTest, self).setUp()
+        self.FAKE_ID = 1000
+        self.appStatus = FakeAppStatus(self.FAKE_ID,
+                                       rd_instance.ServiceStatuses.NEW)
+        self.app = VerticaApp(self.appStatus)
+        self.setread = VolumeDevice.set_readahead_size
+        vertica_system.shell_execute = MagicMock(return_value=('', ''))
+
+        VolumeDevice.set_readahead_size = Mock()
+        self.test_config = ConfigParser.ConfigParser()
+        self.test_config.add_section('credentials')
+        self.test_config.set('credentials',
+                             'dbadmin_password', 'some_password')
+
+    def tearDown(self):
+        super(VerticaAppTest, self).tearDown()
+        self.app = None
+        VolumeDevice.set_readahead_size = self.setread
+
+    def test_install_if_needed_installed(self):
+        with patch.object(pkg.Package, 'pkg_is_installed', return_value=True):
+            with patch.object(pkg.Package, 'pkg_install', return_value=None):
+                self.app.install_if_needed('vertica')
+                pkg.Package.pkg_is_installed.assert_any_call('vertica')
+                self.assertEqual(pkg.Package.pkg_install.call_count, 0)
+
+    def test_install_if_needed_not_installed(self):
+        with patch.object(pkg.Package, 'pkg_is_installed', return_value=False):
+            with patch.object(pkg.Package, 'pkg_install', return_value=None):
+                self.app.install_if_needed('vertica')
+                pkg.Package.pkg_is_installed.assert_any_call('vertica')
+                self.assertEqual(pkg.Package.pkg_install.call_count, 1)
+
+    def test_prepare_for_install_vertica(self):
+        self.app.prepare_for_install_vertica()
+        arguments = vertica_system.shell_execute.call_args_list[0]
+        self.assertEqual(VolumeDevice.set_readahead_size.call_count, 1)
+        expected_command = (
+            "VERT_DBA_USR=dbadmin VERT_DBA_HOME=/home/dbadmin "
+            "VERT_DBA_GRP=verticadba /opt/vertica/oss/python/bin/python"
+            " -m vertica.local_coerce")
+        arguments.assert_called_with(expected_command)
+
+    def test_install_vertica(self):
+        with patch.object(self.app, 'write_config',
+                          return_value=None):
+            self.app.install_vertica(members='10.0.0.2')
+        arguments = vertica_system.shell_execute.call_args_list[0]
+        expected_command = (
+            vertica_system.INSTALL_VERTICA % ('10.0.0.2', '/var/lib/vertica'))
+        arguments.assert_called_with(expected_command)
+
+    def test_create_db(self):
+        with patch.object(self.app, 'read_config',
+                          return_value=self.test_config):
+            self.app.create_db(members='10.0.0.2')
+        arguments = vertica_system.shell_execute.call_args_list[0]
+        expected_command = (vertica_system.CREATE_DB % ('10.0.0.2', 'db_srvr',
+                                                        '/var/lib/vertica',
+                                                        '/var/lib/vertica',
+                                                        'some_password'))
+        arguments.assert_called_with(expected_command, 'dbadmin')
+
+    def test_vertica_write_config(self):
+        temp_file_handle = tempfile.NamedTemporaryFile(delete=False)
+        mock_mkstemp = MagicMock(return_value=(temp_file_handle))
+        mock_unlink = Mock(return_value=0)
+        self.app.write_config(config=self.test_config,
+                              temp_function=mock_mkstemp,
+                              unlink_function=mock_unlink)
+
+        arguments = vertica_system.shell_execute.call_args_list[0]
+        expected_command = (
+            ("install -o root -g root -m 644 %(source)s %(target)s"
+             ) % {'source': temp_file_handle.name,
+                  'target': vertica_system.VERTICA_CONF})
+        arguments.assert_called_with(expected_command)
+        mock_mkstemp.assert_called_once()
+
+        configuration_data = ConfigParser.ConfigParser()
+        configuration_data.read(temp_file_handle.name)
+        self.assertEqual(
+            self.test_config.get('credentials', 'dbadmin_password'),
+            configuration_data.get('credentials', 'dbadmin_password'))
+        self.assertEqual(mock_unlink.call_count, 1)
+        # delete the temporary_config_file
+        os.unlink(temp_file_handle.name)
+
+    def test_vertica_error_in_write_config_verify_unlink(self):
+        mock_unlink = Mock(return_value=0)
+        temp_file_handle = tempfile.NamedTemporaryFile(delete=False)
+        mock_mkstemp = MagicMock(return_value=temp_file_handle)
+
+        with patch.object(vertica_system, 'shell_execute',
+                          side_effect=ProcessExecutionError('some exception')):
+            self.assertRaises(ProcessExecutionError,
+                              self.app.write_config,
+                              config=self.test_config,
+                              temp_function=mock_mkstemp,
+                              unlink_function=mock_unlink)
+
+        self.assertEqual(mock_unlink.call_count, 1)
+
+        # delete the temporary_config_file
+        os.unlink(temp_file_handle.name)
+
+    def test_restart(self):
+        mock_status = MagicMock()
+        app = VerticaApp(mock_status)
+        mock_status.begin_restart = MagicMock(return_value=None)
+        with patch.object(VerticaApp, 'stop_db', return_value=None):
+            with patch.object(VerticaApp, 'start_db', return_value=None):
+                mock_status.end_install_or_restart = MagicMock(
+                    return_value=None)
+                app.restart()
+                mock_status.begin_restart.assert_any_call()
+                VerticaApp.stop_db.assert_any_call()
+                VerticaApp.start_db.assert_any_call()
+                mock_status.end_install_or_restart.assert_any_call()
+
+    def test_start_db(self):
+        mock_status = MagicMock()
+        app = VerticaApp(mock_status)
+        with patch.object(app, '_enable_db_on_boot', return_value=None):
+            with patch.object(app, 'read_config',
+                              return_value=self.test_config):
+                mock_status.wait_for_real_status_to_change_to = MagicMock(
+                    return_value=True)
+                mock_status.end_install_or_restart = MagicMock(
+                    return_value=None)
+
+                app.start_db()
+
+                arguments = vertica_system.shell_execute.call_args_list[0]
+                expected_cmd = (vertica_system.START_DB % ('db_srvr',
+                                                           'some_password'))
+                self.assertTrue(
+                    mock_status.wait_for_real_status_to_change_to.called)
+                arguments.assert_called_with(expected_cmd, 'dbadmin')
+
+    def test_start_db_failure(self):
+        mock_status = MagicMock()
+        app = VerticaApp(mock_status)
+        with patch.object(app, '_enable_db_on_boot', return_value=None):
+            with patch.object(app, 'read_config',
+                              return_value=self.test_config):
+                mock_status.wait_for_real_status_to_change_to = MagicMock(
+                    return_value=None)
+                mock_status.end_install_or_restart = MagicMock(
+                    return_value=None)
+                self.assertRaises(RuntimeError, app.start_db)
+
+    def test_stop_db(self):
+        mock_status = MagicMock()
+        app = VerticaApp(mock_status)
+        with patch.object(app, '_disable_db_on_boot', return_value=None):
+            with patch.object(app, 'read_config',
+                              return_value=self.test_config):
+                mock_status.wait_for_real_status_to_change_to = MagicMock(
+                    return_value=True)
+                mock_status.end_install_or_restart = MagicMock(
+                    return_value=None)
+
+                app.stop_db()
+
+                arguments = vertica_system.shell_execute.call_args_list[0]
+                expected_command = (vertica_system.STOP_DB % ('db_srvr',
+                                                              'some_password'))
+                self.assertTrue(
+                    mock_status.wait_for_real_status_to_change_to.called)
+                arguments.assert_called_with(expected_command, 'dbadmin')
+
+    def test_stop_db_failure(self):
+        mock_status = MagicMock()
+        app = VerticaApp(mock_status)
+        with patch.object(app, '_disable_db_on_boot', return_value=None):
+            with patch.object(app, 'read_config',
+                              return_value=self.test_config):
+                mock_status.wait_for_real_status_to_change_to = MagicMock(
+                    return_value=None)
+                mock_status.end_install_or_restart = MagicMock(
+                    return_value=None)
+                self.assertRaises(RuntimeError, app.stop_db)

trove/tests/unittests/guestagent/test_vertica_manager.py

@@ -0,0 +1,146 @@
+#Copyright [2015] Hewlett-Packard Development Company, L.P.
+#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 testtools
+from mock import MagicMock
+from trove.common.context import TroveContext
+from trove.guestagent import volume
+from trove.guestagent.datastore.experimental.vertica.manager import Manager
+from trove.guestagent.datastore.experimental.vertica.service import VerticaApp
+from trove.guestagent.volume import VolumeDevice
+
+
+class GuestAgentManagerTest(testtools.TestCase):
+    def setUp(self):
+        super(GuestAgentManagerTest, self).setUp()
+        self.context = TroveContext()
+        self.manager = Manager()
+        self.origin_format = volume.VolumeDevice.format
+        self.origin_migrate_data = volume.VolumeDevice.migrate_data
+        self.origin_mount = volume.VolumeDevice.mount
+        self.origin_unmount = volume.VolumeDevice.unmount
+        self.origin_mount_points = volume.VolumeDevice.mount_points
+        self.origin_set_read = volume.VolumeDevice.set_readahead_size
+        self.origin_install_vertica = VerticaApp.install_vertica
+        self.origin_create_db = VerticaApp.create_db
+        self.origin_stop_db = VerticaApp.stop_db
+        self.origin_start_db = VerticaApp.start_db
+        self.origin_restart = VerticaApp.restart
+        self.origin_install_if = VerticaApp.install_if_needed
+        self.origin_complete_install = VerticaApp.complete_install_or_restart
+
+    def tearDown(self):
+        super(GuestAgentManagerTest, self).tearDown()
+        volume.VolumeDevice.format = self.origin_format
+        volume.VolumeDevice.migrate_data = self.origin_migrate_data
+        volume.VolumeDevice.mount = self.origin_mount
+        volume.VolumeDevice.unmount = self.origin_unmount
+        volume.VolumeDevice.mount_points = self.origin_mount_points
+        volume.VolumeDevice.set_readahead_size = self.origin_set_read
+        VerticaApp.create_db = self.origin_create_db
+        VerticaApp.install_vertica = self.origin_install_vertica
+        VerticaApp.stop_db = self.origin_stop_db
+        VerticaApp.start_db = self.origin_start_db
+        VerticaApp.restart = self.origin_restart
+        VerticaApp.install_if_needed = self.origin_install_if
+        VerticaApp.complete_install_or_restart = self.origin_complete_install
+
+    def test_update_status(self):
+        mock_status = MagicMock()
+        self.manager.appStatus = mock_status
+        self.manager.update_status(self.context)
+        mock_status.update.assert_any_call()
+
+    def _prepare_dynamic(self, packages,
+                         config_content='MockContent', device_path='/dev/vdb',
+                         backup_id=None,
+                         overrides=None, is_mounted=False):
+        # covering all outcomes is starting to cause trouble here
+        expected_vol_count = 1 if device_path else 0
+        if not backup_id:
+            backup_info = {'id': backup_id,
+                           'location': 'fake-location',
+                           'type': 'InnoBackupEx',
+                           'checksum': 'fake-checksum',
+                           }
+
+        mock_status = MagicMock()
+        self.manager.appStatus = mock_status
+
+        mock_status.begin_install = MagicMock(return_value=None)
+        path_exists_function = MagicMock(return_value=True)
+        volume.VolumeDevice.format = MagicMock(return_value=None)
+        volume.VolumeDevice.migrate_data = MagicMock(return_value=None)
+        volume.VolumeDevice.mount = MagicMock(return_value=None)
+        mount_points = []
+        if is_mounted:
+            mount_points = ['/mnt']
+        VolumeDevice.mount_points = MagicMock(return_value=mount_points)
+        VolumeDevice.unmount = MagicMock(return_value=None)
+
+        VerticaApp.install_if_needed = MagicMock(return_value=None)
+        VerticaApp.install_vertica = MagicMock(return_value=None)
+        VerticaApp.create_db = MagicMock(return_value=None)
+        VerticaApp.prepare_for_install_vertica = MagicMock(return_value=None)
+        VerticaApp.complete_install_or_restart = MagicMock(return_value=None)
+        # invocation
+        self.manager.prepare(context=self.context, packages=packages,
+                             config_contents=config_content,
+                             databases=None,
+                             memory_mb='2048', users=None,
+                             device_path=device_path,
+                             mount_point="/var/lib/vertica",
+                             backup_info=backup_info,
+                             overrides=None,
+                             cluster_config=None,
+                             path_exists_function=path_exists_function)
+
+        self.assertEqual(VolumeDevice.format.call_count, expected_vol_count)
+        self.assertEqual(VolumeDevice.migrate_data.call_count,
+                         expected_vol_count)
+        self.assertEqual(VolumeDevice.mount_points.call_count,
+                         expected_vol_count)
+        if is_mounted:
+            self.assertEqual(VolumeDevice.unmount.call_count, 1)
+        else:
+            self.assertEqual(VolumeDevice.unmount.call_count, 0)
+
+        VerticaApp.install_if_needed.assert_any_call(packages)
+        VerticaApp.prepare_for_install_vertica.assert_any_call()
+        VerticaApp.install_vertica.assert_any_call()
+        VerticaApp.create_db.assert_any_call()
+        VerticaApp.complete_install_or_restart.assert_any_call()
+
+    def test_prepare_pkg(self):
+        self._prepare_dynamic(['vertica'])
+
+    def test_prepare_no_pkg(self):
+        self._prepare_dynamic([])
+
+    def test_restart(self):
+        mock_status = MagicMock()
+        self.manager.appStatus = mock_status
+        VerticaApp.restart = MagicMock(return_value=None)
+        #invocation
+        self.manager.restart(self.context)
+        #verification/assertion
+        VerticaApp.restart.assert_any_call()
+
+    def test_stop_db(self):
+        mock_status = MagicMock()
+        self.manager.appStatus = mock_status
+        VerticaApp.stop_db = MagicMock(return_value=None)
+        #invocation
+        self.manager.stop_db(self.context)
+        #verification/assertion
+        VerticaApp.stop_db.assert_any_call(do_not_start_on_reboot=False)

trove/tests/unittests/guestagent/test_volume.py

@@ -145,6 +145,19 @@ class VolumeDeviceTest(testtools.TestCase):
         self.assertEqual(COUNT, fake_spawn.expect.call_count)
         os.path.exists = origin_
 
+    def test_set_readahead_size(self):
+        origin_check_device_exists = self.volumeDevice._check_device_exists
+        self.volumeDevice._check_device_exists = MagicMock()
+        mock_execute = MagicMock(return_value=None)
+        readahead_size = 2048
+        self.volumeDevice.set_readahead_size(readahead_size,
+                                             execute_function=mock_execute)
+        blockdev = mock_execute.call_args_list[0]
+
+        blockdev.assert_called_with("sudo", "blockdev", "--setra",
+                                    readahead_size, "/dev/vdb")
+        self.volumeDevice._check_device_exists = origin_check_device_exists
+
 
 class VolumeMountPointTest(testtools.TestCase):
     def setUp(self):