openstack/ironic-inspector
Code Issues Proposed changes

Merge "Introduce oslo.messaging and sync rpc call"

Browse Source
This commit is contained in:
Zuul 2018-06-01 02:32:41 +00:00 committed by Gerrit Code Review
commit c6ad0f0ead
11 changed files with 328 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
ironic_inspector/common/rpc.py Normal file
View File

@ -0,0 +1,56 @@
# 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 oslo_config import cfg
import oslo_messaging as messaging
from oslo_messaging.rpc import dispatcher
from ironic_inspector.conductor import manager
CONF = cfg.CONF
_SERVER = None
TRANSPORT = None
TOPIC = 'ironic-inspector-worker'
SERVER_NAME = 'ironic-inspector-rpc-server'
def get_transport():
global TRANSPORT
if TRANSPORT is None:
TRANSPORT = messaging.get_rpc_transport(CONF, url='fake://')
return TRANSPORT
def get_client():
target = messaging.Target(topic=TOPIC, server=SERVER_NAME,
version='1.0')
transport = get_transport()
return messaging.RPCClient(transport, target)
def get_server():
"""Get the singleton RPC server."""
global _SERVER
if _SERVER is None:
transport = get_transport()
target = messaging.Target(topic=TOPIC, server=SERVER_NAME,
version='1.0')
mgr = manager.ConductorManager()
_SERVER = messaging.get_rpc_server(
transport, target, [mgr], executor='eventlet',
access_policy=dispatcher.DefaultRPCAccessPolicy)
return _SERVER

0
ironic_inspector/conductor/__init__.py Normal file
View File

37
ironic_inspector/conductor/manager.py Normal file
View File

@ -0,0 +1,37 @@
# 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 oslo_messaging as messaging
from ironic_inspector import introspect
from ironic_inspector import process
from ironic_inspector import utils
class ConductorManager(object):
"""ironic inspector conductor manager"""
RPC_API_VERSION = '1.0'
target = messaging.Target(version=RPC_API_VERSION)
@messaging.expected_exceptions(utils.Error)
def do_introspection(self, context, node_id, token=None):
introspect.introspect(node_id, token=token)
@messaging.expected_exceptions(utils.Error)
def do_abort(self, context, node_id, token=None):
introspect.abort(node_id, token=token)
@messaging.expected_exceptions(utils.Error)
def do_reapply(self, context, node_id, token=None):
process.reapply(node_id)

14
ironic_inspector/main.py
View File

