最近在学习neutron 安全组相关逻辑，所以梳理一下neutron关于安全组的具体代码实现，下面是neutron创建安全组时的代码，总体来说所有service的创建流程是一样的

1. 创建安全组

#neutron --debug security-group-create sg-debug-can-delete

curl -g -i -X POST http://10.x.x.x:9696/v2.0/security-groups -H "Accept: application/json" -H "Content-Type: application/json" -H "User-Agent: python-neutronclient" -H "X-Auth-Token: {SHA256}3655d2b6e755fbbb194d185fda9d339c642ab064dba48da834e1b2660e140123" -d '{"security_group": {"name": "sg-debug-can-delete"}}'

2. neutron/api/v2/base.py

安全组创建create请求由controller处理，调用create函数，self._notifier.info 这里先init 了neutron-lib库中的rpc.py中的NOTIFIER,然后调用oslo_message发送了一个info的通知 security_group.create.start, 调用_create 获取body以及action，加载policy ，obj_creator 最终是执行了 plugin neutron.plugins.ml2.plugin.Ml2Plugin 的函数 create_security_group，create_security_group来自实现的类SecurityGroupDbMixin中的，所以do_create中最终执行了SecurityGroupDbMixin的create_security_group函数，db中创建安全组obj，再notify，并同时将obj 安全组信息http 返回

 def create(self, request, body=None, **kwargs):self._notifier.info(request.context,self._resource + '.create.start',body)return self._create(request, body, **kwargs)@db_api.retry_db_errorsdef _create(self, request, body, **kwargs):"""Creates a new instance of the requested entity."""parent_id = kwargs.get(self._parent_id_name)body = Controller.prepare_request_body(request.context,body, True,self._resource, self._attr_info,allow_bulk=self._allow_bulk)action = self._plugin_handlers[self.CREATE]# Check authzif self._collection in body:# Have to account for bulk createitems = body[self._collection]else:items = [body]# Ensure policy engine is initializedpolicy.init()# Store requested resource amounts grouping them by tenant# This won't work with multiple resources. However because of the# current structure of this controller there will hardly be more than# one resource for which reservations are being maderequest_deltas = collections.defaultdict(int)for item in items:self._validate_network_tenant_ownership(request,item[self._resource])# For ext resources policy check, we support two types, such as# parent_id is in request body, another type is parent_id is in# request url, which we can get from kwargs.self._set_parent_id_into_ext_resources_request(request, item[self._resource], parent_id)policy.enforce(request.context,action,item[self._resource],pluralized=self._collection)if 'tenant_id' not in item[self._resource]:# no tenant_id - no quota checkcontinuetenant_id = item[self._resource]['tenant_id']request_deltas[tenant_id] += 1# Quota enforcementreservations = []try:for (tenant, delta) in request_deltas.items():reservation = quota.QUOTAS.make_reservation(request.context,tenant,{self._resource: delta},self._plugin)reservations.append(reservation)except exceptions.QuotaResourceUnknown as e:# We don't want to quota this resourceLOG.debug(e)def notify(create_result):# Ensure usage trackers for all resources affected by this API# operation are marked as dirtywith db_api.CONTEXT_WRITER.using(request.context):# Commit the reservation(s)for reservation in reservations:quota.QUOTAS.commit_reservation(request.context, reservation.reservation_id)resource_registry.set_resources_dirty(request.context)notifier_method = self._resource + '.create.end'self._notifier.info(request.context,notifier_method,create_result)registry.publish(self._resource, events.BEFORE_RESPONSE, self,payload=events.APIEventPayload(request.context, notifier_method, action,request_body=body,states=({}, create_result,),collection_name=self._collection))return create_resultdef do_create(body, bulk=False, emulated=False):kwargs = {self._parent_id_name: parent_id} if parent_id else {}if bulk and not emulated:obj_creator = getattr(self._plugin, "%s_bulk" % action)else:#这里获取了ML2plugin的create_security_group属性obj_creator = getattr(self._plugin, action)try:if emulated:return self._emulate_bulk_create(obj_creator, request,body, parent_id)else:if self._collection in body:# This is weird but fixing it requires changes to the# plugin interfacekwargs.update({self._collection: body})else:kwargs.update({self._resource: body})#这里调用了create_security_group函数return obj_creator(request.context, **kwargs)except Exception:# In case of failure the plugin will always raise an# exception. Cancel the reservationwith excutils.save_and_reraise_exception():for reservation in reservations:quota.QUOTAS.cancel_reservation(request.context, reservation.reservation_id)if self._collection in body and self._native_bulk:# plugin does atomic bulk create operationsobjs = do_create(body, bulk=True)# Use first element of list to discriminate attributes which# should be removed because of authZ policiesfields_to_strip = self._exclude_attributes_by_policy(request.context, objs[0])return notify({self._collection: [self._filter_attributes(obj, fields_to_strip=fields_to_strip)for obj in objs]})else:if self._collection in body:# Emulate atomic bulk behaviorobjs = do_create(body, bulk=True, emulated=True)return notify({self._collection: objs})else:#走到这里创建了db记录,notify并 http 返回信息obj = do_create(body)return notify({self._resource: self._view(request.context,obj)})

