basic API service: Create the base DAL into the DB

Create the base data abstraction layer into the DB in basic
API service.I create a database table services as a example.
This table is used to save the running status of deployed
services.
The command smaug-manage is used to manage the database.
we can use this command 'smaug-manage db sync' to sync
the smaug database up to the most recent version.
smaug-manage version list: exposing the smaug codebase
version.
smaug-manage config list: exposing the configuration.
smaug-manage service list: showing a list of all smaug
services status.

Change-Id: I5d64c5d38780449e1fb005acf04f69d482ac59cc
Closes-Bug: #1525794

MANIFEST.in

@@ -1,5 +1,6 @@
 include AUTHORS
 include ChangeLog
+recursive-include smaug *.cfg
 exclude .gitignore
 exclude .gitreview
 

devstack/plugin.sh

@@ -77,6 +77,14 @@ if [[ "$Q_ENABLE_SMAUG" == "True" ]]; then
     elif [[ "$1" == "stack" && "$2" == "extra" ]]; then
         echo_summary "Initializing Smaug Service"
         SMAUG_BIN_DIR=$(get_python_exec_prefix)
+
+        if is_service_enabled $DATABASE_BACKENDS; then
+            # (re)create smaug database
+            recreate_database smaug utf8
+
+            # Migrate smaug database
+            $SMAUG_BIN_DIR/smaug-manage db sync
+        fi
         if is_service_enabled smaug-api; then
             run_process smaug-api "$SMAUG_BIN_DIR/smaug-api --config-file $SMAUG_API_CONF"
         fi

requirements.txt

@@ -26,5 +26,6 @@ Routes!=2.0,>=1.12.3;python_version!='2.7'
 six>=1.9.0
 SQLAlchemy<1.1.0,>=0.9.9
 sqlalchemy-migrate>=0.9.6
+stevedore>=1.5.0 # Apache-2.0
 WebOb>=1.2.3
 oslo.i18n>=1.5.0 # Apache-2.0

setup.cfg

@@ -32,6 +32,10 @@ data_files =
 [entry_points]
 console_scripts =
     smaug-api = smaug.cmd.api:main
+    smaug-manage = smaug.cmd.manage:main
+
+smaug.database.migration_backend =
+    sqlalchemy = oslo_db.sqlalchemy.migration
 
 [build_sphinx]
 source-dir = doc/source

smaug/api/openstack/wsgi.py

@@ -620,7 +620,7 @@ class Resource(wsgi.Application):
         action_args.update(contents)
 
         project_id = action_args.pop("project_id", None)
-        context = request.environ.get('smuag.context')
+        context = request.environ.get('smaug.context')
         if (context and project_id and (project_id != context.project_id)):
             msg = _("Malformed request url")
             return Fault(webob.exc.HTTPBadRequest(explanation=msg))

smaug/cmd/manage.py

