neutron resource notifications




对neutron各种资源操作时,如果配置了oslo_messaging_notifications的dirver,如下所示:

[oslo_messaging_notifications]

#
# From oslo.messaging
#

# The Drivers(s) to handle sending notifications. Possible values are
# messaging, messagingv2, routing, log, test, noop (multi valued)
# Deprecated group/name - [DEFAULT]/notification_driver
driver = messagingv2

# A URL representing the messaging driver to use for notifications. If not set,
# we fall back to the same configuration used for RPC. (string value)
# Deprecated group/name - [DEFAULT]/notification_transport_url
#transport_url = <None>

# AMQP topic used for OpenStack notifications. (list value)
# Deprecated group/name - [rpc_notifier2]/topics
# Deprecated group/name - [DEFAULT]/notification_topics
topics = notifications

则会发送消息到Queue notifications.info,例如更新port属性:

[root@host-10-0-80-25 nova]# neutron port-update --name xxxx f8de4810-6213-4e27-8da2-98de8d20ae70

neutron-server日志可以看到(需要打开debug级别):

2018-01-26 13:51:34.138 32124 DEBUG oslo_messaging._drivers.amqpdriver [req-effb57d9-1f33-4034-8041-f2cb50332bd5 admin -] CAST u
nique_id: bc9df75c6ebe43bbb9f6589c9210617d NOTIFY exchange 'neutron' topic 'notifications.info' _send /usr/lib/python2.7/site-pa
ckages/oslo_messaging/_drivers/amqpdriver.py:438

消息队列中信息为:

{
	"oslo.message": {
		"_context_domain": null,
		"_context_roles": ["admin"],
		"event_type": "port.update.start",   ### start和end各一条
		"_context_request_id": "req-279b35a0-cb29-4c59-9dbd-6861fff446ce",
		"timestamp": "2018-01-26 06:40:01.553105",
		"_context_tenant_id": "b6767f5a386a4e519b15e02ad2fc8cd4",
		"_context_user": "60a9ac2fcdc74319aa7ef419874fe1e0",
		"_unique_id": "e356028dadb543119236e0425c4c551a",
		"_context_resource_uuid": null,
		"_context_tenant_name": "admin",
		"_context_user_id": "60a9ac2fcdc74319aa7ef419874fe1e0",
		"payload": {
			"port": {
				"name": "zzz"
			},
			"id": "f8de4810-6213-4e27-8da2-98de8d20ae70"
		},
		"_context_project_name": "admin",
		"_context_user_identity": "60a9ac2fcdc74319aa7ef419874fe1e0 b6767f5a386a4e519b15e02ad2fc8cd4 - - -",
		"_context_auth_token": "gAAAAABaas1AzGDQYyZ5lHnXYh3WbHe8y-WESX5nzozGYESjsOHCi25mQ0SVxeGxuhV6BKIx1aq569p6vZxn_QNzH2PLPifAN74h5ffWjKL0uq8U3eiHk_9CR9U-I_jwjSb9VHyyV2VGbuaBvQDld55wDB44wJYpixGdEENsqwWi74glc1iEl-A",
		"_context_show_deleted": false,
		"_context_tenant": "b6767f5a386a4e519b15e02ad2fc8cd4",
		"priority": "INFO",
		"_context_read_only": false,
		"_context_is_admin": true,
		"_context_project_id": "b6767f5a386a4e519b15e02ad2fc8cd4",
		"_context_project_domain": null,
		"_context_timestamp": "2018-01-26 06:40:01.551879",
		"_context_user_domain": null,
		"_context_user_name": "admin",
		"publisher_id": "network.host-10-0-80-25",
		"message_id": "8d4258a7-f982-4cde-b0e4-c2df87808ec9"
	},
	"oslo.version": "2.0"
}

加入断点调试,bt看调用栈如下:

(Pdb) bt
  /usr/lib/python2.7/site-packages/eventlet/greenpool.py(82)_spawn_n_impl()
-> func(*args, **kwargs)
  /usr/lib/python2.7/site-packages/eventlet/wsgi.py(719)process_request()
-> proto.__init__(sock, address, self)
  /usr/lib64/python2.7/SocketServer.py(649)__init__()
-> self.handle()
  /usr/lib64/python2.7/BaseHTTPServer.py(342)handle()
-> self.handle_one_request()
  /usr/lib/python2.7/site-packages/eventlet/wsgi.py(384)handle_one_request()
-> self.handle_one_response()
  /usr/lib/python2.7/site-packages/eventlet/wsgi.py(481)handle_one_response()
-> result = self.application(self.environ, start_response)
  /usr/lib/python2.7/site-packages/paste/urlmap.py(216)__call__()
-> return app(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(130)__call__()
-> resp = self.call_func(req, *args, **self.kwargs)
  /usr/lib/python2.7/site-packages/webob/dec.py(195)call_func()
-> return self.func(req, *args, **kwargs)
  /usr/lib/python2.7/site-packages/oslo_middleware/base.py(114)__call__()
-> response = req.get_response(self.application)
  /usr/lib/python2.7/site-packages/webob/request.py(1317)send()
-> application, catch_exc_info=False)
  /usr/lib/python2.7/site-packages/webob/request.py(1281)call_application()