3. neutron/db/securitygroups_db.py

db中创建安全组以及安全组egress规则， 创建完后 AFTER_CREATE，通过notify发送rpc通知, 执行 neutron.plugins.ml2.ovo_rpc._ObjectChangeHandler.handle_event–9223372036799819083’

 @db_api.retry_if_session_inactive()def create_security_group(self, context, security_group, default_sg=False):"""Create security group.If default_sg is true that means we are a default security group fora given tenant if it does not exist."""s = security_group['security_group']kwargs = {'context': context,'security_group': s,'is_default': default_sg,}#创建前rpc通知订阅serviceself._registry_notify(resources.SECURITY_GROUP, events.BEFORE_CREATE,exc_cls=ext_sg.SecurityGroupConflict,payload=events.DBEventPayload(context, metadata={'is_default': default_sg},request_body=security_group,desired_state=s))tenant_id = s['tenant_id']with db_api.CONTEXT_WRITER.using(context):sg = sg_obj.SecurityGroup(context, id=s.get('id') or uuidutils.generate_uuid(),description=s['description'], project_id=tenant_id,name=s['name'], is_default=default_sg)sg.create()for ethertype in ext_sg.sg_supported_ethertypes:egress_rule = sg_obj.SecurityGroupRule(context, id=uuidutils.generate_uuid(),project_id=tenant_id, security_group_id=sg.id,direction='egress', ethertype=ethertype)egress_rule.create()sg.rules.append(egress_rule)sg.obj_reset_changes(['rules'])# fetch sg from db to load the sg rules with sg model.sg = sg_obj.SecurityGroup.get_object(context, id=sg.id)secgroup_dict = self._make_security_group_dict(sg)kwargs['security_group'] = secgroup_dictself._registry_notify(resources.SECURITY_GROUP,events.PRECOMMIT_CREATE,exc_cls=ext_sg.SecurityGroupConflict,**kwargs)registry.notify(resources.SECURITY_GROUP, events.AFTER_CREATE, self,**kwargs)return secgroup_dict

4. neutron_lib/callbacks/manager.py

   因为event是events.BEFORE_RESPONSE，所以最终执行了self._notify_loop，最终执行了callback

   @db_utils.reraise_as_retryrequestdef notify(self, resource, event, trigger, **kwargs):"""Notify all subscribed callback(s).Dispatch the resource's event to the subscribed callbacks.:param resource: The resource for the event.:param event: The event.:param trigger: The trigger. A reference to the sender of the event.:param kwargs: (deprecated) Unstructured key/value pairs to invokethe callback with. Using event objects with publish() is preferred.:raises CallbackFailure: CallbackFailure is raised if the underlyingcallback has errors."""errors = self._notify_loop(resource, event, trigger, **kwargs)if errors:if event.startswith(events.BEFORE):abort_event = event.replace(events.BEFORE, events.ABORT)self._notify_loop(resource, abort_event, trigger, **kwargs)raise exceptions.CallbackFailure(errors=errors)if event.startswith(events.PRECOMMIT):raise exceptions.CallbackFailure(errors=errors)def clear(self):"""Brings the manager to a clean slate."""self._callbacks = collections.defaultdict(dict)self._index = collections.defaultdict(dict)def _notify_loop(self, resource, event, trigger, **kwargs):"""The notification loop."""errors = []# NOTE(yamahata): Since callback may unsubscribe it,# convert iterator to list to avoid runtime error.callbacks = list(itertools.chain(*[pri_callbacks.items() for (priority, pri_callbacks)in self._callbacks[resource].get(event, [])]))LOG.debug("Notify callbacks %s for %s, %s",[c[0] for c in callbacks], resource, event)# TODO(armax): consider using a GreenPilefor callback_id, callback in callbacks:try:callback(resource, event, trigger, **kwargs)except Exception as e:abortable_event = (event.startswith(events.BEFORE) orevent.startswith(events.PRECOMMIT))if not abortable_event:LOG.exception("Error during notification for ""%(callback)s %(resource)s, %(event)s",{'callback': callback_id,'resource': resource, 'event': event})else:LOG.debug("Callback %(callback)s raised %(error)s",{'callback': callback_id, 'error': e})errors.append(exceptions.NotificationError(callback_id, e))return errors
   

