openstack
/
mistral
Code Issues Proposed changes

Add retries to read-only db operations 

To prevent possible failures because of operational errors
we should add retries and send additional requests to the
database.

This change will increase the overall stability of the system
and will avoid obscure errors from the database.

Change-Id: I9ddef565acabbbfaddcb96f6a850a21875e6dcd5
Closes-bug: #1730426
This commit is contained in:
Mike Fedosin 2017-11-07 05:08:05 +03:00
parent 0f39daab98
commit bff3ebf30b
21 changed files with 399 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
mistral/api/controllers/v2/action.py
View File

@ -57,7 +57,9 @@ class ActionsController(rest.RestController, hooks.HookController):
LOG.info("Fetch action [identifier=%s]", identifier)
db_model = db_api.get_action_definition(identifier)
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(db_api.get_action_definition, identifier)
return resources.Action.from_db_model(db_model)

8
mistral/api/controllers/v2/action_execution.py
View File

@ -16,6 +16,8 @@
from oslo_config import cfg
from oslo_log import log as logging
from pecan import rest
import sqlalchemy as sa
import tenacity
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
@ -49,6 +51,12 @@ def _load_deferred_output_field(action_ex):
hasattr(action_ex, 'output')
# Use retries to prevent possible failures.
@tenacity.retry(
retry=tenacity.retry_if_exception_type(sa.exc.OperationalError),
stop=tenacity.stop_after_attempt(10),
wait=tenacity.wait_incrementing(increment=100) # 0.1 seconds
)
def _get_action_execution(id):
with db_api.transaction():
return _get_action_execution_resource(db_api.get_action_execution(id))

5
mistral/api/controllers/v2/cron_trigger.py
View File

@ -41,7 +41,10 @@ class CronTriggersController(rest.RestController):
LOG.info('Fetch cron trigger [identifier=%s]', identifier)
db_model = db_api.get_cron_trigger(identifier)
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(db_api.get_cron_trigger, identifier)
return resources.CronTrigger.from_db_model(db_model)
@rest_utils.wrap_wsme_controller_exception

4
mistral/api/controllers/v2/environment.py
View File

@ -110,7 +110,9 @@ class EnvironmentController(rest.RestController):
LOG.info("Fetch environment [name=%s]", name)
db_model = db_api.get_environment(name)
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(db_api.get_environment, name)
return resources.Environment.from_db_model(db_model)

4
mistral/api/controllers/v2/event_trigger.py
View File

@ -41,7 +41,9 @@ class EventTriggersController(rest.RestController):
LOG.info('Fetch event trigger [id=%s]', id)
db_model = db_api.get_event_trigger(id)
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(db_api.get_event_trigger, id)
return resources.EventTrigger.from_db_model(db_model)

28
mistral/api/controllers/v2/execution.py
View File

@ -17,6 +17,8 @@
from oslo_log import log as logging
from pecan import rest
import sqlalchemy as sa
import tenacity
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
@ -55,6 +57,24 @@ def _get_execution_resource(wf_ex):
return resources.Execution.from_db_model(wf_ex)
# Use retries to prevent possible failures.
@tenacity.retry(
retry=tenacity.retry_if_exception_type(sa.exc.OperationalError),
stop=tenacity.stop_after_attempt(10),
wait=tenacity.wait_incrementing(increment=100) # 0.1 seconds
)
def _get_workflow_execution(id):
with db_api.transaction():
wf_ex = db_api.get_workflow_execution(id)
# If a single object is requested we need to explicitly load
# 'output' attribute. We don't do this for collections to reduce
# amount of DB queries and network traffic.
hasattr(wf_ex, 'output')
return wf_ex
# TODO(rakhmerov): Make sure to make all needed renaming on public API.
@ -72,13 +92,7 @@ class ExecutionsController(rest.RestController):
LOG.info("Fetch execution [id=%s]", id)
with db_api.transaction():
wf_ex = db_api.get_workflow_execution(id)
# If a single object is requested we need to explicitly load
# 'output' attribute. We don't do this for collections to reduce
# amount of DB queries and network traffic.
hasattr(wf_ex, 'output')
wf_ex = _get_workflow_execution(id)
return resources.Execution.from_db_model(wf_ex)

5
mistral/api/controllers/v2/member.py
View File

