openstack/deb-python-taskflow
Code Issues Proposed changes

Unify usage of storage error exception type

Browse Source 
- Instead of formatting the exception by creating
  a storage error instance we can just let the already
  included cause make this formatting occur.
- Handle cases in impl_dir which were not catching and
  raising storage errors.
- Unify usage of 'backend internal error' messaging.
- Pass the cause to storage errors where applicable.

Change-Id: Ieb14256b202ccbfa3a96f68fa35db7e40f92114d
This commit is contained in:
Joshua Harlow
2014-02-06 18:20:56 -08:00
parent ddbdb8f3dd
commit 037a71f6ed
4 changed files with 44 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
taskflow/exceptions.py
View File

@@ -41,6 +41,8 @@ class StorageError(TaskFlowException):
"""Raised when logbook can not be read/saved/deleted.""" """Raised when logbook can not be read/saved/deleted."""
def __init__(self, message, cause=None): def __init__(self, message, cause=None):
if cause is not None:
message += ": %s" % (cause)
super(StorageError, self).__init__(message) super(StorageError, self).__init__(message)
self.cause = cause self.cause = cause

49
taskflow/persistence/backends/impl_dir.py
View File

@@ -109,8 +109,7 @@ class Connection(base.Connection):
except Exception as e: except Exception as e:
LOG.exception("Failed running locking file based session") LOG.exception("Failed running locking file based session")
# NOTE(harlowja): trap all other errors as storage errors. # NOTE(harlowja): trap all other errors as storage errors.
raise exc.StorageError("Failed running locking file based " raise exc.StorageError("Storage backend internal error", e)
"session: %s" % e, e)
def _get_logbooks(self): def _get_logbooks(self):
lb_uuids = [] lb_uuids = []
@@ -127,8 +126,13 @@ class Connection(base.Connection):
pass pass
def get_logbooks(self): def get_logbooks(self):
for b in list(self._get_logbooks()): try:
yield b books = list(self._get_logbooks())
except EnvironmentError as e:
raise exc.StorageError("Unable to fetch logbooks", e)
else:
for b in books:
yield b
@property @property
def backend(self): def backend(self):
@@ -286,11 +290,20 @@ class Connection(base.Connection):
def upgrade(self): def upgrade(self):
def _step_create(): def _step_create():
for d in (self._book_path, self._flow_path, self._task_path): for path in (self._book_path, self._flow_path, self._task_path):
misc.ensure_tree(d) try:
misc.ensure_tree(path)
except EnvironmentError as e:
raise exc.StorageError("Unable to create logbooks"
" required child path %s" % path, e)
for path in (self._backend.base_path, self._backend.lock_path):
try:
misc.ensure_tree(path)
except EnvironmentError as e:
raise exc.StorageError("Unable to create logbooks required"
" path %s" % path, e)
misc.ensure_tree(self._backend.base_path)
misc.ensure_tree(self._backend.lock_path)
self._run_with_process_lock("init", _step_create) self._run_with_process_lock("init", _step_create)
def clear_all(self): def clear_all(self):
@@ -316,32 +329,36 @@ class Connection(base.Connection):
def _destroy_tasks(task_details): def _destroy_tasks(task_details):
for task_detail in task_details: for task_detail in task_details:
task_path = os.path.join(self._task_path, task_detail.uuid)
try: try:
shutil.rmtree(os.path.join(self._task_path, shutil.rmtree(task_path)
task_detail.uuid))
except EnvironmentError as e: except EnvironmentError as e:
if e.errno != errno.ENOENT: if e.errno != errno.ENOENT:
raise raise exc.StorageError("Unable to remove task"
" directory %s" % task_path, e)
def _destroy_flows(flow_details): def _destroy_flows(flow_details):
for flow_detail in flow_details: for flow_detail in flow_details:
flow_path = os.path.join(self._flow_path, flow_detail.uuid)
self._run_with_process_lock("task", _destroy_tasks, self._run_with_process_lock("task", _destroy_tasks,
list(flow_detail)) list(flow_detail))
try: try:
shutil.rmtree(os.path.join(self._flow_path, shutil.rmtree(flow_path)
flow_detail.uuid))
except EnvironmentError as e: except EnvironmentError as e:
if e.errno != errno.ENOENT: if e.errno != errno.ENOENT:
raise raise exc.StorageError("Unable to remove flow"
" directory %s" % flow_path, e)
def _destroy_book(): def _destroy_book():
book = self._get_logbook(book_uuid) book = self._get_logbook(book_uuid)
book_path = os.path.join(self._book_path, book.uuid)
self._run_with_process_lock("flow", _destroy_flows, list(book)) self._run_with_process_lock("flow", _destroy_flows, list(book))
try: try:
shutil.rmtree(os.path.join(self._book_path, book.uuid)) shutil.rmtree(book_path)
except EnvironmentError as e: except EnvironmentError as e:
if e.errno != errno.ENOENT: if e.errno != errno.ENOENT:
raise raise exc.StorageError("Unable to remove book"
" directory %s" % book_path, e)
# Acquire all locks by going through this little hierarchy. # Acquire all locks by going through this little hierarchy.
self._run_with_process_lock("book", _destroy_book) self._run_with_process_lock("book", _destroy_book)

22
taskflow/persistence/backends/impl_sqlalchemy.py
View File

