旅游景区网站建设策划书,淄博亿泰网站建设推广,互联网保险的弊端,rp网站自动跳转图片怎么做最近在学习neutron 安全组相关逻辑#xff0c;所以梳理一下neutron关于安全组的具体代码实现#xff0c;下面是neutron创建安全组时的代码#xff0c;总体来说所有service的创建流程是一样的
创建安全组
#neutron --debug security-group-create sg-debug-can-deletecurl …最近在学习neutron 安全组相关逻辑所以梳理一下neutron关于安全组的具体代码实现下面是neutron创建安全组时的代码总体来说所有service的创建流程是一样的
创建安全组
#neutron --debug security-group-create sg-debug-can-deletecurl -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}}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_groupcreate_security_group来自实现的类SecurityGroupDbMixin中的所以do_create中最终执行了SecurityGroupDbMixin的create_security_group函数db中创建安全组obj再notify并同时将obj 安全组信息http 返回 def create(self, request, bodyNone, **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_bulkself._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 wont 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],pluralizedself._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 dont 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.endself._notifier.info(request.context,notifier_method,create_result)registry.publish(self._resource, events.BEFORE_RESPONSE, self,payloadevents.APIEventPayload(request.context, notifier_method, action,request_bodybody,states({}, create_result,),collection_nameself._collection))return create_resultdef do_create(body, bulkFalse, emulatedFalse):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, bulkTrue)# 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_stripfields_to_strip)for obj in objs]})else:if self._collection in body:# Emulate atomic bulk behaviorobjs do_create(body, bulkTrue, emulatedTrue)return notify({self._collection: objs})else:#走到这里创建了db记录,notify并 http 返回信息obj do_create(body)return notify({self._resource: self._view(request.context,obj)})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_sgFalse):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_clsext_sg.SecurityGroupConflict,payloadevents.DBEventPayload(context, metadata{is_default: default_sg},request_bodysecurity_group,desired_states))tenant_id s[tenant_id]with db_api.CONTEXT_WRITER.using(context):sg sg_obj.SecurityGroup(context, ids.get(id) or uuidutils.generate_uuid(),descriptions[description], project_idtenant_id,names[name], is_defaultdefault_sg)sg.create()for ethertype in ext_sg.sg_supported_ethertypes:egress_rule sg_obj.SecurityGroupRule(context, iduuidutils.generate_uuid(),project_idtenant_id, security_group_idsg.id,directionegress, ethertypeethertype)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, idsg.id)secgroup_dict self._make_security_group_dict(sg)kwargs[security_group] secgroup_dictself._registry_notify(resources.SECURITY_GROUP,events.PRECOMMIT_CREATE,exc_clsext_sg.SecurityGroupConflict,**kwargs)registry.notify(resources.SECURITY_GROUP, events.AFTER_CREATE, self,**kwargs)return secgroup_dictneutron_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 resources 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(errorserrors)if event.startswith(events.PRECOMMIT):raise exceptions.CallbackFailure(errorserrors)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 errorsneutron/plugins/ml2/ovo_rpc.py
ObjectChangeHandler初始化时subscribe了AFTER_CREATEAFTER_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 dont 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 dont 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 dont 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, idresource_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(idresource_id)else:rpc_event rpc_events.UPDATEDself._resource_push_api.push(context, [obj], rpc_event)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_version1.1)dehydrated_resources [resource.obj_to_primitive(target_versionversion)for resource in resource_list]cctxt.cast(context, push,resource_listdehydrated_resources,event_typeevent_type)日志
Pushing event updated for resources: {SecurityGroup: [ID47991532-81e1-4454-a010-7d1d45c07db1,revision_number1]} 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