@@ -0,0 +1,242 @@
+#!/usr/bin/env python
+#    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.
+
+"""
+  CLI interface for smaug management.
+"""
+
+from __future__ import print_function
+
+
+import os
+import sys
+
+from oslo_config import cfg
+from oslo_db.sqlalchemy import migration
+from oslo_log import log as logging
+
+from smaug import i18n
+i18n.enable_lazy()
+
+# Need to register global_opts
+from smaug.common import config  # noqa
+from smaug import context
+from smaug import db
+from smaug.db import migration as db_migration
+from smaug.db.sqlalchemy import api as db_api
+from smaug.i18n import _
+from smaug import utils
+from smaug import version
+
+
+CONF = cfg.CONF
+
+
+# Decorators for actions
+def args(*args, **kwargs):
+    def _decorator(func):
+        func.__dict__.setdefault('args', []).insert(0, (args, kwargs))
+        return func
+    return _decorator
+
+
+class DbCommands(object):
+    """Class for managing the database."""
+
+    def __init__(self):
+        pass
+
+    @args('version', nargs='?', default=None,
+          help='Database version')
+    def sync(self, version=None):
+        """Sync the database up to the most recent version."""
+        return db_migration.db_sync(version)
+
+    def version(self):
+        """Print the current database version."""
+        print(db_migration.MIGRATE_REPO_PATH)
+        print(migration.db_version(db_api.get_engine(),
+                                   db_migration.MIGRATE_REPO_PATH,
+                                   db_migration.INIT_VERSION))
+
+
+class VersionCommands(object):
+    """Class for exposing the codebase version."""
+
+    def __init__(self):
+        pass
+
+    def list(self):
+        print(version.version_string())
+
+    def __call__(self):
+        self.list()
+
+
+class ConfigCommands(object):
+    """Class for exposing the flags defined by flag_file(s)."""
+
+    def __init__(self):
+        pass
+
+    @args('param', nargs='?', default=None,
+          help='Configuration parameter to display (default: %(default)s)')
+    def list(self, param=None):
+        """List parameters configured for smaug.
+
+        Lists all parameters configured for smaug unless an optional argument
+        is specified.  If the parameter is specified we only print the
+        requested parameter.  If the parameter is not found an appropriate
+        error is produced by .get*().
+        """
+        param = param and param.strip()
+        if param:
+            print('%s = %s' % (param, CONF.get(param)))
+        else:
+            for key, value in CONF.items():
+                print('%s = %s' % (key, value))
+
+
+class ServiceCommands(object):
+    """Methods for managing services."""
+    def list(self):
+        """Show a list of all smaug services."""
+
+        ctxt = context.get_admin_context()
+        services = db.service_get_all(ctxt)
+        print_format = "%-16s %-36s %-16s %-10s %-5s %-10s"
+        print(print_format % (_('Binary'),
+                              _('Host'),
+                              _('Zone'),
+                              _('Status'),
+                              _('State'),
+                              _('Updated At')))
+        for svc in services:
+            alive = utils.service_is_up(svc)
+            art = ":-)" if alive else "XXX"
+            status = 'enabled'
+            if svc['disabled']:
+                status = 'disabled'
+            print(print_format % (svc['binary'], svc['host'].partition('.')[0],
+                                  svc['availability_zone'], status, art,
+                                  svc['updated_at']))
+
+
+CATEGORIES = {
+    'config': ConfigCommands,
+    'db': DbCommands,
+    'service': ServiceCommands,
+    'version': VersionCommands,
+}
+
+
+def methods_of(obj):
+    """Return non-private methods from an object.
+
+    Get all callable methods of an object that don't start with underscore
+    :return: a list of tuples of the form (method_name, method)
+    """
+    result = []
+    for i in dir(obj):
+        if callable(getattr(obj, i)) and not i.startswith('_'):
+            result.append((i, getattr(obj, i)))
+    return result
+
+
+def add_command_parsers(subparsers):
+    for category in CATEGORIES:
+        command_object = CATEGORIES[category]()
+
+        parser = subparsers.add_parser(category)
+        parser.set_defaults(command_object=command_object)
+
+        category_subparsers = parser.add_subparsers(dest='action')
+
+        for (action, action_fn) in methods_of(command_object):
+            parser = category_subparsers.add_parser(action)
+
+            action_kwargs = []
+            for args, kwargs in getattr(action_fn, 'args', []):
+                parser.add_argument(*args, **kwargs)
+
+            parser.set_defaults(action_fn=action_fn)
+            parser.set_defaults(action_kwargs=action_kwargs)
+
+
+category_opt = cfg.SubCommandOpt('category',
+                                 title='Command categories',
+                                 handler=add_command_parsers)
+
+
+def get_arg_string(args):
+    arg = None
+    if args[0] == '-':
+        # (Note)zhiteng: args starts with FLAGS.oparser.prefix_chars
+        # is optional args. Notice that cfg module takes care of
+        # actual ArgParser so prefix_chars is always '-'.
+        if args[1] == '-':
+            # This is long optional arg
+            arg = args[2:]
+        else:
+            arg = args[1:]
+    else:
+        arg = args
+
+    return arg
+
+
+def fetch_func_args(func):
+    fn_args = []
+    for args, kwargs in getattr(func, 'args', []):
+        arg = get_arg_string(args[0])
+        fn_args.append(getattr(CONF.category, arg))
+
+    return fn_args
+
+
+def main():
+    """Parse options and call the appropriate class/method."""
+    CONF.register_cli_opt(category_opt)
+    script_name = sys.argv[0]
+    if len(sys.argv) < 2:
+        print(_("\nOpenStack Smaug version: %(version)s\n") %
+              {'version': version.version_string()})
+        print(script_name + " category action [<args>]")
+        print(_("Available categories:"))
+        for category in CATEGORIES:
+            print(_("\t%s") % category)
+        sys.exit(2)
+
+    try:
+        CONF(sys.argv[1:], project='smaug',
+             version=version.version_string())
+        logging.setup(CONF, "smaug")
+    except cfg.ConfigDirNotFoundError as details:
+        print(_("Invalid directory: %s") % details)
+        sys.exit(2)
+    except cfg.ConfigFilesNotFoundError:
+        cfgfile = CONF.config_file[-1] if CONF.config_file else None
+        if cfgfile and not os.access(cfgfile, os.R_OK):
+            st = os.stat(cfgfile)
+            print(_("Could not read %s. Re-running with sudo") % cfgfile)
+            try:
+                os.execvp('sudo', ['sudo', '-u', '#%s' % st.st_uid] + sys.argv)
+            except Exception:
+                print(_('sudo failed, continuing as if nothing happened'))
+
+        print(_('Please re-run smaug-manage as root.'))
+        sys.exit(2)
+
+    fn = CONF.category.action_fn
+    fn_args = fetch_func_args(fn)
+    fn(*fn_args)

