AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

OpenStack Swift源码初探--proxy下的server.py

2016-06-07 2759
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
日志服务 SLS，月写入数据量 50GB 1个月
简介：

          OpenStack Swift作为开源的云存储工具，被越来越多的公司使用。为了记录和巩固学习swift的开源源码，所以进行一系列的源码开源学习笔记，供初学者快速学习和理解swift的内部功能。

          proxy下面的server.py模块是所有对account，container，object等对象进行管理操作的在swift的proxy端的总入口。在swift系统在接收到url请求后，先是经过middleware处理链分别验证处理后，再进入到proxy下面的server模块，进入 _call_方法调用后，把对应的请求分发给不同的controller处理，controller再调用各自的nodeStroage服务程序进行处理，返回各自的resp结果到server模块，最后通过mimmdleware处理链再反方向返回最终的请求处理结果

 

     1.首先server.py的整个结构如下图：包括4部分：一堆import引用，一个required_filters的字典，一个名为“Application(object)”的class ，一个app_factory(global_conf, **local_conf) 方法

   

     2.主要介绍“Application(object)”的class，其中包含了所有主要的功能方法

   

 

 2.1 _init_ 方法，Application类的初始化方法，主要就是初始化一些对象，包括：conf配置文件参数 的初始化，log日志初始化，memcache对象初始化，account_ring,container_ring, object_ring对象初始化等

 

2.2  check_config(self) 方法，主要是检查配置文件proxy-server.conf中配置的“read_affinity” 和“sorting_method”属性值是否正确，该方法在 app_factory(global_conf, **local_conf):方法时调用  

   

2.3 get_controller(self, path)方法，主要是根据传入的 urlPath解析并返回对应的control类和一个字典对象，其中字典对象的值根据传入url格式的不同返回不同的值

 

[python]  view plain copy
  1. def get_controller(self, path):  
  2.         """ 
  3.         Get the controller to handle a request. 
  4.  
  5.         :param path: path from request 
  6.         :returns: tuple of (controller class, path dictionary) 
  7.  
  8.         :raises: ValueError (thrown by split_path) if given invalid path 
  9.         """  
  10.         if path == '/info':  #url是/info 则返回InfoController和包括version,expose_info,disallowed_sections,admin_key的字典  
  11.             d = dict(version=None,  
  12.                      expose_info=self.expose_info,  
  13.                      disallowed_sections=self.disallowed_sections,  
  14.                      admin_key=self.admin_key)  
  15.             return InfoController, d  
  16.   
  17.         version, account, container, obj = split_path(path, 1, 4, True)  #以/拆分url为数列，并取对应的1到4位的数据返回给对应的变量  
  18.         d = dict(version=version,  
  19.                  account_name=account,  
  20.                  container_name=container,  
  21.                  object_name=obj)  
  22.         if obj and container and account: #根据解析出的account值，congtainer和object值得有无，确定适用的Controller是那种  
  23.             return ObjectController, d  
  24.         elif container and account:  
  25.             return ContainerController, d  
  26.         elif account and not container and not obj:  
  27.             return AccountController, d  
  28.         return None, d  


 

 

2.4  __call__(self, env, start_response)方法，是server模块的实际对account、container、object等对象调用处理的功能入口。

 

[python]  view plain copy
  1. def __call__(self, env, start_response):  
  2.         """ 
  3.         WSGI entry point. 
  4.         Wraps env in swob.Request object and passes it down. 
  5.  
  6.         :param env: WSGI environment dictionary 
  7.         :param start_response: WSGI callable 
  8.         """  
  9.         try:  
  10.             if self.memcache is None: #首先判断是否memcache值存在，不存在再去获取一次  
  11.                 self.memcache = cache_from_env(env)  
  12.             req = self.update_request(Request(env)) #判断header中是否有x-storage-token和x-auth-token  
  13.             return self.handle_request(req)(env, start_response) #调用handle_request方法并返回处理的结果resp对象  
  14.         except UnicodeError:  
  15.             err = HTTPPreconditionFailed(  
  16.                 request=req, body='Invalid UTF8 or contains NULL')  
  17.             return err(env, start_response)  
  18.         except (Exception, Timeout):  
  19.             start_response('500 Server Error',  
  20.                            [('Content-Type', 'text/plain')])  
  21.             return ['Internal server error.\n']  


2.5 update_request(self, req)方法，根据请求中header里面的x-storage-token有而x-auth-token没有的情况，把x-storage-token的值赋予x-auth-token

 


2.6 handle_request(self, req)方法，server模块实际处理request请求的方法，熟悉servlet的同学可以把它理解成servlet的作用

 