@ -66,7 +66,10 @@ class MembersController(rest.RestController):
member_id
)
member_db = db_api.get_resource_member(
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
member_db = r.call(
db_api.get_resource_member,
self.resource_id,
self.type,
member_id

20
mistral/api/controllers/v2/task.py
View File

@ -17,6 +17,8 @@ import json
from oslo_log import log as logging
from pecan import rest
import sqlalchemy as sa
import tenacity
from wsme import types as wtypes
import wsmeext.pecan as wsme_pecan
@ -130,6 +132,19 @@ class TaskExecutionsController(rest.RestController):
)
# Use retries to prevent possible failures.
@tenacity.retry(
retry=tenacity.retry_if_exception_type(sa.exc.OperationalError),
stop=tenacity.stop_after_attempt(10),
wait=tenacity.wait_incrementing(increment=100) # 0.1 seconds
)
def _get_task_execution(id):
with db_api.transaction():
task_ex = db_api.get_task_execution(id)
return _get_task_resource_with_result(task_ex)
class TasksController(rest.RestController):
action_executions = action_execution.TasksActionExecutionController()
workflow_executions = TaskExecutionsController()
@ -144,10 +159,7 @@ class TasksController(rest.RestController):
acl.enforce('tasks:get', context.ctx())
LOG.info("Fetch task [id=%s]", id)
with db_api.transaction():
task_ex = db_api.get_task_execution(id)
return _get_task_resource_with_result(task_ex)
return _get_task_execution(id)
@rest_utils.wrap_wsme_controller_exception
@wsme_pecan.wsexpose(resources.Tasks, types.uuid, int, types.uniquelist,

4
mistral/api/controllers/v2/workbook.py
View File

@ -53,7 +53,9 @@ class WorkbooksController(rest.RestController, hooks.HookController):
LOG.info("Fetch workbook [name=%s]", name)
db_model = db_api.get_workbook(name)
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(db_api.get_workbook, name)
return resources.Workbook.from_db_model(db_model)

5
mistral/api/controllers/v2/workflow.py
View File

@ -87,7 +87,10 @@ class WorkflowsController(rest.RestController, hooks.HookController):
LOG.info("Fetch workflow [identifier=%s]", identifier)
db_model = db_api.get_workflow_definition(
# Use retries to prevent possible failures.
r = rest_utils.create_db_retry_object()
db_model = r.call(
db_api.get_workflow_definition,
identifier,
namespace=namespace
)

30
mistral/tests/unit/api/v2/test_action_executions.py
View File

@ -18,9 +18,9 @@ import datetime
import json
import mock
from oslo_config import cfg
import oslo_messaging
import sqlalchemy as sa
from mistral.api.controllers.v2 import action_execution
from mistral.db.v2 import api as db_api
@ -229,6 +229,19 @@ class TestActionExecutionsController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(ACTION_EX, resp.json)
@mock.patch.object(db_api, 'get_action_execution')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
ACTION_EX_DB # Successful run
]
resp = self.app.get('/v2/action_executions/123')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(ACTION_EX, resp.json)
def test_basic_get(self):
resp = self.app.get('/v2/action_executions/')
self.assertEqual(200, resp.status_int)
@ -502,6 +515,21 @@ class TestActionExecutionsController(base.APITest):
self.assertEqual(1, len(resp.json['action_executions']))
self.assertDictEqual(ACTION_EX, resp.json['action_executions'][0])
@mock.patch.object(db_api, 'get_action_executions')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[ACTION_EX_DB] # Successful run
]
resp = self.app.get('/v2/action_executions')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['action_executions']))
self.assertDictEqual(ACTION_EX, resp.json['action_executions'][0])
@mock.patch.object(db_api, 'get_action_executions', MOCK_ACTIONS)
@mock.patch.object(rest_utils, 'get_all')
def test_get_all_with_and_without_output(self, mock_get_all):

30
mistral/tests/unit/api/v2/test_actions.py
View File

@ -13,7 +13,9 @@
# limitations under the License.
import copy
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -133,6 +135,19 @@ class TestActionsController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(ACTION, resp.json)
@mock.patch.object(db_api, 'get_action_definition')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
ACTION_DB # Successful run
]
resp = self.app.get('/v2/actions/my_action')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(ACTION, resp.json)
@mock.patch.object(
db_api, "get_action_definition", MOCK_NOT_FOUND)
def test_get_not_found(self):
@ -323,6 +338,21 @@ class TestActionsController(base.APITest):
self.assertEqual(1, len(resp.json['actions']))
self.assertDictEqual(ACTION, resp.json['actions'][0])
@mock.patch.object(db_api, 'get_action_definitions')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[ACTION_DB] # Successful run
]
resp = self.app.get('/v2/actions')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['actions']))
self.assertDictEqual(ACTION, resp.json['actions'][0])
@mock.patch.object(
db_api, "get_action_definitions", MOCK_EMPTY)
def test_get_all_empty(self):