-> app_iter = application(self.environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(130)__call__()
-> resp = self.call_func(req, *args, **self.kwargs)
  /usr/lib/python2.7/site-packages/webob/dec.py(195)call_func()
-> return self.func(req, *args, **kwargs)
  /usr/lib/python2.7/site-packages/oslo_middleware/request_id.py(37)__call__()
-> response = req.get_response(self.application)
  /usr/lib/python2.7/site-packages/webob/request.py(1317)send()
-> application, catch_exc_info=False)
  /usr/lib/python2.7/site-packages/webob/request.py(1281)call_application()
-> app_iter = application(self.environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(130)__call__()
-> resp = self.call_func(req, *args, **self.kwargs)
  /usr/lib/python2.7/site-packages/webob/dec.py(195)call_func()
-> return self.func(req, *args, **kwargs)
  /usr/lib/python2.7/site-packages/oslo_middleware/catch_errors.py(38)__call__()
-> response = req.get_response(self.application)
  /usr/lib/python2.7/site-packages/webob/request.py(1317)send()
-> application, catch_exc_info=False)
  /usr/lib/python2.7/site-packages/webob/request.py(1281)call_application()
-> app_iter = application(self.environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(130)__call__()
-> resp = self.call_func(req, *args, **self.kwargs)
  /usr/lib/python2.7/site-packages/webob/dec.py(195)call_func()
-> return self.func(req, *args, **kwargs)
  /usr/lib/python2.7/site-packages/keystonemiddleware/auth_token/__init__.py(467)__call__()
-> response = req.get_response(self._app)
  /usr/lib/python2.7/site-packages/webob/request.py(1317)send()
-> application, catch_exc_info=False)
  /usr/lib/python2.7/site-packages/webob/request.py(1281)call_application()
-> app_iter = application(self.environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(144)__call__()
-> return resp(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(144)__call__()
-> return resp(environ, start_response)
  /usr/lib/python2.7/site-packages/routes/middleware.py(136)__call__()
-> response = self.app(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(144)__call__()
-> return resp(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(144)__call__()
-> return resp(environ, start_response)
  /usr/lib/python2.7/site-packages/routes/middleware.py(136)__call__()
-> response = self.app(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(144)__call__()
-> return resp(environ, start_response)
  /usr/lib/python2.7/site-packages/webob/dec.py(130)__call__()
-> resp = self.call_func(req, *args, **self.kwargs)
  /usr/lib/python2.7/site-packages/webob/dec.py(195)call_func()
-> return self.func(req, *args, **kwargs)
  /opt/stack/neutron/neutron/api/v2/resource.py(84)resource()
-> result = method(request=request, **args)
  /opt/stack/neutron/neutron/api/v2/base.py(581)update()
-> payload)
  /usr/lib/python2.7/site-packages/oslo_messaging/notify/notifier.py(288)info()
-> self._notify(ctxt, event_type, payload, 'INFO')
  /usr/lib/python2.7/site-packages/oslo_messaging/notify/notifier.py(369)_notify()
-> super(_SubNotifier, self)._notify(ctxt, event_type, payload, priority)
  /usr/lib/python2.7/site-packages/oslo_messaging/notify/notifier.py(246)_notify()
-> self._driver_mgr.map(do_notify)
  /usr/lib/python2.7/site-packages/stevedore/extension.py(225)map()
-> self._invoke_one_plugin(response.append, func, e, args, kwds)
  /usr/lib/python2.7/site-packages/stevedore/extension.py(256)_invoke_one_plugin()
-> response_callback(func(e, *args, **kwds))
  /usr/lib/python2.7/site-packages/oslo_messaging/notify/notifier.py(239)do_notify()
-> ext.obj.notify(ctxt, msg, priority, retry or self.retry)
  /usr/lib/python2.7/site-packages/oslo_messaging/notify/messaging.py(49)notify()
-> retry=retry)
  /usr/lib/python2.7/site-packages/oslo_messaging/transport.py(97)_send_notification()
-> retry=retry)
  /usr/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.py(475)send_notification()
-> envelope=(version == 2.0), notify=True, retry=retry)
> /usr/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.py(440)_send()
-> conn.notify_send(exchange, target.topic, msg, retry=retry)

找到发送notification的位置:

    def update(self, request, id, body=None, **kwargs):
        """Updates the specified entity's attributes."""
        try:
            payload = body.copy()
        except AttributeError:
            msg = _("Invalid format: %s") % request.body
            raise exceptions.BadRequest(resource='body', msg=msg)
        payload['id'] = id
        self._notifier.info(request.context,
                            self._resource + '.update.start',  ### update.start消息
                            payload)
        return self._update(request, id, body, **kwargs)
    @db_api.retry_db_errors
    def _update(self, request, id, body, **kwargs):
        body = Controller.prepare_request_body(request.context,
                                               copy.deepcopy(body), False,
                                               self._resource, self._attr_info,
                                               allow_bulk=self._allow_bulk)
        action = self._plugin_handlers[self.UPDATE]
        ......
        result = {self._resource: self._view(request.context, obj)}
        notifier_method = self._resource + '.update.end'
        self._notifier.info(request.context, notifier_method, result)  ### update.end消息
        self._send_dhcp_notification(request.context,
                                     result,
                                     notifier_method)
        self._send_nova_notification(action, orig_object_copy, result)
        return result

上面的self._resource就是neutron的资源之一,如network、subnet、port、router等(也可以是各种extension资源如vpnaas、lbaas、fwaas等)。

发送notification使用的是oslo.messaging库,因为功能单一,所以文档比较简单,有需要可以考虑使用:https://docs.openstack.org/oslo.messaging/latest/