5. neutron/plugins/ml2/ovo_rpc.py

ObjectChangeHandler初始化时subscribe了AFTER_CREATE，AFTER_UPDATE,AFTER_DELETE的event，最后启用了一个线程将event分发下去

     def handle_event(self, resource, event, trigger,context, *args, **kwargs):"""Callback handler for resource change that pushes change to RPC.We always retrieve the latest state and ignore what was in thepayload to ensure that we don't get any stale data."""if self._is_session_semantic_violated(context, resource, event):returnresource_id = self._extract_resource_id(kwargs)# we preserve the context so we can trace a receive on the agent back# to the server-side event that triggered itself._resources_to_push[resource_id] = context.to_dict()# spawn worker so we don't block main AFTER_UPDATE threadself.fts.append(self._worker_pool.submit(self.dispatch_events))@lockutils.synchronized('event-dispatch')def dispatch_events(self):# this is guarded by a lock to ensure we don't get too many concurrent# dispatchers hitting the database simultaneously.to_dispatch, self._resources_to_push = self._resources_to_push, {}# TODO(kevinbenton): now that we are batching these, convert to a# single get_objects call for all of themfor resource_id, context_dict in to_dispatch.items():context = n_ctx.Context.from_dict(context_dict)# attempt to get regardless of event type so concurrent delete# after create/update is the same code-path as a delete eventwith db_api.get_context_manager().independent.reader.using(context):obj = self._obj_class.get_object(context, id=resource_id)# CREATE events are always treated as UPDATE events to ensure# listeners are written to handle out-of-order messagesif obj is None:rpc_event = rpc_events.DELETED# construct a fake object with the right ID so we can# have a payload for the delete message.obj = self._obj_class(id=resource_id)else:rpc_event = rpc_events.UPDATEDself._resource_push_api.push(context, [obj], rpc_event)

6. neutron/api/rpc/handlers/resources_rpc.py

最终发送cast广播出去了

class ResourcesPushRpcApi(object): def push(self, context, resource_list, event_type):"""Push an event and list of resources to agents, batched per type.When a list of different resource types is passed to this method,the push will be sent as separate individual list pushes, one perresource type."""resources_by_type = self._classify_resources_by_type(resource_list)LOG.debug("Pushing event %s for resources: %s", event_type,{t: ["ID=%s,revision_number=%s" % (getattr(obj, 'id', None),getattr(obj, 'revision_number', None))for obj in resources_by_type[t]]for t in resources_by_type})for resource_type, type_resources in resources_by_type.items():self._push(context, resource_type, type_resources, event_type)def _push(self, context, resource_type, resource_list, event_type):"""Push an event and list of resources of the same type to agents."""_validate_resource_type(resource_type)for version in version_manager.get_resource_versions(resource_type):cctxt = self._prepare_object_fanout_context(resource_list[0], version, rpc_version='1.1')dehydrated_resources = [resource.obj_to_primitive(target_version=version)for resource in resource_list]cctxt.cast(context, 'push',resource_list=dehydrated_resources,event_type=event_type)

日志：

Pushing event updated for resources: {'SecurityGroup': ['ID=47991532-81e1-4454-a010-7d1d45c07db1,revision_number=1']} push /var/lib/kolla/venv/lib/python2.7/site-packages/neutron/api/rpc/handlers/resources_rpc.py:243CAST unique_id: 0d594a5224974f5aa28788fb340e3c90 FANOUT topic 'neutron-vo-SecurityGroup-1.1' _send /var/lib/kolla/venv/lib/python2.7/site-packages/oslo_messaging/_drivers/amqpdriver.py:617Exchange neutron-vo-SecurityGroup-1.1_fanout(fanout) with routing key None