basic API service: Create the base DAL into the DB

Create the basic data abstraction layer into the DB in basic
API service.I create a database table services as an 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
This commit is contained in:
chenying 2015-12-18 14:44:00 +08:00
parent bda35bb321
commit 351a0c29f2
25 changed files with 1171 additions and 8 deletions

View File

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

View File

@ -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

View File

@ -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

View File

@ -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

243
smaug/cmd/manage.py Normal file
View File

@ -0,0 +1,243 @@
#!/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'),
_('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],
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, Please try running this"
"command again as root/Administrator privilege"
"using 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)

View File

@ -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'),

16
smaug/db/__init__.py Normal file
View File

@ -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

120
smaug/db/api.py Normal file
View File

@ -0,0 +1,120 @@
# 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_get_by_args(context, host, binary):
"""Get the state of an service by node name and binary."""
return IMPL.service_get_by_args(context, host, binary)
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)

38
smaug/db/base.py Normal file
View File

@ -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()

57
smaug/db/migration.py Normal file
View File

@ -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)

View File

308
smaug/db/sqlalchemy/api.py Normal file
View File

@ -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

View File

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

View File

@ -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__)))

View File

@ -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=[]

View File

@ -0,0 +1,73 @@
# 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('disabled_reason', String(length=255)),
Column('modified_at', DateTime),
Column('rpc_current_version', String(36)),
Column('rpc_available_version', String(36)),
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()

View File

@ -0,0 +1,76 @@
# 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)
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)
rpc_current_version = Column(String(36))
rpc_available_version = Column(String(36))
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)

View File

@ -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.")

View File

@ -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(

View File

@ -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__), '..', '..', '..')))

View File

View File

@ -0,0 +1,91 @@
# 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})
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})
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})
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})
service_id = service_ref['id']
service_update_ref = db.service_update(self.ctxt, service_id,
{'host': 'hosttest4',
'binary': 'binarytest4',
'topic': 'topictest4',
'report_count': 0})
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})
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')

View File

@ -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=True) -
last_heartbeat).total_seconds()
return abs(elapsed) <= CONF.service_down_time