[python]  view plain copy
  1. def handle_request(self, req):  
  2.         """ 
  3.         Entry point for proxy server. 
  4.         Should return a WSGI-style callable (such as swob.Response). 
  5.  
  6.         :param req: swob.Request object 
  7.         """  
  8.         try:  
  9.             self.logger.set_statsd_prefix('proxy-server') #在日志的开头加上‘proxy-server’,方便跟踪分析  
  10.             if req.content_length and req.content_length < 0: #检查header里面中的Content-Length是否有值，无值返回错误请求，并日志记录  
  11.                 self.logger.increment('errors')  
  12.                 return HTTPBadRequest(request=req,  
  13.                                       body='Invalid Content-Length')  
  14.   
  15.             try:  
  16.                 if not check_utf8(req.path_info): #检查Pathde的编码是否不满足utf8，不满足返回错误请求，并日志记录  
  17.                     self.logger.increment('errors')  
  18.                     return HTTPPreconditionFailed(  
  19.                         request=req, body='Invalid UTF8 or contains NULL')  
  20.             except UnicodeError:  
  21.                 self.logger.increment('errors')  
  22.                 return HTTPPreconditionFailed(  
  23.                     request=req, body='Invalid UTF8 or contains NULL')  
  24.   
  25.             try:  
  26.                 controller, path_parts = self.get_controller(req.path) #调用get_controller(self,path)方法返回正确的controller类和字典对象  
  27.                 p = req.path_info  
  28.                 if isinstance(p, unicode):  
  29.                     p = p.encode('utf-8') #path编码Unicode转换utf-8  
  30.             except ValueError:           #发生值异常，返回错误请求，并日志记录  
  31.                 self.logger.increment('errors')  
  32.                 return HTTPNotFound(request=req)  
  33.             if not controller:         #为找到对应处理的controller类时，返回错误请求，并日志记录  
  34.                 self.logger.increment('errors')  
  35.                 return HTTPPreconditionFailed(request=req, body='Bad URL')  
  36.             if self.deny_host_headers and \    #当proxy-server.conf中deny_host_headers有值，且请求的header中的host在deny_host_headers中，则返回错误请求，并日志记录  
  37.                     req.host.split(':')[0] in self.deny_host_headers:  
  38.                 return HTTPForbidden(request=req, body='Invalid host header')  
  39.   
  40.             self.logger.set_statsd_prefix('proxy-server.' +  
  41.                                           controller.server_type.lower()) #在日志的开头加上‘proxy-server.controller类中的请求类型（eg：HEAD/GET/PUT）’,方便跟踪分析  
  42.             controller = controller(self, **path_parts)  #初始化实际的controller对象（AccountController、ContainerController、ObjectController、InfoController其中之一）  
  43.             if 'swift.trans_id' not in req.environ:    #如果没有trans_id在env中，则重新生成一个，有些类似于http请求中的seesionID的感觉，是一种UUID  
  44.                 # if this wasn't set by an earlier middleware, set it now  
  45.                 trans_id = generate_trans_id(self.trans_id_suffix)  
  46.                 req.environ['swift.trans_id'] = trans_id  
  47.                 self.logger.txn_id = trans_id  
  48.             req.headers['x-trans-id'] = req.environ['swift.trans_id']  
  49.             controller.trans_id = req.environ['swift.trans_id']  
  50.             self.logger.client_ip = get_remote_client(req)  #把请求中获取出请求端的IP信息，加入logger对象，方便后续日志查看分析  
  51.             try:  
  52.                 handler = getattr(controller, req.method) #根据req.method方法获取对应controller对象中的方法（可能是多个，有的有public标签，有的没有）  
  53.                 getattr(handler, 'publicly_accessible') #再根据public标签获取最终的处理方法。（在方法前面可以加 @public 和@delay_denial）  
  54.             except AttributeError:  
  55.                 allowed_methods = getattr(controller, 'allowed_methods', set())  
  56.                 return HTTPMethodNotAllowed(  
  57.                     request=req, headers={'Allow': ', '.join(allowed_methods)})  
  58.             if 'swift.authorize' in req.environ: #做鉴权操作  
  59.                 # We call authorize before the handler, always. If authorized,  
  60.                 # we remove the swift.authorize hook so isn't ever called  
  61.                 # again. If not authorized, we return the denial unless the  
  62.                 # controller's method indicates it'd like to gather more  
  63.                 # information and try again later.  
  64.                 resp = req.environ['swift.authorize'](req)  
  65.                 if not resp:  
  66.                     # No resp means authorized, no delayed recheck required.  
  67.                     del req.environ['swift.authorize']  
  68.                 else:  
  69.                     # Response indicates denial, but we might delay the denial  
  70.                     # and recheck later. If not delayed, return the error now.  
  71.                     if not getattr(handler, 'delay_denial', None):  
  72.                         return resp  
  73.             # Save off original request method (GET, POST, etc.) in case it  
  74.             # gets mutated during handling.  This way logging can display the  
  75.             # method the client actually sent.  
  76.             req.environ['swift.orig_req_method'] = req.method  
  77.             return handler(req)    #调用最终的method方法，并返回resp结果  
  78.         except HTTPException as error_response:  
  79.             return error_response  
  80.         except (Exception, Timeout):  
  81.             self.logger.exception(_('ERROR Unhandled exception in request'))  
  82.             return HTTPServerError(request=req)  

 

2.7 sort_nodes(self, nodes)方法，对nodes对象进行排序处理，该方法在iter_nodes(self, ring, partition, node_iter=None)中调用

 

[python]  view plain copy
  1. def sort_nodes(self, nodes):  
  2.         ''''' 
  3.         Sorts nodes in-place (and returns the sorted list) according to 
  4.         the configured strategy. The default "sorting" is to randomly 
  5.         shuffle the nodes. If the "timing" strategy is chosen, the nodes 
  6.         are sorted according to the stored timing data. 
  7.         '''  
  8.         # In the case of timing sorting, shuffling ensures that close timings  
  9.         # (ie within the rounding resolution) won't prefer one over another.  
  10.         # Python's sort is stable (http://wiki.python.org/moin/HowTo/Sorting/)  
  11.         shuffle(nodes)  
  12.         if self.sorting_method == 'timing':  #配置文件中排序方法为timing时，以时间排序  
  13.             now = time()  
  14.   
  15.             def key_func(node):  
  16.                 timing, expires = self.node_timings.get(node['ip'], (-1.0, 0))  
  17.                 return timing if expires > now else -1.0  
  18.             nodes.sort(key=key_func)  
  19.         elif self.sorting_method == 'affinity': #配置文件中排序方法为affinity时，以自定义的亲和力规则排序  
  20.             nodes.sort(key=self.read_affinity_sort_key)  
  21.         return nodes  

 

2.8 set_node_timing(self, node, timing)方法，提供给外部程序调用

 