30
mistral/tests/unit/api/v2/test_cron_triggers.py
View File

@ -14,7 +14,9 @@
import copy
import json
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -77,6 +79,19 @@ class TestCronTriggerController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TRIGGER, resp.json)
@mock.patch.object(db_api, 'get_cron_trigger')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
TRIGGER_DB # Successful run
]
resp = self.app.get('/v2/cron_triggers/my_cron_trigger')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TRIGGER, resp.json)
@mock.patch.object(db_api, "get_cron_trigger",
return_value=TRIGGER_DB_WITH_PROJECT_ID)
def test_get_within_project_id(self, mock_get):
@ -178,6 +193,21 @@ class TestCronTriggerController(base.APITest):
self.assertEqual(1, len(resp.json['cron_triggers']))
self.assertDictEqual(TRIGGER, resp.json['cron_triggers'][0])
@mock.patch.object(db_api, 'get_cron_triggers')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[TRIGGER_DB] # Successful run
]
resp = self.app.get('/v2/cron_triggers')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['cron_triggers']))
self.assertDictEqual(TRIGGER, resp.json['cron_triggers'][0])
@mock.patch.object(db_api, 'get_cron_triggers')
@mock.patch('mistral.context.MistralContext.from_environ')
def test_get_all_projects_admin(self, mock_context, mock_get_triggers):

29
mistral/tests/unit/api/v2/test_environment.py
View File

@ -18,6 +18,7 @@ import json
import mock
import six
import sqlalchemy as sa
from mistral.api.controllers.v2 import resources
from mistral.db.v2 import api as db_api
@ -160,6 +161,21 @@ class TestEnvironmentController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['environments']))
@mock.patch.object(db_api, 'get_environments')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[ENVIRONMENT_DB] # Successful run
]
resp = self.app.get('/v2/environments')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['environments']))
self._assert_dict_equal(ENVIRONMENT, resp.json['environments'][0])
def test_get_all_empty(self):
resp = self.app.get('/v2/environments')
@ -173,6 +189,19 @@ class TestEnvironmentController(base.APITest):
self.assertEqual(200, resp.status_int)
self._assert_dict_equal(ENVIRONMENT, resp.json)
@mock.patch.object(db_api, 'get_environment')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
ENVIRONMENT_DB # Successful run
]
resp = self.app.get('/v2/environments/123')
self.assertEqual(200, resp.status_int)
self._assert_dict_equal(ENVIRONMENT, resp.json)
@mock.patch.object(db_api, 'get_environment',
return_value=ENVIRONMENT_DB_WITH_PROJECT_ID)
def test_get_within_project_id(self, mock_get):

32
mistral/tests/unit/api/v2/test_event_trigger.py
View File

@ -14,7 +14,9 @@
import copy
import json
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -74,6 +76,21 @@ class TestEventTriggerController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TRIGGER, resp.json)
@mock.patch.object(db_api, 'get_event_trigger')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
TRIGGER_DB # Successful run
]
resp = self.app.get(
'/v2/event_triggers/09cc56a9-d15e-4494-a6e2-c4ec8bdaacae'
)
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TRIGGER, resp.json)
@mock.patch.object(db_api, "get_event_trigger", MOCK_NOT_FOUND)
def test_get_not_found(self):
resp = self.app.get(
@ -230,6 +247,21 @@ class TestEventTriggerController(base.APITest):
self.assertEqual(1, len(resp.json['event_triggers']))
self.assertDictEqual(TRIGGER, resp.json['event_triggers'][0])
@mock.patch.object(db_api, 'get_event_triggers')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[TRIGGER_DB] # Successful run
]
resp = self.app.get('/v2/event_triggers')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['event_triggers']))
self.assertDictEqual(TRIGGER, resp.json['event_triggers'][0])
@mock.patch('mistral.db.v2.api.get_event_triggers')
@mock.patch('mistral.context.MistralContext.from_environ')
def test_get_all_projects_admin(self, mock_context, mock_get_wf_defs):

29
mistral/tests/unit/api/v2/test_executions.py
View File