@ -23,10 +23,10 @@ from ironic_inspector import api_tools
from ironic_inspector.common import context
from ironic_inspector.common.i18n import _
from ironic_inspector.common import ironic as ir_utils
from ironic_inspector.common import rpc
from ironic_inspector.common import swift
import ironic_inspector.conf
from ironic_inspector.conf import opts as conf_opts
from ironic_inspector import introspect
from ironic_inspector import node_cache
from ironic_inspector import process
from ironic_inspector import rules
@ -239,8 +239,9 @@ def api_continue():
methods=['GET', 'POST'])
def api_introspection(node_id):
if flask.request.method == 'POST':
introspect.introspect(node_id,
token=flask.request.headers.get('X-Auth-Token'))
client = rpc.get_client()
client.call({}, 'do_introspection', node_id=node_id,
token=flask.request.headers.get('X-Auth-Token'))
return '', 202
else:
node_info = node_cache.get_node(node_id)
@ -263,7 +264,9 @@ def api_introspection_statuses():
@api('/v1/introspection/<node_id>/abort', rule="introspection:abort",
methods=['POST'])
def api_introspection_abort(node_id):
introspect.abort(node_id, token=flask.request.headers.get('X-Auth-Token'))
client = rpc.get_client()
client.call({}, 'do_abort', node_id=node_id,
token=flask.request.headers.get('X-Auth-Token'))
return '', 202
@ -291,7 +294,8 @@ def api_introspection_reapply(node_id):
'supported yet'), code=400)
if CONF.processing.store_data == 'swift':
process.reapply(node_id)
client = rpc.get_client()
client.call({}, 'do_reapply', node_id=node_id)
return '', 202
else:
return error_response(_('Inspector is not configured to store'

7
ironic_inspector/test/functional.py
View File

@ -136,6 +136,13 @@ class Base(base.NodeTest):
conf_file = get_test_conf_file()
self.cfg.set_config_files([conf_file])
# FIXME(milan) FakeListener.poll calls time.sleep() which leads to
# busy polling with no sleep at all, effectively blocking the whole
# process by consuming all CPU cycles in a single thread. MonkeyPatch
# with eventlet.sleep seems to help this.
self.useFixture(fixtures.MonkeyPatch(
'oslo_messaging._drivers.impl_fake.time.sleep', eventlet.sleep))
def tearDown(self):
super(Base, self).tearDown()
node_cache._delete_node(self.uuid)

138
ironic_inspector/test/unit/test_main.py
View File

@ -15,13 +15,15 @@ import datetime
import json
import unittest
import fixtures
import mock
import oslo_messaging as messaging
from oslo_utils import uuidutils
from ironic_inspector.common import ironic as ir_utils
from ironic_inspector.common import rpc
import ironic_inspector.conf
from ironic_inspector.conf import opts as conf_opts
from ironic_inspector import introspect
from ironic_inspector import introspection_state as istate
from ironic_inspector import main
from ironic_inspector import node_cache
@ -49,36 +51,44 @@ class BaseAPITest(test_base.BaseTest):
class TestApiIntrospect(BaseAPITest):
@mock.patch.object(introspect, 'introspect', autospec=True)
def test_introspect_no_authentication(self, introspect_mock):
CONF.set_override('auth_strategy', 'noauth')
res = self.app.post('/v1/introspection/%s' % self.uuid)
self.assertEqual(202, res.status_code)
introspect_mock.assert_called_once_with(self.uuid,
token=None)
def setUp(self):
super(TestApiIntrospect, self).setUp()
self.rpc_get_client_mock = self.useFixture(
fixtures.MockPatchObject(rpc, 'get_client', autospec=True)).mock
self.client_mock = mock.MagicMock(spec=messaging.RPCClient)
self.rpc_get_client_mock.return_value = self.client_mock
def test_introspect_no_authentication(self):
CONF.set_override('auth_strategy', 'noauth')
@mock.patch.object(introspect, 'introspect', autospec=True)
def test_intospect_failed(self, introspect_mock):
introspect_mock.side_effect = utils.Error("boom")
res = self.app.post('/v1/introspection/%s' % self.uuid)
self.assertEqual(202, res.status_code)
self.client_mock.call.assert_called_once_with({}, 'do_introspection',
node_id=self.uuid,
token=None)
def test_intospect_failed(self):
self.client_mock.call.side_effect = utils.Error("boom")
res = self.app.post('/v1/introspection/%s' % self.uuid)
self.assertEqual(400, res.status_code)
self.assertEqual(
'boom',
json.loads(res.data.decode('utf-8'))['error']['message'])
introspect_mock.assert_called_once_with(
self.uuid,
token=None)
self.client_mock.call.assert_called_once_with({}, 'do_introspection',
node_id=self.uuid,
token=None)
@mock.patch.object(utils, 'check_auth', autospec=True)
@mock.patch.object(introspect, 'introspect', autospec=True)
def test_introspect_failed_authentication(self, introspect_mock,
auth_mock):
def test_introspect_failed_authentication(self, auth_mock):
CONF.set_override('auth_strategy', 'keystone')
auth_mock.side_effect = utils.Error('Boom', code=403)
res = self.app.post('/v1/introspection/%s' % self.uuid,
headers={'X-Auth-Token': 'token'})
self.assertEqual(403, res.status_code)
self.assertFalse(introspect_mock.called)
self.assertFalse(self.client_mock.call.called)
@mock.patch.object(process, 'process', autospec=True)
@ -107,45 +117,57 @@ class TestApiContinue(BaseAPITest):
self.assertFalse(process_mock.called)
@mock.patch.object(introspect, 'abort', autospec=True)
class TestApiAbort(BaseAPITest):
def test_ok(self, abort_mock):
abort_mock.return_value = '', 202
def setUp(self):
super(TestApiAbort, self).setUp()
self.rpc_get_client_mock = self.useFixture(
fixtures.MockPatchObject(rpc, 'get_client', autospec=True)).mock
self.client_mock = mock.MagicMock(spec=messaging.RPCClient)
self.rpc_get_client_mock.return_value = self.client_mock
def test_ok(self):
res = self.app.post('/v1/introspection/%s/abort' % self.uuid,
headers={'X-Auth-Token': 'token'})
abort_mock.assert_called_once_with(self.uuid, token='token')
self.client_mock.call.assert_called_once_with({}, 'do_abort',
node_id=self.uuid,
token='token')
self.assertEqual(202, res.status_code)
self.assertEqual(b'', res.data)
def test_no_authentication(self, abort_mock):
abort_mock.return_value = b'', 202
def test_no_authentication(self):
res = self.app.post('/v1/introspection/%s/abort' % self.uuid)
abort_mock.assert_called_once_with(self.uuid, token=None)
self.client_mock.call.assert_called_once_with({}, 'do_abort',
node_id=self.uuid,
token=None)
self.assertEqual(202, res.status_code)
self.assertEqual(b'', res.data)
def test_node_not_found(self, abort_mock):
def test_node_not_found(self):
exc = utils.Error("Not Found.", code=404)
abort_mock.side_effect = exc
self.client_mock.call.side_effect = exc
res = self.app.post('/v1/introspection/%s/abort' % self.uuid)
abort_mock.assert_called_once_with(self.uuid, token=None)
self.client_mock.call.assert_called_once_with({}, 'do_abort',
node_id=self.uuid,
token=None)
self.assertEqual(404, res.status_code)
data = json.loads(str(res.data.decode()))
self.assertEqual(str(exc), data['error']['message'])
def test_abort_failed(self, abort_mock):
def test_abort_failed(self):
exc = utils.Error("Locked.", code=409)
abort_mock.side_effect = exc
self.client_mock.call.side_effect = exc
res = self.app.post('/v1/introspection/%s/abort' % self.uuid)
abort_mock.assert_called_once_with(self.uuid, token=None)
self.client_mock.call.assert_called_once_with({}, 'do_abort',
node_id=self.uuid,
token=None)
self.assertEqual(409, res.status_code)
data = json.loads(res.data.decode())
self.assertEqual(str(exc), data['error']['message'])
@ -297,29 +319,33 @@ class TestApiGetData(BaseAPITest):
get_mock.assert_called_once_with('name1', fields=['uuid'])
@mock.patch.object(process, 'reapply', autospec=True)
class TestApiReapply(BaseAPITest):
def setUp(self):
super(TestApiReapply, self).setUp()
self.rpc_get_client_mock = self.useFixture(
fixtures.MockPatchObject(rpc, 'get_client', autospec=True)).mock
self.client_mock = mock.MagicMock(spec=messaging.RPCClient)
self.rpc_get_client_mock.return_value = self.client_mock
CONF.set_override('store_data', 'swift', 'processing')
def test_ok(self, reapply_mock):
def test_ok(self):
self.app.post('/v1/introspection/%s/data/unprocessed' %
self.uuid)
reapply_mock.assert_called_once_with(self.uuid)
self.client_mock.call.assert_called_once_with({}, 'do_reapply',
node_id=self.uuid)
def test_user_data(self, reapply_mock):
def test_user_data(self):
res = self.app.post('/v1/introspection/%s/data/unprocessed' %
self.uuid, data='some data')
self.assertEqual(400, res.status_code)
message = json.loads(res.data.decode())['error']['message']
self.assertEqual('User data processing is not supported yet',
message)
self.assertFalse(reapply_mock.called)
self.assertFalse(self.client_mock.call.called)
def test_swift_disabled(self, reapply_mock):
def test_swift_disabled(self):
CONF.set_override('store_data', 'none', 'processing')
res = self.app.post('/v1/introspection/%s/data/unprocessed' %
@ -330,35 +356,11 @@ class TestApiReapply(BaseAPITest):
'data. Set the [processing] store_data '
'configuration option to change this.',
message)
self.assertFalse(reapply_mock.called)
self.assertFalse(self.client_mock.call.called)
def test_node_locked(self, reapply_mock):
exc = utils.Error('Locked.', code=409)
reapply_mock.side_effect = exc
res = self.app.post('/v1/introspection/%s/data/unprocessed' %
self.uuid)
self.assertEqual(409, res.status_code)
message = json.loads(res.data.decode())['error']['message']
self.assertEqual(str(exc), message)
reapply_mock.assert_called_once_with(self.uuid)
def test_node_not_found(self, reapply_mock):
exc = utils.Error('Not found.', code=404)
reapply_mock.side_effect = exc
res = self.app.post('/v1/introspection/%s/data/unprocessed' %
self.uuid)
self.assertEqual(404, res.status_code)
message = json.loads(res.data.decode())['error']['message']
self.assertEqual(str(exc), message)
reapply_mock.assert_called_once_with(self.uuid)
def test_generic_error(self, reapply_mock):
def test_generic_error(self):
exc = utils.Error('Oops', code=400)
reapply_mock.side_effect = exc
self.client_mock.call.side_effect = exc
res = self.app.post('/v1/introspection/%s/data/unprocessed' %
self.uuid)
@ -366,7 +368,8 @@ class TestApiReapply(BaseAPITest):
self.assertEqual(400, res.status_code)
message = json.loads(res.data.decode())['error']['message']
self.assertEqual(str(exc), message)
reapply_mock.assert_called_once_with(self.uuid)
self.client_mock.call.assert_called_once_with({}, 'do_reapply',
node_id=self.uuid)
class TestApiRules(BaseAPITest):
@ -566,8 +569,11 @@ class TestApiVersions(BaseAPITest):
# API version on unknown pages
self._check_version_present(self.app.get('/v1/foobar'))
@mock.patch.object(rpc, 'get_client', autospec=True)
@mock.patch.object(node_cache, 'get_node', autospec=True)
def test_usual_requests(self, get_mock):
def test_usual_requests(self, get_mock, rpc_mock):
client_mock = mock.MagicMock(spec=messaging.RPCClient)
rpc_mock.return_value = client_mock
get_mock.return_value = node_cache.NodeInfo(uuid=self.uuid,
started_at=42.0)
# Successful

133
ironic_inspector/test/unit/test_manager.py Normal file
View File

@ -0,0 +1,133 @@
# 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 mock
import oslo_messaging as messaging
from ironic_inspector.conductor import manager
import ironic_inspector.conf
from ironic_inspector import introspect
from ironic_inspector import process
from ironic_inspector.test import base as test_base
from ironic_inspector import utils
CONF = ironic_inspector.conf.CONF
class BaseManagerTest(test_base.NodeTest):
def setUp(self):
super(BaseManagerTest, self).setUp()
self.manager = manager.ConductorManager()
self.context = {}
self.token = None
class TestManagerIntrospect(BaseManagerTest):
@mock.patch.object(introspect, 'introspect', autospec=True)
def test_do_introspect(self, introspect_mock):
self.manager.do_introspection(self.context, self.uuid, self.token)
introspect_mock.assert_called_once_with(self.uuid, self.token)
@mock.patch.object(introspect, 'introspect', autospec=True)
def test_introspect_failed(self, introspect_mock):
introspect_mock.side_effect = utils.Error("boom")
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_introspection,
self.context, self.uuid, self.token)
self.assertEqual(utils.Error, exc.exc_info[0])
introspect_mock.assert_called_once_with(self.uuid, token=None)
class TestManagerAbort(BaseManagerTest):
@mock.patch.object(introspect, 'abort', autospec=True)
def test_abort_ok(self, abort_mock):
self.manager.do_abort(self.context, self.uuid, self.token)
abort_mock.assert_called_once_with(self.uuid, token=self.token)
@mock.patch.object(introspect, 'abort', autospec=True)
def test_abort_node_not_found(self, abort_mock):
abort_mock.side_effect = utils.Error("Not Found.", code=404)
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_abort,
self.context, self.uuid, self.token)
self.assertEqual(utils.Error, exc.exc_info[0])
abort_mock.assert_called_once_with(self.uuid, token=None)
@mock.patch.object(introspect, 'abort', autospec=True)
def test_abort_failed(self, abort_mock):
exc = utils.Error("Locked.", code=409)
abort_mock.side_effect = exc
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_abort,
self.context, self.uuid, self.token)
self.assertEqual(utils.Error, exc.exc_info[0])
abort_mock.assert_called_once_with(self.uuid, token=None)
@mock.patch.object(process, 'reapply', autospec=True)
class TestManagerReapply(BaseManagerTest):
def setUp(self):
super(TestManagerReapply, self).setUp()
CONF.set_override('store_data', 'swift', 'processing')
def test_ok(self, reapply_mock):
self.manager.do_reapply(self.context, self.uuid)
reapply_mock.assert_called_once_with(self.uuid)
def test_node_locked(self, reapply_mock):
exc = utils.Error('Locked.', code=409)
reapply_mock.side_effect = exc
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_reapply,
self.context, self.uuid)
self.assertEqual(utils.Error, exc.exc_info[0])
self.assertIn('Locked.', str(exc.exc_info[1]))
self.assertEqual(409, exc.exc_info[1].http_code)
reapply_mock.assert_called_once_with(self.uuid)
def test_node_not_found(self, reapply_mock):
exc = utils.Error('Not found.', code=404)
reapply_mock.side_effect = exc
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_reapply,
self.context, self.uuid)
self.assertEqual(utils.Error, exc.exc_info[0])
self.assertIn('Not found.', str(exc.exc_info[1]))
self.assertEqual(404, exc.exc_info[1].http_code)
reapply_mock.assert_called_once_with(self.uuid)
def test_generic_error(self, reapply_mock):
exc = utils.Error('Oops', code=400)
reapply_mock.side_effect = exc
exc = self.assertRaises(messaging.rpc.ExpectedException,
self.manager.do_reapply,
self.context, self.uuid)
self.assertEqual(utils.Error, exc.exc_info[0])
self.assertIn('Oops', str(exc.exc_info[1]))
self.assertEqual(400, exc.exc_info[1].http_code)
reapply_mock.assert_called_once_with(self.uuid)