@@ -308,16 +308,14 @@ class Connection(base.Connection):
return functor(session, *args, **kwargs) return functor(session, *args, **kwargs)
except sa_exc.SQLAlchemyError as e: except sa_exc.SQLAlchemyError as e:
LOG.exception('Failed running database session') LOG.exception('Failed running database session')
raise exc.StorageError("Failed running database session: %s" % e, raise exc.StorageError("Storage backend internal error", e)
e)
def _make_session(self): def _make_session(self):
try: try:
return self._session_maker() return self._session_maker()
except sa_exc.SQLAlchemyError as e: except sa_exc.SQLAlchemyError as e:
LOG.exception('Failed creating database session') LOG.exception('Failed creating database session')
raise exc.StorageError("Failed creating database session: %s" raise exc.StorageError("Failed creating database session", e)
% e, e)
def upgrade(self): def upgrade(self):
try: try:
@@ -334,8 +332,7 @@ class Connection(base.Connection):
migration.db_sync(conn) migration.db_sync(conn)
except sa_exc.SQLAlchemyError as e: except sa_exc.SQLAlchemyError as e:
LOG.exception('Failed upgrading database version') LOG.exception('Failed upgrading database version')
raise exc.StorageError("Failed upgrading database version: %s" % e, raise exc.StorageError("Failed upgrading database version", e)
e)
def _clear_all(self, session): def _clear_all(self, session):
# NOTE(harlowja): due to how we have our relationship setup and # NOTE(harlowja): due to how we have our relationship setup and
@@ -345,7 +342,7 @@ class Connection(base.Connection):
return session.query(models.LogBook).delete() return session.query(models.LogBook).delete()
except sa_exc.DBAPIError as e: except sa_exc.DBAPIError as e:
LOG.exception('Failed clearing all entries') LOG.exception('Failed clearing all entries')
raise exc.StorageError("Failed clearing all entries: %s" % e, e) raise exc.StorageError("Failed clearing all entries", e)
def clear_all(self): def clear_all(self):
return self._run_in_session(self._clear_all) return self._run_in_session(self._clear_all)
@@ -380,8 +377,7 @@ class Connection(base.Connection):
session.delete(lb) session.delete(lb)
except sa_exc.DBAPIError as e: except sa_exc.DBAPIError as e:
LOG.exception('Failed destroying logbook') LOG.exception('Failed destroying logbook')
raise exc.StorageError("Failed destroying" raise exc.StorageError("Failed destroying logbook %s" % lb_id, e)
" logbook %s: %s" % (lb_id, e), e)
def destroy_logbook(self, book_uuid): def destroy_logbook(self, book_uuid):
return self._run_in_session(self._destroy_logbook, lb_id=book_uuid) return self._run_in_session(self._destroy_logbook, lb_id=book_uuid)
@@ -402,8 +398,7 @@ class Connection(base.Connection):
return _convert_lb_to_external(lb_m) return _convert_lb_to_external(lb_m)
except sa_exc.DBAPIError as e: except sa_exc.DBAPIError as e:
LOG.exception('Failed saving logbook') LOG.exception('Failed saving logbook')
raise exc.StorageError("Failed saving logbook %s: %s" % raise exc.StorageError("Failed saving logbook %s" % lb.uuid, e)
(lb.uuid, e), e)
def save_logbook(self, book): def save_logbook(self, book):
return self._run_in_session(self._save_logbook, lb=book) return self._run_in_session(self._save_logbook, lb=book)
@@ -415,8 +410,7 @@ class Connection(base.Connection):
return _convert_lb_to_external(lb) return _convert_lb_to_external(lb)
except sa_exc.DBAPIError as e: except sa_exc.DBAPIError as e:
LOG.exception('Failed getting logbook') LOG.exception('Failed getting logbook')
raise exc.StorageError("Failed getting logbook %s: %s" raise exc.StorageError("Failed getting logbook %s" % book_uuid, e)
% (book_uuid, e), e)
def get_logbooks(self): def get_logbooks(self):
session = self._make_session() session = self._make_session()
@@ -425,7 +419,7 @@ class Connection(base.Connection):
books = [_convert_lb_to_external(lb) for lb in raw_books] books = [_convert_lb_to_external(lb) for lb in raw_books]
except sa_exc.DBAPIError as e: except sa_exc.DBAPIError as e:
LOG.exception('Failed getting logbooks') LOG.exception('Failed getting logbooks')
raise exc.StorageError("Failed getting logbooks: %s" % e, e) raise exc.StorageError("Failed getting logbooks", e)
for lb in books: for lb in books:
yield lb yield lb

2
taskflow/persistence/backends/impl_zookeeper.py
View File

@@ -138,7 +138,7 @@ class ZkConnection(base.Connection):
except k_exc.NodeExistsError as e: except k_exc.NodeExistsError as e:
raise exc.AlreadyExists("Storage backend duplicate node: %s" % e) raise exc.AlreadyExists("Storage backend duplicate node: %s" % e)
except (k_exc.KazooException, k_exc.ZookeeperError) as e: except (k_exc.KazooException, k_exc.ZookeeperError) as e:
raise exc.StorageError("Storage backend internal error: %s" % e) raise exc.StorageError("Storage backend internal error", e)
def update_task_details(self, td): def update_task_details(self, td):
"""Update a task_detail transactionally.""" """Update a task_detail transactionally."""