@ -23,6 +23,7 @@ import mock
from oslo_config import cfg
import oslo_messaging
from oslo_utils import uuidutils
import sqlalchemy as sa
from webtest import app as webtest_app
from mistral.api.controllers.v2 import execution
@ -143,6 +144,19 @@ class TestExecutionsController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WF_EX_JSON_WITH_DESC, resp.json)
@mock.patch.object(db_api, 'get_workflow_execution')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
WF_EX # Successful run
]
resp = self.app.get('/v2/executions/123')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WF_EX_JSON_WITH_DESC, resp.json)
@mock.patch.object(db_api, 'get_workflow_execution', MOCK_SUB_WF_EX)
def test_get_sub_wf_ex(self):
resp = self.app.get('/v2/executions/123')
@ -529,6 +543,21 @@ class TestExecutionsController(base.APITest):
self.assertEqual(1, len(resp.json['executions']))
self.assertDictEqual(WF_EX_JSON_WITH_DESC, resp.json['executions'][0])
@mock.patch.object(db_api, 'get_workflow_executions')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[WF_EX] # Successful run
]
resp = self.app.get('/v2/executions')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['executions']))
self.assertDictEqual(WF_EX_JSON_WITH_DESC, resp.json['executions'][0])
@mock.patch.object(db_api, 'get_workflow_executions', MOCK_EMPTY)
def test_get_all_empty(self):
resp = self.app.get('/v2/executions')

25
mistral/tests/unit/api/v2/test_members.py
View File

@ -17,6 +17,7 @@ import copy
import mock
from oslo_config import cfg
from oslo_utils import uuidutils
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.services import security
@ -165,6 +166,30 @@ class TestMembersController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertEqual(0, len(resp.json['members']))
@mock.patch('mistral.context.AuthHook.before')
@mock.patch.object(db_api, 'get_resource_member')
def test_get_operational_error(self, mocked_get, auth_mock):
member_data = {'member_id': '11-22-33'}
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
member_data # Successful run
]
resp = self.app.post_json(MEMBER_URL, member_data)
self.assertEqual(201, resp.status_int)
# Using mock to switch to another tenant.
get_mock = mock.MagicMock(return_value='other_tenant')
with mock.patch(GET_PROJECT_PATH, get_mock):
resp = self.app.get(MEMBER_URL)
self.assertEqual(200, resp.status_int)
self.assertEqual(0, len(resp.json['members']))
@mock.patch('mistral.context.AuthHook.before')
def test_get_memberships_nonexistent_wf(self, auth_mock):
nonexistent_wf_id = uuidutils.generate_uuid()

30
mistral/tests/unit/api/v2/test_tasks.py
View File

@ -16,7 +16,9 @@
import copy
import datetime
import json
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -159,6 +161,19 @@ class TestTasksController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TASK, resp.json)
@mock.patch.object(db_api, 'get_task_execution')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
TASK_EX # Successful run
]
resp = self.app.get('/v2/tasks/123')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(TASK, resp.json)
@mock.patch.object(db_api, 'get_task_execution', MOCK_NOT_FOUND)
def test_get_not_found(self):
resp = self.app.get('/v2/tasks/123', expect_errors=True)
@ -174,6 +189,21 @@ class TestTasksController(base.APITest):
self.assertEqual(1, len(resp.json['tasks']))
self.assertDictEqual(TASK_WITHOUT_RESULT, resp.json['tasks'][0])
@mock.patch.object(db_api, 'get_task_executions')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[TASK_EX] # Successful run
]
resp = self.app.get('/v2/tasks')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['tasks']))
self.assertDictEqual(TASK_WITHOUT_RESULT, resp.json['tasks'][0])
@mock.patch.object(db_api, 'get_task_execution',
return_value=TASK_EX_WITH_PROJECT_ID)
def test_get_within_project_id(self, mock_get):

30
mistral/tests/unit/api/v2/test_workbooks.py
View File

@ -15,7 +15,9 @@
import copy
import datetime
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -112,6 +114,19 @@ class TestWorkbooksController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WORKBOOK, resp.json)
@mock.patch.object(db_api, 'get_workbook')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
WORKBOOK_DB # Successful run
]
resp = self.app.get('/v2/workbooks/123')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WORKBOOK, resp.json)
@mock.patch.object(db_api, "get_workbook", MOCK_NOT_FOUND)
def test_get_not_found(self):
resp = self.app.get('/v2/workbooks/123', expect_errors=True)
@ -213,6 +228,21 @@ class TestWorkbooksController(base.APITest):
self.assertEqual(1, len(resp.json['workbooks']))
self.assertDictEqual(WORKBOOK, resp.json['workbooks'][0])
@mock.patch.object(db_api, 'get_workbooks')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[WORKBOOK_DB] # Successful run
]
resp = self.app.get('/v2/workbooks')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['workbooks']))
self.assertDictEqual(WORKBOOK, resp.json['workbooks'][0])
@mock.patch.object(db_api, "get_workbooks", MOCK_EMPTY)
def test_get_all_empty(self):
resp = self.app.get('/v2/workbooks')