2.9    error_limited(self, node)方法，该方法在iter_nodes(self, ring, partition, node_iter=None)中调用

 

[python]  view plain copy
  1. def error_limited(self, node):  
  2.         """ 
  3.         Check if the node is currently error limited. 
  4.  
  5.         :param node: dictionary of node to check 
  6.         :returns: True if error limited, False otherwise 
  7.         """  
  8.         now = time()  
  9.         if 'errors' not in node:  #errors没在node中时返回false  
  10.             return False  
  11.         if 'last_error' in node and node['last_error'] < \   #last_error在node中有并且 last_error小于现在时间减去系统允许的时间间隔  
  12.                 now - self.error_suppression_interval:  
  13.             del node['last_error'] #node去掉last_error  
  14.             if 'errors' in node: #errors在node中时返回 去掉errors，且返回false  
  15.                 del node['errors']  
  16.             return False  
  17.         limited = node['errors'] > self.error_suppression_limit  #errors在node中中的个数多与系统允许的个数，是返回true，且做日志记录  
  18.         if limited:  
  19.             self.logger.debug(  
  20.                 _('Node error limited %(ip)s:%(port)s (%(device)s)'), node)  
  21.         return limited  

 

 

2.10  error_limit(self, node, msg)方法，提供给外部程序调用，用于给node直接增加errors到比系统允许的次数+1，并记录last_error时间，和做日志记录

 

2.11  error_limit(self, node, msg)方法，提供给外部程序调用，用于给node增加errors次数，并记录last_error时间，和做日志记录

 

2.12  iter_nodes(self, ring, partition, node_iter=None)方法，提供给外部程序调用，用于对nodes做排序后生成的nodes迭代器

 

2.13  exception_occurred(self, node, typ, additional_info)方法，提供给外部程序调用，用于当node发生异常了，进行日志记录

2.14  modify_wsgi_pipeline(self, pipe)方法，提供给外部程序调用，用于系统启动时，初始化pipeline，并做日志记录

 

[python]  view plain copy
  1. def modify_wsgi_pipeline(self, pipe):  
  2.         """ 
  3.         Called during WSGI pipeline creation. Modifies the WSGI pipeline 
  4.         context to ensure that mandatory middleware is present in the pipeline. 
  5.  
  6.         :param pipe: A PipelineWrapper object 
  7.         """  
  8.         pipeline_was_modified = False  
  9.         for filter_spec in reversed(required_filters):  #当required_filters字典中定义了需要重新排序的app时，进行pipeline的重新排序处理  
  10.             filter_name = filter_spec['name']  
  11.             if filter_name not in pipe:  
  12.                 afters = filter_spec.get('after_fn', lambda _junk: [])(pipe)  
  13.                 insert_at = 0  
  14.                 for after in afters:  
  15.                     try:  
  16.                         insert_at = max(insert_at, pipe.index(after) + 1)  
  17.                     except ValueError:  # not in pipeline; ignore it  
  18.                         pass  
  19.                 self.logger.info(  
  20.                     'Adding required filter %s to pipeline at position %d' %  
  21.                     (filter_name, insert_at))  
  22.                 ctx = pipe.create_filter(filter_name)  
  23.                 pipe.insert_filter(ctx, index=insert_at)  
  24.                 pipeline_was_modified = True  
  25.   
  26.         if pipeline_was_modified:    
  27.             self.logger.info("Pipeline was modified. New pipeline is \"%s\".",  
  28.                              pipe)  
  29.         else:  
  30.             self.logger.debug("Pipeline is \"%s\"", pipe)  


 

以下源码为2014、3、12的最新的Proxy的server.py源码,只加了部分代码注释：

 