6
ironic_inspector/test/unit/test_wsgi_service.py
View File

@ -20,6 +20,7 @@ import fixtures
import mock
from oslo_config import cfg
from ironic_inspector.common import rpc
from ironic_inspector.test import base as test_base
from ironic_inspector import wsgi_service
@ -40,6 +41,8 @@ class BaseWSGITest(test_base.BaseTest):
self.mock_log = self.useFixture(fixtures.MockPatchObject(
wsgi_service, 'LOG')).mock
self.service = wsgi_service.WSGIService()
self.mock_rpc_server = self.useFixture(fixtures.MockPatchObject(
rpc, 'get_server')).mock
class TestWSGIServiceInitMiddleware(BaseWSGITest):
@ -199,6 +202,8 @@ class TestWSGIServiceRun(BaseWSGITest):
self.mock__create_ssl_context.assert_called_once_with()
self.mock__init_middleware.assert_called_once_with()
self.mock__init_host.assert_called_once_with()
self.mock_rpc_server.assert_called_once_with()
self.service.rpc_server.start.assert_called_once_with()
self.app.run.assert_called_once_with(
host=CONF.listen_address, port=CONF.listen_port,
ssl_context=self.mock__create_ssl_context.return_value)
@ -247,6 +252,7 @@ class TestWSGIServiceShutdown(BaseWSGITest):
self.service, '_periodics_worker')).mock
self.mock_exit = self.useFixture(fixtures.MockPatchObject(
wsgi_service.sys, 'exit')).mock
self.service.rpc_server = self.mock_rpc_server
def test_shutdown(self):
class MyError(Exception):