28
mistral/tests/unit/api/v2/test_workflows.py
View File

@ -17,6 +17,7 @@ import copy
import datetime
import mock
import sqlalchemy as sa
from mistral.db.v2 import api as db_api
from mistral.db.v2.sqlalchemy import models
@ -197,6 +198,19 @@ class TestWorkflowsController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WF, resp.json)
@mock.patch.object(db_api, 'get_workflow_definition')
def test_get_operational_error(self, mocked_get):
mocked_get.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
WF_DB # Successful run
]
resp = self.app.get('/v2/workflows/123')
self.assertEqual(200, resp.status_int)
self.assertDictEqual(WF, resp.json)
@mock.patch.object(db_api, "get_workflow_definition", MOCK_WF_WITH_INPUT)
def test_get_with_input(self):
resp = self.app.get('/v2/workflows/123')
@ -436,9 +450,21 @@ class TestWorkflowsController(base.APITest):
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['workflows']))
self.assertDictEqual(WF, resp.json['workflows'][0])
print(resp.json['workflows'][0])
@mock.patch.object(db_api, 'get_workflow_definitions')
def test_get_all_operational_error(self, mocked_get_all):
mocked_get_all.side_effect = [
# Emulating DB OperationalError
sa.exc.OperationalError('Mock', 'mock', 'mock'),
[WF_DB] # Successful run
]
resp = self.app.get('/v2/workflows')
self.assertEqual(200, resp.status_int)
self.assertEqual(1, len(resp.json['workflows']))
self.assertDictEqual(WF, resp.json['workflows'][0])
@mock.patch.object(db_api, "get_workflow_definitions", MOCK_EMPTY)

62
mistral/utils/rest_utils.py
View File

@ -19,7 +19,8 @@ import json
from oslo_log import log as logging
import pecan
import six
import sqlalchemy as sa
import tenacity
import webob
from wsme import exc as wsme_exc
@ -174,28 +175,7 @@ def get_all(list_cls, cls, get_all_function, get_function,
if marker:
marker_obj = get_function(marker)
rest_resources = []
# If only certain fields are requested then we ignore "resource_function"
# parameter because it doesn't make sense anymore.
if fields:
db_list = get_all_function(
limit=limit,
marker=marker_obj,
sort_keys=sort_keys,
sort_dirs=sort_dirs,
fields=fields,
insecure=insecure,
**filters
)
for obj_values in db_list:
# Note: in case if only certain fields have been requested
# "db_list" contains tuples with values of db objects.
rest_resources.append(
cls.from_tuples(zip(fields, obj_values))
)
else:
def _get_all_function():
with db_api.transaction():
db_models = get_all_function(
limit=limit,
@ -214,6 +194,34 @@ def get_all(list_cls, cls, get_all_function, get_function,
rest_resources.append(rest_resource)
rest_resources = []
r = create_db_retry_object()
# If only certain fields are requested then we ignore "resource_function"
# parameter because it doesn't make sense anymore.
if fields:
# Use retries to prevent possible failures.
db_list = r.call(
get_all_function,
limit=limit,
marker=marker_obj,
sort_keys=sort_keys,
sort_dirs=sort_dirs,
fields=fields,
insecure=insecure,
**filters
)
for obj_values in db_list:
# Note: in case if only certain fields have been requested
# "db_list" contains tuples with values of db objects.
rest_resources.append(
cls.from_tuples(zip(fields, obj_values))
)
else:
r.call(_get_all_function)
return list_cls.convert_with_links(
rest_resources,
limit,
@ -223,3 +231,11 @@ def get_all(list_cls, cls, get_all_function, get_function,
fields=','.join(fields) if fields else '',
**filters
)
def create_db_retry_object():
return tenacity.Retrying(
retry=tenacity.retry_if_exception_type(sa.exc.OperationalError),
stop=tenacity.stop_after_attempt(10),
wait=tenacity.wait_incrementing(increment=100) # 0.1 seconds
)