[python]  view plain copy
  1. # Copyright (c) 2010-2012 OpenStack Foundation  
  2. #  
  3. # Licensed under the Apache License, Version 2.0 (the "License");  
  4. # you may not use this file except in compliance with the License.  
  5. # You may obtain a copy of the License at  
  6. #  
  7. #    http://www.apache.org/licenses/LICENSE-2.0  
  8. #  
  9. # Unless required by applicable law or agreed to in writing, software  
  10. # distributed under the License is distributed on an "AS IS" BASIS,  
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or  
  12. # implied.  
  13. # See the License for the specific language governing permissions and  
  14. # limitations under the License.  
  15.   
  16. import mimetypes  
  17. import os  
  18. import socket  
  19. from swift import gettext_ as _  
  20. from random import shuffle  
  21. from time import time  
  22. import itertools  
  23.   
  24. from eventlet import Timeout  
  25.   
  26. from swift import __canonical_version__ as swift_version  
  27. from swift.common import constraints  
  28. from swift.common.ring import Ring  
  29. from swift.common.utils import cache_from_env, get_logger, \  
  30.     get_remote_client, split_path, config_true_value, generate_trans_id, \  
  31.     affinity_key_function, affinity_locality_predicate, list_from_csv, \  
  32.     register_swift_info  
  33. from swift.common.constraints import check_utf8  
  34. from swift.proxy.controllers import AccountController, ObjectController, \  
  35.     ContainerController, InfoController  
  36. from swift.common.swob import HTTPBadRequest, HTTPForbidden, \  
  37.     HTTPMethodNotAllowed, HTTPNotFound, HTTPPreconditionFailed, \  
  38.     HTTPServerError, HTTPException, Request  
  39.   
  40.   
  41. # List of entry points for mandatory middlewares.  
  42. #  
  43. # Fields:  
  44. #  
  45. # "name" (required) is the entry point name from setup.py.  
  46. #  
  47. # "after_fn" (optional) a function that takes a PipelineWrapper object as its  
  48. # single argument and returns a list of middlewares that this middleware  
  49. # should come after. Any middlewares in the returned list that are not present  
  50. # in the pipeline will be ignored, so you can safely name optional middlewares  
  51. # to come after. For example, ["catch_errors", "bulk"] would install this  
  52. # middleware after catch_errors and bulk if both were present, but if bulk  
  53. # were absent, would just install it after catch_errors.  
  54.   
  55. required_filters = [  
  56.     {'name': 'catch_errors'},  
  57.     {'name': 'gatekeeper',  
  58.      'after_fn': lambda pipe: (['catch_errors']  
  59.                                if pipe.startswith("catch_errors")  
  60.                                else [])},  
  61.     {'name': 'dlo', 'after_fn': lambda _junk: ['catch_errors', 'gatekeeper',  
  62.                                                'proxy_logging']}]  
  63.   
  64.   
  65. class Application(object):  
  66.     """WSGI application for the proxy server."""  
  67.   
  68.     def __init__(self, conf, memcache=None, logger=None, account_ring=None,  
  69.                  container_ring=None, object_ring=None):  
  70.         if conf is None:  
  71.             conf = {}  
  72.         if logger is None:  
  73.             self.logger = get_logger(conf, log_route='proxy-server')  
  74.         else:  
  75.             self.logger = logger  
  76.   
  77.         swift_dir = conf.get('swift_dir', '/etc/swift')  
  78.         self.node_timeout = int(conf.get('node_timeout', 10))  
  79.         self.recoverable_node_timeout = int(  
  80.             conf.get('recoverable_node_timeout', self.node_timeout))  
  81.         self.conn_timeout = float(conf.get('conn_timeout', 0.5))  
  82.         self.client_timeout = int(conf.get('client_timeout', 60))  
  83.         self.put_queue_depth = int(conf.get('put_queue_depth', 10))  
  84.         self.object_chunk_size = int(conf.get('object_chunk_size', 65536))  
  85.         self.client_chunk_size = int(conf.get('client_chunk_size', 65536))  
  86.         self.trans_id_suffix = conf.get('trans_id_suffix', '')  
  87.         self.post_quorum_timeout = float(conf.get('post_quorum_timeout', 0.5))  
  88.         self.error_suppression_interval = \  
  89.             int(conf.get('error_suppression_interval', 60))  
  90.         self.error_suppression_limit = \  
  91.             int(conf.get('error_suppression_limit', 10))  
  92.         self.recheck_container_existence = \  
  93.             int(conf.get('recheck_container_existence', 60))  
  94.         self.recheck_account_existence = \  
  95.             int(conf.get('recheck_account_existence', 60))  
  96.         self.allow_account_management = \  
  97.             config_true_value(conf.get('allow_account_management', 'no'))  
  98.         self.object_post_as_copy = \  
  99.             config_true_value(conf.get('object_post_as_copy', 'true'))  
  100.         self.object_ring = object_ring or Ring(swift_dir, ring_name='object')  
  101.         self.container_ring = container_ring or Ring(swift_dir,  
  102.                                                      ring_name='container')  
  103.         self.account_ring = account_ring or Ring(swift_dir,  
  104.                                                  ring_name='account')  
  105.         self.memcache = memcache  
  106.         mimetypes.init(mimetypes.knownfiles +  
  107.                        [os.path.join(swift_dir, 'mime.types')])  
  108.         self.account_autocreate = \  
  109.             config_true_value(conf.get('account_autocreate', 'no'))  
  110.         self.expiring_objects_account = \  
  111.             (conf.get('auto_create_account_prefix') or '.') + \  
  112.             (conf.get('expiring_objects_account_name') or 'expiring_objects')  
  113.         self.expiring_objects_container_divisor = \  
  114.             int(conf.get('expiring_objects_container_divisor') or 86400)  
  115.         self.max_containers_per_account = \  
  116.             int(conf.get('max_containers_per_account') or 0)  
  117.         self.max_containers_whitelist = [  
  118.             a.strip()  
  119.             for a in conf.get('max_containers_whitelist', '').split(',')  
  120.             if a.strip()]  
  121.         self.deny_host_headers = [  
  122.             host.strip() for host in  
  123.             conf.get('deny_host_headers', '').split(',') if host.strip()]  
  124.         self.rate_limit_after_segment = \  
  125.             int(conf.get('rate_limit_after_segment', 10))  
  126.         self.rate_limit_segments_per_sec = \  
  127.             int(conf.get('rate_limit_segments_per_sec', 1))  
  128.         self.log_handoffs = config_true_value(conf.get('log_handoffs', 'true'))  
  129.         self.cors_allow_origin = [  
  130.             a.strip()  
  131.             for a in conf.get('cors_allow_origin', '').split(',')  
  132.             if a.strip()]  
  133.         self.strict_cors_mode = config_true_value(  
  134.             conf.get('strict_cors_mode', 't'))  
  135.         self.node_timings = {}  
  136.         self.timing_expiry = int(conf.get('timing_expiry', 300))  
  137.         self.sorting_method = conf.get('sorting_method', 'shuffle').lower()  
  138.         self.max_large_object_get_time = float(  
  139.             conf.get('max_large_object_get_time', '86400'))  
  140.         value = conf.get('request_node_count', '2 * replicas').lower().split()  
  141.         if len(value) == 1:  
  142.             value = int(value[0])  
  143.             self.request_node_count = lambda replicas: value  
  144.         elif len(value) == 3 and value[1] == '*' and value[2] == 'replicas':  
  145.             value = int(value[0])  
  146.             self.request_node_count = lambda replicas: value * replicas  
  147.         else:  
  148.             raise ValueError(  
  149.                 'Invalid request_node_count value: %r' % ''.join(value))  
  150.         try:  
  151.             self._read_affinity = read_affinity = conf.get('read_affinity', '')  
  152.             self.read_affinity_sort_key = affinity_key_function(read_affinity)  
  153.         except ValueError as err:  
  154.             # make the message a little more useful  
  155.             raise ValueError("Invalid read_affinity value: %r (%s)" %  
  156.                              (read_affinity, err.message))  
  157.         try:  
  158.             write_affinity = conf.get('write_affinity', '')  
  159.             self.write_affinity_is_local_fn \  
  160.                 = affinity_locality_predicate(write_affinity)  
  161.         except ValueError as err:  
  162.             # make the message a little more useful  
  163.             raise ValueError("Invalid write_affinity value: %r (%s)" %  
  164.                              (write_affinity, err.message))  
  165.         value = conf.get('write_affinity_node_count',  
  166.                          '2 * replicas').lower().split()  
  167.         if len(value) == 1:  
  168.             value = int(value[0])  
  169.             self.write_affinity_node_count = lambda replicas: value  
  170.         elif len(value) == 3 and value[1] == '*' and value[2] == 'replicas':  
  171.             value = int(value[0])  
  172.             self.write_affinity_node_count = lambda replicas: value * replicas  
  173.         else:  
  174.             raise ValueError(  
  175.                 'Invalid write_affinity_node_count value: %r' % ''.join(value))  
  176.         # swift_owner_headers are stripped by the account and container  
  177.         # controllers; we should extend header stripping to object controller  
  178.         # when a privileged object header is implemented.  
  179.         swift_owner_headers = conf.get(  
  180.             'swift_owner_headers',  
  181.             'x-container-read, x-container-write, '  
  182.             'x-container-sync-key, x-container-sync-to, '  
  183.             'x-account-meta-temp-url-key, x-account-meta-temp-url-key-2, '  
  184.             'x-account-access-control')  
  185.         self.swift_owner_headers = [  
  186.             name.strip().title()  
  187.             for name in swift_owner_headers.split(',') if name.strip()]  
  188.         # Initialization was successful, so now apply the client chunk size  
  189.         # parameter as the default read / write buffer size for the network  
  190.         # sockets.  
  191.         #  
  192.         # NOTE WELL: This is a class setting, so until we get set this on a  
  193.         # per-connection basis, this affects reading and writing on ALL  
  194.         # sockets, those between the proxy servers and external clients, and  
  195.         # those between the proxy servers and the other internal servers.  
  196.         #  
  197.         # ** Because it affects the client as well, currently, we use the  
  198.         # client chunk size as the govenor and not the object chunk size.  
  199.         socket._fileobject.default_bufsize = self.client_chunk_size  
  200.         self.expose_info = config_true_value(  
  201.             conf.get('expose_info', 'yes'))  
  202.         self.disallowed_sections = list_from_csv(  
  203.             conf.get('disallowed_sections'))  
  204.         self.admin_key = conf.get('admin_key', None)  
  205.         register_swift_info(  
  206.             version=swift_version,  
  207.             strict_cors_mode=self.strict_cors_mode,  
  208.             **constraints.EFFECTIVE_CONSTRAINTS)  
  209.   
  210.     def check_config(self):  
  211.         """ 
  212.         Check the configuration for possible errors 
  213.         """  
  214.         if self._read_affinity and self.sorting_method != 'affinity':  
  215.             self.logger.warn("sorting_method is set to '%s', not 'affinity'; "  
  216.                              "read_affinity setting will have no effect." %  
  217.                              self.sorting_method)  
  218.   
  219.     def get_controller(self, path):  
  220.         """ 
  221.         Get the controller to handle a request. 
  222.  
  223.         :param path: path from request 
  224.         :returns: tuple of (controller class, path dictionary) 
  225.  
  226.         :raises: ValueError (thrown by split_path) if given invalid path 
  227.         """  
  228.         if path == '/info':  #url是/info 则返回InfoController和包括version,expose_info,disallowed_sections,admin_key的字典  
  229.             d = dict(version=None,  
  230.                      expose_info=self.expose_info,  
  231.                      disallowed_sections=self.disallowed_sections,  
  232.                      admin_key=self.admin_key)  
  233.             return InfoController, d  
  234.   
  235.         version, account, container, obj = split_path(path, 1, 4, True)  #以/拆分url为数列，并取对应的1到4位的数据返回给对应的变量  
  236.         d = dict(version=version,  
  237.                  account_name=account,  
  238.                  container_name=container,  
  239.                  object_name=obj)  
  240.         if obj and container and account: #根据解析出的account值，congtainer和object值得有无，确定适用的Controller是那种  
  241.             return ObjectController, d  
  242.         elif container and account:  
  243.             return ContainerController, d  
  244.         elif account and not container and not obj:  
  245.             return AccountController, d  
  246.         return None, d  
  247.   
  248.     def __call__(self, env, start_response):  
  249.         """ 
  250.         WSGI entry point. 
  251.         Wraps env in swob.Request object and passes it down. 
  252.  
  253.         :param env: WSGI environment dictionary 
  254.         :param start_response: WSGI callable 
  255.         """  
  256.         try:  
  257.             if self.memcache is None: #首先判断是否memcache值存在，不存在再去获取一次  
  258.                 self.memcache = cache_from_env(env)  
  259.             req = self.update_request(Request(env)) #判断header中是否有x-storage-token和x-auth-token  
  260.             return self.handle_request(req)(env, start_response) #调用handle_request方法并返回处理的结果resp对象  
  261.         except UnicodeError:  
  262.             err = HTTPPreconditionFailed(  
  263.                 request=req, body='Invalid UTF8 or contains NULL')  
  264.             return err(env, start_response)  
  265.         except (Exception, Timeout):  
  266.             start_response('500 Server Error',  
  267.                            [('Content-Type', 'text/plain')])  
  268.             return ['Internal server error.\n']  
  269.   
  270.     def update_request(self, req):  
  271.         if 'x-storage-token' in req.headers and \  
  272.                 'x-auth-token' not in req.headers:  
  273.             req.headers['x-auth-token'] = req.headers['x-storage-token']  
  274.         return req  
  275.   
  276.     def handle_request(self, req):  
  277.         """ 
  278.         Entry point for proxy server. 
  279.         Should return a WSGI-style callable (such as swob.Response). 
  280.  
  281.         :param req: swob.Request object 
  282.         """  
  283.         try:  
  284.             self.logger.set_statsd_prefix('proxy-server') #在日志的开头加上‘proxy-server’,方便跟踪分析  
  285.             if req.content_length and req.content_length < 0: #检查header里面中的Content-Length是否有值，无值返回错误请求，并日志记录  
  286.                 self.logger.increment('errors')  
  287.                 return HTTPBadRequest(request=req,  
  288.                                       body='Invalid Content-Length')  
  289.   
  290.             try:  
  291.                 if not check_utf8(req.path_info): #检查Pathde的编码是否不满足utf8，不满足返回错误请求，并日志记录  
  292.                     self.logger.increment('errors')  
  293.                     return HTTPPreconditionFailed(  
  294.                         request=req, body='Invalid UTF8 or contains NULL')  
  295.             except UnicodeError:  
  296.                 self.logger.increment('errors')  
  297.                 return HTTPPreconditionFailed(  
  298.                     request=req, body='Invalid UTF8 or contains NULL')  
  299.   
  300.             try:  
  301.                 controller, path_parts = self.get_controller(req.path) #调用get_controller(self,path)方法返回正确的controller类和字典对象  
  302.                 p = req.path_info  
  303.                 if isinstance(p, unicode):  
  304.                     p = p.encode('utf-8') #path编码Unicode转换utf-8  
  305.             except ValueError:           #发生值异常，返回错误请求，并日志记录  
  306.                 self.logger.increment('errors')  
  307.                 return HTTPNotFound(request=req)  
  308.             if not controller:         #为找到对应处理的controller类时，返回错误请求，并日志记录  
  309.                 self.logger.increment('errors')  
  310.                 return HTTPPreconditionFailed(request=req, body='Bad URL')  
  311.             if self.deny_host_headers and \    #当proxy-server.conf中deny_host_headers有值，且请求的header中的host在deny_host_headers中，则返回错误请求，并日志记录  
  312.                     req.host.split(':')[0] in self.deny_host_headers:  
  313.                 return HTTPForbidden(request=req, body='Invalid host header')  
  314.   
  315.             self.logger.set_statsd_prefix('proxy-server.' +  
  316.                                           controller.server_type.lower()) #在日志的开头加上‘proxy-server.controller类中的请求类型（eg：HEAD/GET/PUT）’,方便跟踪分析  
  317.             controller = controller(self, **path_parts)  #初始化实际的controller对象（AccountController、ContainerController、ObjectController、InfoController其中之一）  
  318.             if 'swift.trans_id' not in req.environ:    #如果没有trans_id在env中，则重新生成一个，有些类似于http请求中的seesionID的感觉，是一种UUID  
  319.                 # if this wasn't set by an earlier middleware, set it now  
  320.                 trans_id = generate_trans_id(self.trans_id_suffix)  
  321.                 req.environ['swift.trans_id'] = trans_id  
  322.                 self.logger.txn_id = trans_id  
  323.             req.headers['x-trans-id'] = req.environ['swift.trans_id']  
  324.             controller.trans_id = req.environ['swift.trans_id']  
  325.             self.logger.client_ip = get_remote_client(req)  #把请求中获取出请求端的IP信息，加入logger对象，方便后续日志查看分析  
  326.             try:  
  327.                 handler = getattr(controller, req.method) #根据req.method方法获取对应controller对象中的方法（可能是多个，有的有public标签，有的没有）  
  328.                 getattr(handler, 'publicly_accessible') #再根据public标签获取最终的处理方法。（在方法前面可以加 @public 和@delay_denial）  
  329.             except AttributeError:  
  330.                 allowed_methods = getattr(controller, 'allowed_methods', set())  
  331.                 return HTTPMethodNotAllowed(  
  332.                     request=req, headers={'Allow': ', '.join(allowed_methods)})  
  333.             if 'swift.authorize' in req.environ: #做鉴权操作  
  334.                 # We call authorize before the handler, always. If authorized,  
  335.                 # we remove the swift.authorize hook so isn't ever called  
  336.                 # again. If not authorized, we return the denial unless the  
  337.                 # controller's method indicates it'd like to gather more  
  338.                 # information and try again later.  
  339.                 resp = req.environ['swift.authorize'](req)  
  340.                 if not resp:  
  341.                     # No resp means authorized, no delayed recheck required.  
  342.                     del req.environ['swift.authorize']  
  343.                 else:  
  344.                     # Response indicates denial, but we might delay the denial  
  345.                     # and recheck later. If not delayed, return the error now.  
  346.                     if not getattr(handler, 'delay_denial', None):  
  347.                         return resp  
  348.             # Save off original request method (GET, POST, etc.) in case it  
  349.             # gets mutated during handling.  This way logging can display the  
  350.             # method the client actually sent.  
  351.             req.environ['swift.orig_req_method'] = req.method  
  352.             return handler(req)    #调用最终的method方法，并返回resp结果  
  353.         except HTTPException as error_response:  
  354.             return error_response  
  355.         except (Exception, Timeout):  
  356.             self.logger.exception(_('ERROR Unhandled exception in request'))  
  357.             return HTTPServerError(request=req)  
  358.   
  359.     def sort_nodes(self, nodes):  
  360.         ''''' 
  361.         Sorts nodes in-place (and returns the sorted list) according to 
  362.         the configured strategy. The default "sorting" is to randomly 
  363.         shuffle the nodes. If the "timing" strategy is chosen, the nodes 
  364.         are sorted according to the stored timing data. 
  365.         '''  
  366.         # In the case of timing sorting, shuffling ensures that close timings  
  367.         # (ie within the rounding resolution) won't prefer one over another.  
  368.         # Python's sort is stable (http://wiki.python.org/moin/HowTo/Sorting/)  
  369.         shuffle(nodes)  
  370.         if self.sorting_method == 'timing':  #配置文件中排序方法为timing时，以时间排序  
  371.             now = time()  
  372.   
  373.             def key_func(node):  
  374.                 timing, expires = self.node_timings.get(node['ip'], (-1.0, 0))  
  375.                 return timing if expires > now else -1.0  
  376.             nodes.sort(key=key_func)  
  377.         elif self.sorting_method == 'affinity': #配置文件中排序方法为affinity时，以自定义的亲和力规则排序  
  378.             nodes.sort(key=self.read_affinity_sort_key)  
  379.         return nodes  
  380.   
  381.     def set_node_timing(self, node, timing):  
  382.         if self.sorting_method != 'timing':  
  383.             return  
  384.         now = time()  
  385.         timing = round(timing, 3)  # sort timings to the millisecond  
  386.         self.node_timings[node['ip']] = (timing, now + self.timing_expiry)  
  387.   
  388.     def error_limited(self, node):  
  389.         """ 
  390.         Check if the node is currently error limited. 
  391.  
  392.         :param node: dictionary of node to check 
  393.         :returns: True if error limited, False otherwise 
  394.         """  
  395.         now = time()  
  396.         if 'errors' not in node:  #errors没在node中时返回false  
  397.             return False  
  398.         if 'last_error' in node and node['last_error'] < \   #last_error在node中有并且 last_error小于现在时间减去系统允许的时间间隔  
  399.                 now - self.error_suppression_interval:  
  400.             del node['last_error'] #node去掉last_error  
  401.             if 'errors' in node: #errors在node中时返回 去掉errors，且返回false  
  402.                 del node['errors']  
  403.             return False  
  404.         limited = node['errors'] > self.error_suppression_limit  #errors在node中中的个数多与系统允许的个数，是返回true，且做日志记录  
  405.         if limited:  
  406.             self.logger.debug(  
  407.                 _('Node error limited %(ip)s:%(port)s (%(device)s)'), node)  
  408.         return limited  
  409.   
  410.     def error_limit(self, node, msg):  
  411.         """ 
  412.         Mark a node as error limited. This immediately pretends the 
  413.         node received enough errors to trigger error suppression. Use 
  414.         this for errors like Insufficient Storage. For other errors 
  415.         use :func:`error_occurred`. 
  416.  
  417.         :param node: dictionary of node to error limit 
  418.         :param msg: error message 
  419.         """  
  420.         node['errors'] = self.error_suppression_limit + 1  
  421.         node['last_error'] = time()  
  422.         self.logger.error(_('%(msg)s %(ip)s:%(port)s/%(device)s'),  
  423.                           {'msg': msg, 'ip': node['ip'],  
  424.                           'port': node['port'], 'device': node['device']})  
  425.   
  426.     def error_occurred(self, node, msg):  
  427.         """ 
  428.         Handle logging, and handling of errors. 
  429.  
  430.         :param node: dictionary of node to handle errors for 
  431.         :param msg: error message 
  432.         """  
  433.         node['errors'] = node.get('errors', 0) + 1  
  434.         node['last_error'] = time()  
  435.         self.logger.error(_('%(msg)s %(ip)s:%(port)s/%(device)s'),  
  436.                           {'msg': msg, 'ip': node['ip'],  
  437.                           'port': node['port'], 'device': node['device']})  
  438.   
  439.     def iter_nodes(self, ring, partition, node_iter=None):  
  440.         """ 
  441.         Yields nodes for a ring partition, skipping over error 
  442.         limited nodes and stopping at the configurable number of 
  443.         nodes. If a node yielded subsequently gets error limited, an 
  444.         extra node will be yielded to take its place. 
  445.  
  446.         Note that if you're going to iterate over this concurrently from 
  447.         multiple greenthreads, you'll want to use a 
  448.         swift.common.utils.GreenthreadSafeIterator to serialize access. 
  449.         Otherwise, you may get ValueErrors from concurrent access. (You also 
  450.         may not, depending on how logging is configured, the vagaries of 
  451.         socket IO and eventlet, and the phase of the moon.) 
  452.  
  453.         :param ring: ring to get yield nodes from 
  454.         :param partition: ring partition to yield nodes for 
  455.         :param node_iter: optional iterable of nodes to try. Useful if you 
  456.             want to filter or reorder the nodes. 
  457.         """  
  458.         part_nodes = ring.get_part_nodes(partition)  
  459.         if node_iter is None:  
  460.             node_iter = itertools.chain(part_nodes,  
  461.                                         ring.get_more_nodes(partition))  
  462.         num_primary_nodes = len(part_nodes)  
  463.   
  464.         # Use of list() here forcibly yanks the first N nodes (the primary  
  465.         # nodes) from node_iter, so the rest of its values are handoffs.  
  466.         primary_nodes = self.sort_nodes(  
  467.             list(itertools.islice(node_iter, num_primary_nodes)))  
  468.         handoff_nodes = node_iter  
  469.         nodes_left = self.request_node_count(len(primary_nodes))  
  470.   
  471.         for node in primary_nodes:  
  472.             if not self.error_limited(node):  
  473.                 yield node  
  474.                 if not self.error_limited(node):  
  475.                     nodes_left -= 1  
  476.                     if nodes_left <= 0:  
  477.                         return  
  478.         handoffs = 0  
  479.         for node in handoff_nodes:  
  480.             if not self.error_limited(node):  
  481.                 handoffs += 1  
  482.                 if self.log_handoffs:  
  483.                     self.logger.increment('handoff_count')  
  484.                     self.logger.warning(  
  485.                         'Handoff requested (%d)' % handoffs)  
  486.                     if handoffs == len(primary_nodes):  
  487.                         self.logger.increment('handoff_all_count')  
  488.                 yield node  
  489.                 if not self.error_limited(node):  
  490.                     nodes_left -= 1  
  491.                     if nodes_left <= 0:  
  492.                         return  
  493.   
  494.     def exception_occurred(self, node, typ, additional_info):  
  495.         """ 
  496.         Handle logging of generic exceptions. 
  497.  
  498.         :param node: dictionary of node to log the error for 
  499.         :param typ: server type 
  500.         :param additional_info: additional information to log 
  501.         """  
  502.         self.logger.exception(  
  503.             _('ERROR with %(type)s server %(ip)s:%(port)s/%(device)s re: '  
  504.               '%(info)s'),  
  505.             {'type': typ, 'ip': node['ip'], 'port': node['port'],  
  506.              'device': node['device'], 'info': additional_info})  
  507.   
  508.     def modify_wsgi_pipeline(self, pipe):  
  509.         """ 
  510.         Called during WSGI pipeline creation. Modifies the WSGI pipeline 
  511.         context to ensure that mandatory middleware is present in the pipeline. 
  512.  
  513.         :param pipe: A PipelineWrapper object 
  514.         """  
  515.         pipeline_was_modified = False  
  516.         for filter_spec in reversed(required_filters):  #当required_filters字典中定义了需要重新排序的app时，进行pipeline的重新排序处理  
  517.             filter_name = filter_spec['name']  
  518.             if filter_name not in pipe:  
  519.                 afters = filter_spec.get('after_fn', lambda _junk: [])(pipe)  
  520.                 insert_at = 0  
  521.                 for after in afters:  
  522.                     try:  
  523.                         insert_at = max(insert_at, pipe.index(after) + 1)  
  524.                     except ValueError:  # not in pipeline; ignore it  
  525.                         pass  
  526.                 self.logger.info(  
  527.                     'Adding required filter %s to pipeline at position %d' %  
  528.                     (filter_name, insert_at))  
  529.                 ctx = pipe.create_filter(filter_name)  
  530.                 pipe.insert_filter(ctx, index=insert_at)  
  531.                 pipeline_was_modified = True  
  532.   
  533.         if pipeline_was_modified:    
  534.             self.logger.info("Pipeline was modified. New pipeline is \"%s\".",  
  535.                              pipe)  
  536.         else:  
  537.             self.logger.debug("Pipeline is \"%s\"", pipe)  
  538.   
  539.   
  540. def app_factory(global_conf, **local_conf):  
  541.     """paste.deploy app factory for creating WSGI proxy apps."""  
  542.     conf = global_conf.copy()  
  543.     conf.update(local_conf)  
  544.     app = Application(conf)  
  545.     app.check_config()  
  546.     return app  