smaug/common/config.py

@@ -31,7 +31,11 @@ logging.register_options(CONF)
 core_opts = [
     cfg.StrOpt('api_paste_config',
                default="api-paste.ini",
-               help='File name for the paste.deploy config for smaug-api')
+               help='File name for the paste.deploy config for smaug-api'),
+    cfg.StrOpt('state_path',
+               default='/var/lib/smaug',
+               deprecated_name='pybasedir',
+               help="Top-level directory for maintaining smaug's state"),
 ]
 
 debug_opts = [
@@ -41,6 +45,10 @@ CONF.register_cli_opts(core_opts)
 CONF.register_cli_opts(debug_opts)
 
 global_opts = [
+    cfg.IntOpt('service_down_time',
+               default=60,
+               help='Maximum time since last check-in for a service to be '
+                    'considered up'),
     cfg.StrOpt('scheduler_topic',
                default='Smaug-scheduler',
                help='The topic that scheduler nodes listen on'),

smaug/db/__init__.py

@@ -0,0 +1,16 @@
+#    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.
+"""
+DB abstraction for smaug
+"""
+
+from smaug.db.api import *  # noqa

smaug/db/api.py

@@ -0,0 +1,115 @@
+#    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.
+
+"""Defines interface for DB access.
+
+Functions in this module are imported into the smaug.db namespace. Call these
+functions from smaug.db namespace, not the smaug.db.api namespace.
+
+All functions in this module return objects that implement a dictionary-like
+interface. Currently, many of these objects are sqlalchemy objects that
+implement a dictionary interface. However, a future goal is to have all of
+these objects be simple dictionaries.
+
+
+**Related Flags**
+
+:connection:  string specifying the sqlalchemy connection to use, like:
+              `sqlite:///var/lib/smaug/smaug.sqlite`.
+
+:enable_new_services:  when adding a new service to the database, is it in the
+                       pool of available hardware (Default: True)
+
+"""
+
+from oslo_config import cfg
+from oslo_db import concurrency as db_concurrency
+from oslo_db import options as db_options
+
+
+db_opts = [
+    cfg.BoolOpt('enable_new_services',
+                default=True,
+                help='Services to be added to the available pool on create'),
+]
+
+
+CONF = cfg.CONF
+CONF.register_opts(db_opts)
+db_options.set_defaults(CONF)
+CONF.set_default('sqlite_db', 'smaug.sqlite', group='database')
+
+_BACKEND_MAPPING = {'sqlalchemy': 'smaug.db.sqlalchemy.api'}
+
+
+IMPL = db_concurrency.TpoolDbapiWrapper(CONF, _BACKEND_MAPPING)
+
+# The maximum value a signed INT type may have
+MAX_INT = 0x7FFFFFFF
+
+
+###################
+
+def dispose_engine():
+    """Force the engine to establish new connections."""
+
+    # FIXME(jdg): When using sqlite if we do the dispose
+    # we seem to lose our DB here.  Adding this check
+    # means we don't do the dispose, but we keep our sqlite DB
+    # This likely isn't the best way to handle this
+
+    if 'sqlite' not in IMPL.get_engine().name:
+        return IMPL.dispose_engine()
+    else:
+        return
+
+
+###################
+
+
+def service_destroy(context, service_id):
+    """Destroy the service or raise if it does not exist."""
+    return IMPL.service_destroy(context, service_id)
+
+
+def service_get(context, service_id):
+    """Get a service or raise if it does not exist."""
+    return IMPL.service_get(context, service_id)
+
+
+def service_get_by_host_and_topic(context, host, topic):
+    """Get a service by host it's on and topic it listens to."""
+    return IMPL.service_get_by_host_and_topic(context, host, topic)
+
+
+def service_get_all(context, disabled=None):
+    """Get all services."""
+    return IMPL.service_get_all(context, disabled)
+
+
+def service_get_all_by_topic(context, topic, disabled=None):
+    """Get all services for a given topic."""
+    return IMPL.service_get_all_by_topic(context, topic, disabled=disabled)
+
+
+def service_create(context, values):
+    """Create a service from the values dictionary."""
+    return IMPL.service_create(context, values)
+
+
+def service_update(context, service_id, values):
+    """Set the given properties on an service and update it.
+
+    Raises NotFound if service does not exist.
+
+    """
+    return IMPL.service_update(context, service_id, values)

smaug/db/base.py

@@ -0,0 +1,38 @@
+#    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.
+
+"""Base class for classes that need modular database access."""
+
+
+from oslo_config import cfg
+from oslo_utils import importutils
+
+
+db_driver_opt = cfg.StrOpt('db_driver',
+                           default='smaug.db',
+                           help='Driver to use for database access')
+
+CONF = cfg.CONF
+CONF.register_opt(db_driver_opt)
+
+
+class Base(object):
+    """DB driver is injected in the init method."""
+
+    def __init__(self, db_driver=None):
+        # NOTE(mriedem): Without this call, multiple inheritance involving
+        # the db Base class does not work correctly.
+        super(Base, self).__init__()
+        if not db_driver:
+            db_driver = CONF.db_driver
+        self.db = importutils.import_module(db_driver)  # pylint: disable=C0103
+        self.db.dispose_engine()

smaug/db/migration.py

@@ -0,0 +1,57 @@
+#    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.
+
+"""Database setup and migration commands."""
+
+import os
+import threading
+
+from oslo_config import cfg
+from oslo_db import options
+from stevedore import driver
+
+from smaug.db.sqlalchemy import api as db_api
+
+INIT_VERSION = 000
+
+_IMPL = None
+_LOCK = threading.Lock()
+
+options.set_defaults(cfg.CONF)
+
+MIGRATE_REPO_PATH = os.path.join(
+    os.path.abspath(os.path.dirname(__file__)),
+    'sqlalchemy',
+    'migrate_repo',
+)
+
+
+def get_backend():
+    global _IMPL
+    if _IMPL is None:
+        with _LOCK:
+            if _IMPL is None:
+                _IMPL = driver.DriverManager(
+                    "smaug.database.migration_backend",
+                    cfg.CONF.database.backend).driver
+    return _IMPL
+
+
+def db_sync(version=None, init_version=INIT_VERSION, engine=None):
+    """Migrate the database to `version` or the most recent version."""
+
+    if engine is None:
+        engine = db_api.get_engine()
+    return get_backend().db_sync(engine=engine,
+                                 abs_path=MIGRATE_REPO_PATH,
+                                 version=version,
+                                 init_version=init_version)

smaug/db/sqlalchemy/api.py

@@ -0,0 +1,308 @@
+#    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.
+
+"""Implementation of SQLAlchemy backend."""
+
+
+import functools
+import sys
+import threading
+import time
+
+from oslo_config import cfg
+from oslo_db import exception as db_exc
+from oslo_db import options
+from oslo_db.sqlalchemy import session as db_session
+from oslo_log import log as logging
+from oslo_utils import timeutils
+from sqlalchemy.sql.expression import literal_column
+from sqlalchemy.sql import func
+
+from smaug.db.sqlalchemy import models
+from smaug import exception
+from smaug.i18n import _, _LW
+
+
+CONF = cfg.CONF
+LOG = logging.getLogger(__name__)
+
+options.set_defaults(CONF, connection='sqlite:///$state_path/smaug.sqlite')
+
+_LOCK = threading.Lock()
+_FACADE = None
+
+
+def _create_facade_lazily():
+    global _LOCK
+    with _LOCK:
+        global _FACADE
+        if _FACADE is None:
+            _FACADE = db_session.EngineFacade(
+                CONF.database.connection,
+                **dict(CONF.database)
+            )
+
+        return _FACADE
+
+
+def get_engine():
+    facade = _create_facade_lazily()
+    return facade.get_engine()
+
+
+def get_session(**kwargs):
+    facade = _create_facade_lazily()
+    return facade.get_session(**kwargs)
+
+
+def dispose_engine():
+    get_engine().dispose()
+
+_DEFAULT_QUOTA_NAME = 'default'
+
+
+def get_backend():
+    """The backend is this module itself."""
+
+    return sys.modules[__name__]
+
+
+def is_admin_context(context):
+    """Indicates if the request context is an administrator."""
+    if not context:
+        LOG.warning(_LW('Use of empty request context is deprecated'),
+                    DeprecationWarning)
+        raise Exception('die')
+    return context.is_admin
+
+
+def is_user_context(context):
+    """Indicates if the request context is a normal user."""
+    if not context:
+        return False
+    if context.is_admin:
+        return False
+    if not context.user_id or not context.project_id:
+        return False
+    return True
+
+
+def authorize_project_context(context, project_id):
+    """Ensures a request has permission to access the given project."""
+    if is_user_context(context):
+        if not context.project_id:
+            raise exception.NotAuthorized()
+        elif context.project_id != project_id:
+            raise exception.NotAuthorized()
+
+
+def authorize_user_context(context, user_id):
+    """Ensures a request has permission to access the given user."""
+    if is_user_context(context):
+        if not context.user_id:
+            raise exception.NotAuthorized()
+        elif context.user_id != user_id:
+            raise exception.NotAuthorized()
+
+
+def require_admin_context(f):
+    """Decorator to require admin request context.
+
+    The first argument to the wrapped function must be the context.
+
+    """
+
+    def wrapper(*args, **kwargs):
+        if not is_admin_context(args[0]):
+            raise exception.AdminRequired()
+        return f(*args, **kwargs)
+    return wrapper
+
+
+def require_context(f):
+    """Decorator to require *any* user or admin context.
+
+    This does no authorization for user or project access matching, see
+    :py:func:`authorize_project_context` and
+    :py:func:`authorize_user_context`.
+
+    The first argument to the wrapped function must be the context.
+
+    """
+
+    def wrapper(*args, **kwargs):
+        if not is_admin_context(args[0]) and not is_user_context(args[0]):
+            raise exception.NotAuthorized()
+        return f(*args, **kwargs)
+    return wrapper
+
+
+def _retry_on_deadlock(f):
+    """Decorator to retry a DB API call if Deadlock was received."""
+    @functools.wraps(f)
+    def wrapped(*args, **kwargs):
+        while True:
+            try:
+                return f(*args, **kwargs)
+            except db_exc.DBDeadlock:
+                LOG.warning(_LW("Deadlock detected when running "
+                                "'%(func_name)s': Retrying..."),
+                            dict(func_name=f.__name__))
+                # Retry!
+                time.sleep(0.5)
+                continue
+    functools.update_wrapper(wrapped, f)
+    return wrapped
+
+
+def model_query(context, *args, **kwargs):
+    """Query helper that accounts for context's `read_deleted` field.
+
+    :param context: context to query under
+    :param session: if present, the session to use
+    :param read_deleted: if present, overrides context's read_deleted field.
+    :param project_only: if present and context is user-type, then restrict
+            query to match the context's project_id.
+    """
+    session = kwargs.get('session') or get_session()
+    read_deleted = kwargs.get('read_deleted') or context.read_deleted
+    project_only = kwargs.get('project_only')
+
+    query = session.query(*args)
+    if read_deleted == 'no':
+        query = query.filter_by(deleted=False)
+    elif read_deleted == 'yes':
+        pass  # omit the filter to include deleted and active
+    elif read_deleted == 'only':
+        query = query.filter_by(deleted=True)
+    else:
+        raise Exception(
+            _("Unrecognized read_deleted value '%s'") % read_deleted)
+
+    if project_only and is_user_context(context):
+        query = query.filter_by(project_id=context.project_id)
+
+    return query
+
+
+@require_admin_context
+def service_destroy(context, service_id):
+    session = get_session()
+    with session.begin():
+        service_ref = _service_get(context, service_id, session=session)
+        service_ref.delete(session=session)
+
+
+@require_admin_context
+def _service_get(context, service_id, session=None):
+    result = model_query(
+        context,
+        models.Service,
+        session=session).\
+        filter_by(id=service_id).\
+        first()
+    if not result:
+        raise exception.ServiceNotFound(service_id=service_id)
+
+    return result
+
+
+@require_admin_context
+def service_get(context, service_id):
+    return _service_get(context, service_id)
+
+
+@require_admin_context
+def service_get_all(context, disabled=None):
+    query = model_query(context, models.Service)
+
+    if disabled is not None:
+        query = query.filter_by(disabled=disabled)
+
+    return query.all()
+
+
+@require_admin_context
+def service_get_all_by_topic(context, topic, disabled=None):
+    query = model_query(
+        context, models.Service, read_deleted="no").\
+        filter_by(topic=topic)
+
+    if disabled is not None:
+        query = query.filter_by(disabled=disabled)
+
+    return query.all()
+
+
+@require_admin_context
+def service_get_by_host_and_topic(context, host, topic):
+    result = model_query(
+        context, models.Service, read_deleted="no").\
+        filter_by(disabled=False).\
+        filter_by(host=host).\
+        filter_by(topic=topic).\
+        first()
+    if not result:
+        raise exception.ServiceNotFound(service_id=None)
+    return result
+
+
+@require_admin_context
+def _service_get_all_topic_subquery(context, session, topic, subq, label):
+    sort_value = getattr(subq.c, label)
+    return model_query(context, models.Service,
+                       func.coalesce(sort_value, 0),
+                       session=session, read_deleted="no").\
+        filter_by(topic=topic).\
+        filter_by(disabled=False).\
+        outerjoin((subq, models.Service.host == subq.c.host)).\
+        order_by(sort_value).\
+        all()
+
+
+@require_admin_context
+def service_get_by_args(context, host, binary):
+    results = model_query(context, models.Service).\
+        filter_by(host=host).\
+        filter_by(binary=binary).\
+        all()
+
+    for result in results:
+        if host == result['host']:
+            return result
+
+    raise exception.HostBinaryNotFound(host=host, binary=binary)
+
+
+@require_admin_context
+def service_create(context, values):
+    service_ref = models.Service()
+    service_ref.update(values)
+    if not CONF.enable_new_services:
+        service_ref.disabled = True
+
+    session = get_session()
+    with session.begin():
+        service_ref.save(session)
+        return service_ref
+
+
+@require_admin_context
+def service_update(context, service_id, values):
+    session = get_session()
+    with session.begin():
+        service_ref = _service_get(context, service_id, session=session)
+        if ('disabled' in values):
+            service_ref['modified_at'] = timeutils.utcnow()
+            service_ref['updated_at'] = literal_column('updated_at')
+        service_ref.update(values)
+        return service_ref

smaug/db/sqlalchemy/migrate_repo/README

@@ -0,0 +1,4 @@
+This is a database migration repository.
+
+More information at
+http://code.google.com/p/sqlalchemy-migrate/

smaug/db/sqlalchemy/migrate_repo/manage.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+#    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 smaug.db.sqlalchemy import migrate_repo
+
+from migrate.versioning.shell import main
+
+
+if __name__ == '__main__':
+    main(debug='False',
+         repository=os.path.abspath(os.path.dirname(migrate_repo.__file__)))

smaug/db/sqlalchemy/migrate_repo/migrate.cfg

@@ -0,0 +1,20 @@
+[db_settings]
+# Used to identify which repository this database is versioned under.
+# You can use the name of your project.
+repository_id=smaug
+
+# The name of the database table used to track the schema version.
+# This name shouldn't already be used by your project.
+# If this is changed once a database is under version control, you'll need to 
+# change the table name in each database too. 
+version_table=migrate_version
+
+# When committing a change script, Migrate will attempt to generate the 
+# sql for all supported databases; normally, if one of them fails - probably
+# because you don't have that database installed - it is ignored and the 
+# commit continues, perhaps ending successfully. 
+# Databases in this list MUST compile successfully during a commit, or the 
+# entire commit will fail. List the databases your application will actually 
+# be using to ensure your updates to that database work properly.
+# This must be a list; example: ['postgres','sqlite']
+required_dbs=[]

smaug/db/sqlalchemy/migrate_repo/versions/001_smaug_init.py

@@ -0,0 +1,72 @@
+#    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 sqlalchemy import Boolean, Column, DateTime
+from sqlalchemy import Integer, MetaData, String, Table
+
+
+def define_tables(meta):
+
+    services = Table(
+        'services', meta,
+        Column('created_at', DateTime),
+        Column('updated_at', DateTime),
+        Column('deleted_at', DateTime),
+        Column('deleted', Boolean),
+        Column('id', Integer, primary_key=True, nullable=False),
+        Column('host', String(length=255)),
+        Column('binary', String(length=255)),
+        Column('topic', String(length=255)),
+        Column('report_count', Integer, nullable=False),
+        Column('disabled', Boolean),
+        Column('availability_zone', String(length=255)),
+        Column('disabled_reason', String(length=255)),
+        Column('modified_at', DateTime),
+        mysql_engine='InnoDB'
+    )
+
+    return [services]
+
+
+def upgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+
+    # create all tables
+    # Take care on create order for those with FK dependencies
+    tables = define_tables(meta)
+
+    for table in tables:
+        table.create()
+
+    if migrate_engine.name == "mysql":
+        tables = ["migrate_version",
+                  "services"]
+
+        migrate_engine.execute("SET foreign_key_checks = 0")
+        for table in tables:
+            migrate_engine.execute(
+                "ALTER TABLE %s CONVERT TO CHARACTER SET utf8" % table)
+        migrate_engine.execute("SET foreign_key_checks = 1")
+        migrate_engine.execute(
+            "ALTER DATABASE %s DEFAULT CHARACTER SET utf8" %
+            migrate_engine.url.database)
+        migrate_engine.execute("ALTER TABLE %s Engine=InnoDB" % table)
+
+
+def downgrade(migrate_engine):
+    meta = MetaData()
+    meta.bind = migrate_engine
+    tables = define_tables(meta)
+    tables.reverse()
+    for table in tables:
+        table.drop()

smaug/db/sqlalchemy/models.py

@@ -0,0 +1,75 @@
+#    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.
+"""
+SQLAlchemy models for smaug data.
+"""
+
+from oslo_config import cfg
+from oslo_db.sqlalchemy import models
+from oslo_utils import timeutils
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy import DateTime, Boolean
+
+
+CONF = cfg.CONF
+BASE = declarative_base()
+
+
+class SmaugBase(models.TimestampMixin,
+                models.ModelBase):
+    """Base class for Smaug Models."""
+
+    __table_args__ = {'mysql_engine': 'InnoDB'}
+
+    deleted_at = Column(DateTime)
+    deleted = Column(Boolean, default=False)
+    metadata = None
+
+    def delete(self, session):
+        """Delete this object."""
+        self.deleted = True
+        self.deleted_at = timeutils.utcnow()
+        self.save(session=session)
+
+
+class Service(BASE, SmaugBase):
+    """Represents a running service on a host."""
+
+    __tablename__ = 'services'
+    id = Column(Integer, primary_key=True)
+    host = Column(String(255))  # , ForeignKey('hosts.id'))
+    binary = Column(String(255))
+    topic = Column(String(255))
+    report_count = Column(Integer, nullable=False, default=0)
+    disabled = Column(Boolean, default=False)
+    availability_zone = Column(String(255), default='smaug')
+    disabled_reason = Column(String(255))
+    # adding column modified_at to contain timestamp
+    # for manual enable/disable of smaug services
+    # updated_at column will now contain timestamps for
+    # periodic updates
+    modified_at = Column(DateTime)
+
+
+def register_models():
+    """Register Models and create metadata.
+
+    Called from smaug.db.sqlalchemy.__init__ as part of loading the driver,
+    it will never need to be called explicitly elsewhere unless the
+    connection is lost and needs to be reestablished.
+    """
+    from sqlalchemy import create_engine
+    models = (Service,)
+    engine = create_engine(CONF.database.connection, echo=False)
+    for model in models:
+        model.metadata.create_all(engine)

smaug/exception.py

@@ -169,3 +169,7 @@ class InvalidContentType(Invalid):
 
 class PasteAppNotFound(NotFound):
     message = _("Could not load paste app '%(name)s' from %(path)s")
+
+
+class ServiceNotFound(NotFound):
+    message = _("Service %(service_id)s could not be found.")

smaug/tests/base.py

@@ -1,7 +1,3 @@
-# -*- coding: utf-8 -*-
-
-# Copyright 2010-2011 OpenStack Foundation
-#
 # 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
@@ -13,24 +9,70 @@
 # 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 logging
 import os
+import shutil
 
+import fixtures
 from oslo_config import cfg
 from oslo_log import log
 from oslotest import base
 
 from smaug.common import config  # noqa Need to register global_opts
+from smaug.db import migration
+from smaug.db.sqlalchemy import api as sqla_api
 from smaug.tests.unit import conf_fixture
 
 test_opts = [
-
-]
+    cfg.StrOpt('sqlite_clean_db',
+               default='clean.sqlite',
+               help='File name of clean sqlite db'), ]
 
 CONF = cfg.CONF
 CONF.register_opts(test_opts)
 
 LOG = log.getLogger(__name__)
 
+_DB_CACHE = None
+
+
+class Database(fixtures.Fixture):
+
+    def __init__(self, db_api, db_migrate, sql_connection,
+                 sqlite_db, sqlite_clean_db):
+        self.sql_connection = sql_connection
+        self.sqlite_db = sqlite_db
+        self.sqlite_clean_db = sqlite_clean_db
+
+        # Suppress logging for test runs
+        migrate_logger = logging.getLogger('migrate')
+        migrate_logger.setLevel(logging.WARNING)
+
+        self.engine = db_api.get_engine()
+        self.engine.dispose()
+        conn = self.engine.connect()
+        db_migrate.db_sync()
+        if sql_connection == "sqlite://":
+            conn = self.engine.connect()
+            self._DB = "".join(line for line in conn.connection.iterdump())
+            self.engine.dispose()
+        else:
+            cleandb = os.path.join(CONF.state_path, sqlite_clean_db)
+            testdb = os.path.join(CONF.state_path, sqlite_db)
+            shutil.copyfile(testdb, cleandb)
+
+    def setUp(self):
+        super(Database, self).setUp()
+
+        if self.sql_connection == "sqlite://":
+            conn = self.engine.connect()
+            conn.connection.executescript(self._DB)
+            self.addCleanup(self.engine.dispose)
+        else:
+            shutil.copyfile(
+                os.path.join(CONF.state_path, self.sqlite_clean_db),
+                os.path.join(CONF.state_path, self.sqlite_db))
+
 
 class TestCase(base.BaseTestCase):
 
@@ -43,6 +85,17 @@ class TestCase(base.BaseTestCase):
         conf_fixture.set_defaults(CONF)
         CONF([], default_config_files=[])
 
+        CONF.set_default('connection', 'sqlite://', 'database')
+        CONF.set_default('sqlite_synchronous', False, 'database')
+
+        global _DB_CACHE
+        if not _DB_CACHE:
+            _DB_CACHE = Database(sqla_api, migration,
+                                 sql_connection=CONF.database.connection,
+                                 sqlite_db=CONF.database.sqlite_db,
+                                 sqlite_clean_db=CONF.sqlite_clean_db)
+        self.useFixture(_DB_CACHE)
+
         self.override_config('policy_file',
                              os.path.join(
                                  os.path.abspath(

smaug/tests/unit/conf_fixture.py

@@ -10,7 +10,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
+import os
 from oslo_config import cfg
 
 
@@ -20,7 +20,11 @@ CONF.import_opt('policy_file', 'smaug.policy', group='oslo_policy')
 
 
 def set_defaults(conf):
+    conf.set_default('connection', 'sqlite://', group='database')
+    conf.set_default('sqlite_synchronous', False, group='database')
     conf.set_default('policy_file', 'smaug.tests.unit/policy.json',
                      group='oslo_policy')
     conf.set_default('policy_dirs', [], group='oslo_policy')
     conf.set_default('auth_strategy', 'noauth')
+    conf.set_default('state_path', os.path.abspath(
+        os.path.join(os.path.dirname(__file__), '..', '..', '..')))

smaug/tests/unit/db/test_models.py

@@ -0,0 +1,98 @@
+#    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.
+
+"""Tests for Models Database."""
+
+from oslo_config import cfg
+
+from smaug import context
+from smaug import db
+from smaug import exception
+from smaug.tests import base
+
+
+CONF = cfg.CONF
+
+
+class ServicesDbTestCase(base.TestCase):
+    """Test cases for Services database table."""
+
+    def setUp(self):
+        super(ServicesDbTestCase, self).setUp()
+        self.ctxt = context.RequestContext(user_id='user_id',
+                                           project_id='project_id',
+                                           is_admin=True)
+
+    def test_services_create(self):
+        service_ref = db.service_create(self.ctxt,
+                                        {'host': 'hosttest',
+                                         'binary': 'binarytest',
+                                         'topic': 'topictest',
+                                         'report_count': 0,
+                                         'availability_zone': 'zonetest'})
+        self.assertEqual(service_ref['host'], 'hosttest')
+
+    def test_services_get(self):
+        service_ref = db.service_create(self.ctxt,
+                                        {'host': 'hosttest1',
+                                         'binary': 'binarytest1',
+                                         'topic': 'topictest1',
+                                         'report_count': 0,
+                                         'availability_zone': 'zonetest1'})
+
+        service_get_ref = db.service_get(self.ctxt, service_ref['id'])
+        self.assertEqual(service_ref['host'], 'hosttest1')
+        self.assertEqual(service_get_ref['host'], 'hosttest1')
+
+    def test_service_destroy(self):
+        service_ref = db.service_create(self.ctxt,
+                                        {'host': 'hosttest2',
+                                         'binary': 'binarytest2',
+                                         'topic': 'topictest2',
+                                         'report_count': 0,
+                                         'availability_zone': 'zonetest2'})
+        service_id = service_ref['id']
+        db.service_destroy(self.ctxt, service_id)
+        self.assertRaises(exception.ServiceNotFound, db.service_get,
+                          self.ctxt, service_id)
+
+    def test_service_update(self):
+        service_ref = db.service_create(self.ctxt,
+                                        {'host': 'hosttest3',
+                                         'binary': 'binarytest3',
+                                         'topic': 'topictest3',
+                                         'report_count': 0,
+                                         'availability_zone': 'zonetest3'})
+        service_id = service_ref['id']
+        service_update_ref = db.service_update(self.ctxt, service_id,
+                                               {'host': 'hosttest4',
+                                                'binary': 'binarytest4',
+                                                'topic': 'topictest4',
+                                                'report_count': 0,
+                                                'availability_zone':
+                                                'zonetest4'})
+        self.assertEqual(service_ref['host'], 'hosttest3')
+        self.assertEqual(service_update_ref['host'], 'hosttest4')
+
+    def test_service_get_by_host_and_topic(self):
+        service_ref = db.service_create(self.ctxt,
+                                        {'host': 'hosttest5',
+                                         'binary': 'binarytest5',
+                                         'topic': 'topictest5',
+                                         'report_count': 0,
+                                         'availability_zone': 'zonetest5'})
+
+        service_get_ref = db.service_get_by_host_and_topic(self.ctxt,
+                                                           'hosttest5',
+                                                           'topictest5')
+        self.assertEqual(service_ref['host'], 'hosttest5')
+        self.assertEqual(service_get_ref['host'], 'hosttest5')

smaug/utils.py

@@ -15,6 +15,7 @@ import os
 
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_utils import timeutils
 import six
 
 from smaug import exception
@@ -68,3 +69,12 @@ def check_string_length(value, name, min_length=0, max_length=None):
         msg = _("%(name)s has more than %(max_length)s "
                 "characters.") % {'name': name, 'max_length': max_length}
         raise exception.InvalidInput(message=msg)
+
+
+def service_is_up(service):
+    """Check whether a service is up based on last heartbeat."""
+    last_heartbeat = service['updated_at'] or service['created_at']
+
+    elapsed = (timeutils.utcnow(with_timezone=False) -
+               last_heartbeat).total_seconds()
+    return abs(elapsed) <= CONF.service_down_time