6
ironic_inspector/wsgi_service.py
View File

@ -23,6 +23,7 @@ from oslo_log import log
from oslo_utils import reflection
from ironic_inspector.common import ironic as ir_utils
from ironic_inspector.common import rpc
from ironic_inspector import db
from ironic_inspector import main as app
from ironic_inspector import node_cache
@ -151,6 +152,8 @@ class WSGIService(object):
LOG.debug('Shutting down')
self.rpc_server.stop()
if self._periodics_worker is not None:
try:
self._periodics_worker.stop()
@ -185,6 +188,9 @@ class WSGIService(object):
self._init_host()
self.rpc_server = rpc.get_server()
self.rpc_server.start()
try:
self.app.run(**app_kwargs)
except Exception as e:

1
lower-constraints.txt
View File

@ -67,6 +67,7 @@ oslo.context==2.19.2
oslo.db==4.27.0
oslo.i18n==3.15.3
oslo.log==3.36.0
oslo.messaging==5.32.0
oslo.middleware==3.31.0
oslo.policy==1.30.0
oslo.rootwrap==5.8.0

1
requirements.txt
View File

@ -24,6 +24,7 @@ oslo.context>=2.19.2 # Apache-2.0
oslo.db>=4.27.0 # Apache-2.0
oslo.i18n>=3.15.3 # Apache-2.0
oslo.log>=3.36.0 # Apache-2.0
oslo.messaging>=5.32.0 # Apache-2.0
oslo.middleware>=3.31.0 # Apache-2.0
oslo.policy>=1.30.0 # Apache-2.0
oslo.rootwrap>=5.8.0 # Apache-2.0