文章标签：
日志服务
Python
容器
中间件
关键词：
openstack swift
Swift源码
Swift openstack
openstack源码
相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
铭凡
目录
相关文章
小Lee
|
4月前
|
存储 Swift 对象存储
OpenStack的对象存储（Swift）
【8月更文挑战第24天】
小Lee
211 1
阿西吧ba
|
7月前
|
存储 运维 Swift
openstack cinder、swift运维命令
Cinder是OpenStack的块存储服务，提供虚拟机块设备功能，包括卷的创建、挂载等。它由cinder-api、cinder-scheduler和cinder-volume三个组件构成，前者处理请求，中间者负责调度，后者执行卷管理。Swift是OpenStack的分布式对象存储服务，高可用、持久性，适用于大文件存储。Swift架构对称，无单点故障，提供类似文件系统的简单逻辑和高效读写。Cinder关注虚拟机层面的块存储，而Swift则提供对象存储服务。命令示例展示了如何使用cinder和openstack命令操作云硬盘和Swift容器。
阿西吧ba
166 0
catgod007
|
存储 Swift 对象存储
（七）Open Stack（M）---- Swift安装和配置（下）
（七）Open Stack（M）---- Swift安装和配置（下）
catgod007
95 0
catgod007
|
存储 API Swift
（七）Open Stack（M）---- Swift安装和配置（上）
（七）Open Stack（M）---- Swift安装和配置（上）
catgod007
94 0
auqbllxiu
|
Swift
《OpenStack Swift 海量小文件优化之路》电子版地址
OpenStack Swift 海量小文件优化之路
auqbllxiu
86 0
《OpenStack Swift 海量小文件优化之路》电子版地址
5na6pyt44b2ti
|
存储 JSON HandyJSON
Swift-进阶 07：Mirror源码解析
Swift-进阶 07：Mirror源码解析
5na6pyt44b2ti
421 0
Swift-进阶 07：Mirror源码解析
5na6pyt44b2ti
|
Swift iOS开发 MacOS
Swift-进阶 01：Swift源码编译
Swift-进阶 01：Swift源码编译
5na6pyt44b2ti
531 0
Swift-进阶 01：Swift源码编译
指剑
|
存储 API Swift
(九)OpenStack---M版---双节点搭建---Swift安装和配置(单存储节点)
(九)OpenStack---M版---双节点搭建---Swift安装和配置(单存储节点)
指剑
257 0
(九)OpenStack---M版---双节点搭建---Swift安装和配置(单存储节点)
小曼study
|
Swift
swift 源码地址
https://github.com/apple/swift
小曼study
1532 0
余二五
|
容器
openstack学习笔记七 swift安装
余二五
1265 0