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格式的不同傳回不同的值
def get_controller(self, path):
"""
get the controller to handle a request.
:param path: path from request
:returns: tuple of (controller class, path dictionary)
:raises: valueerror (thrown by split_path) if given invalid path
"""
if path == '/info': #url是/info 則傳回infocontroller和包括version,expose_info,disallowed_sections,admin_key的字典
d = dict(version=none,
expose_info=self.expose_info,
disallowed_sections=self.disallowed_sections,
admin_key=self.admin_key)
return infocontroller, d
version, account, container, obj = split_path(path, 1, 4, true) #以/拆分url為數列，并取對應的1到4位的資料傳回給對應的變量
d = dict(version=version,
account_name=account,
container_name=container,
object_name=obj)
if obj and container and account: #根據解析出的account值，congtainer和object值得有無，确定适用的controller是那種
return objectcontroller, d
elif container and account:
return containercontroller, d
elif account and not container and not obj:
return accountcontroller, d
return none, d
2.4 __call__(self, env, start_response)方法，是server子產品的實際對account、container、object等對象調用處理的功能入口。
def __call__(self, env, start_response):
wsgi entry point.
wraps env in swob.request object and passes it down.
:param env: wsgi environment dictionary
:param start_response: wsgi callable
try:
if self.memcache is none: #首先判斷是否memcache值存在，不存在再去擷取一次
self.memcache = cache_from_env(env)
req = self.update_request(request(env)) #判斷header中是否有x-storage-token和x-auth-token
return self.handle_request(req)(env, start_response) #調用handle_request方法并傳回處理的結果resp對象
except unicodeerror:
err = httppreconditionfailed(
request=req, body='invalid utf8 or contains null')
return err(env, start_response)
except (exception, timeout):
start_response('500 server error',
[('content-type', 'text/plain')])
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的作用
def handle_request(self, req):
entry point for proxy server.
should return a wsgi-style callable (such as swob.response).
:param req: swob.request object
self.logger.set_statsd_prefix('proxy-server') #在日志的開頭加上‘proxy-server’,友善跟蹤分析
if req.content_length and req.content_length < 0: #檢查header裡面中的content-length是否有值，無值傳回錯誤請求，并日志記錄
self.logger.increment('errors')
return httpbadrequest(request=req,
body='invalid content-length')
try:
if not check_utf8(req.path_info): #檢查pathde的編碼是否不滿足utf8，不滿足傳回錯誤請求，并日志記錄
self.logger.increment('errors')
return httppreconditionfailed(
request=req, body='invalid utf8 or contains null')
except unicodeerror:
return httppreconditionfailed(
request=req, body='invalid utf8 or contains null')
controller, path_parts = self.get_controller(req.path) #調用get_controller(self,path)方法傳回正确的controller類和字典對象
p = req.path_info
if isinstance(p, unicode):
p = p.encode('utf-8') #path編碼unicode轉換utf-8
except valueerror: #發生值異常，傳回錯誤請求，并日志記錄
return httpnotfound(request=req)
if not controller: #為找到對應處理的controller類時，傳回錯誤請求，并日志記錄
return httppreconditionfailed(request=req, body='bad url')
if self.deny_host_headers and \ #當proxy-server.conf中deny_host_headers有值，且請求的header中的host在deny_host_headers中，則傳回錯誤請求，并日志記錄
req.host.split(':')[0] in self.deny_host_headers:
return httpforbidden(request=req, body='invalid host header')
self.logger.set_statsd_prefix('proxy-server.' +
controller.server_type.lower()) #在日志的開頭加上‘proxy-server.controller類中的請求類型（eg：head/get/put）’,友善跟蹤分析
controller = controller(self, **path_parts) #初始化實際的controller對象（accountcontroller、containercontroller、objectcontroller、infocontroller其中之一）
if 'swift.trans_id' not in req.environ: #如果沒有trans_id在env中，則重新生成一個，有些類似于http請求中的seesionid的感覺，是一種uuid
# if this wasn't set by an earlier middleware, set it now
trans_id = generate_trans_id(self.trans_id_suffix)
req.environ['swift.trans_id'] = trans_id
self.logger.txn_id = trans_id
req.headers['x-trans-id'] = req.environ['swift.trans_id']
controller.trans_id = req.environ['swift.trans_id']
self.logger.client_ip = get_remote_client(req) #把請求中擷取出請求端的ip資訊，加入logger對象，友善後續日志檢視分析
handler = getattr(controller, req.method) #根據req.method方法擷取對應controller對象中的方法（可能是多個，有的有public标簽，有的沒有）
getattr(handler, 'publicly_accessible') #再根據public标簽擷取最終的處理方法。（在方法前面可以加 @public 和@delay_denial）
except attributeerror:
allowed_methods = getattr(controller, 'allowed_methods', set())
return httpmethodnotallowed(
request=req, headers={'allow': ', '.join(allowed_methods)})
if 'swift.authorize' in req.environ: #做鑒權操作
# we call authorize before the handler, always. if authorized,
# we remove the swift.authorize hook so isn't ever called
# again. if not authorized, we return the denial unless the
# controller's method indicates it'd like to gather more
# information and try again later.
resp = req.environ['swift.authorize'](req)
if not resp:
# no resp means authorized, no delayed recheck required.
del req.environ['swift.authorize']
else:
# response indicates denial, but we might delay the denial
# and recheck later. if not delayed, return the error now.
if not getattr(handler, 'delay_denial', none):
return resp
# save off original request method (get, post, etc.) in case it
# gets mutated during handling. this way logging can display the
# method the client actually sent.
req.environ['swift.orig_req_method'] = req.method
return handler(req) #調用最終的method方法，并傳回resp結果
except httpexception as error_response:
return error_response
self.logger.exception(_('error unhandled exception in request'))
return httpservererror(request=req)
2.7 sort_nodes(self, nodes)方法，對nodes對象進行排序處理，該方法在iter_nodes(self, ring, partition, node_iter=none)中調用
def sort_nodes(self, nodes):
'''''
sorts nodes in-place (and returns the sorted list) according to
the configured strategy. the default "sorting" is to randomly
shuffle the nodes. if the "timing" strategy is chosen, the nodes
are sorted according to the stored timing data.
'''
# in the case of timing sorting, shuffling ensures that close timings
# (ie within the rounding resolution) won't prefer one over another.
# python's sort is stable (http://wiki.python.org/moin/howto/sorting/)
shuffle(nodes)
if self.sorting_method == 'timing': #配置檔案中排序方法為timing時，以時間排序
now = time()
def key_func(node):
timing, expires = self.node_timings.get(node['ip'], (-1.0, 0))
return timing if expires > now else -1.0
nodes.sort(key=key_func)
elif self.sorting_method == 'affinity': #配置檔案中排序方法為affinity時，以自定義的親和力規則排序
nodes.sort(key=self.read_affinity_sort_key)
return nodes
2.8 set_node_timing(self, node, timing)方法，提供給外部程式調用
2.9 error_limited(self, node)方法，該方法在iter_nodes(self, ring, partition, node_iter=none)中調用
def error_limited(self, node):
check if the node is currently error limited.
:param node: dictionary of node to check
:returns: true if error limited, false otherwise
now = time()
if 'errors' not in node: #errors沒在node中時傳回false
return false
if 'last_error' in node and node['last_error'] < \ #last_error在node中有并且 last_error小于現在時間減去系統允許的時間間隔
now - self.error_suppression_interval:
del node['last_error'] #node去掉last_error
if 'errors' in node: #errors在node中時傳回 去掉errors，且傳回false
del node['errors']
limited = node['errors'] > self.error_suppression_limit #errors在node中中的個數多與系統允許的個數，是傳回true，且做日志記錄
if limited:
self.logger.debug(
_('node error limited %(ip)s:%(port)s (%(device)s)'), node)
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，并做日志記錄
def modify_wsgi_pipeline(self, pipe):
called during wsgi pipeline creation. modifies the wsgi pipeline
context to ensure that mandatory middleware is present in the pipeline.
:param pipe: a pipelinewrapper object
pipeline_was_modified = false
for filter_spec in reversed(required_filters): #當required_filters字典中定義了需要重新排序的app時，進行pipeline的重新排序處理
filter_name = filter_spec['name']
if filter_name not in pipe:
afters = filter_spec.get('after_fn', lambda _junk: [])(pipe)
insert_at = 0
for after in afters:
try:
insert_at = max(insert_at, pipe.index(after) + 1)
except valueerror: # not in pipeline; ignore it
pass
self.logger.info(
'adding required filter %s to pipeline at position %d' %
(filter_name, insert_at))
ctx = pipe.create_filter(filter_name)
pipe.insert_filter(ctx, index=insert_at)
pipeline_was_modified = true
if pipeline_was_modified:
self.logger.info("pipeline was modified. new pipeline is \"%s\".",
pipe)
else:
self.logger.debug("pipeline is \"%s\"", pipe)
以下源碼為2014、3、12的最新的proxy的server.py源碼,隻加了部分代碼注釋：
# copyright (c) 2010-2012 openstack foundation
#
# 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 mimetypes
import os
import socket
from swift import gettext_ as _
from random import shuffle
from time import time
import itertools
from eventlet import timeout
from swift import __canonical_version__ as swift_version
from swift.common import constraints
from swift.common.ring import ring
from swift.common.utils import cache_from_env, get_logger, \
get_remote_client, split_path, config_true_value, generate_trans_id, \
affinity_key_function, affinity_locality_predicate, list_from_csv, \
register_swift_info
from swift.common.constraints import check_utf8
from swift.proxy.controllers import accountcontroller, objectcontroller, \
containercontroller, infocontroller
from swift.common.swob import httpbadrequest, httpforbidden, \
httpmethodnotallowed, httpnotfound, httppreconditionfailed, \
httpservererror, httpexception, request
# list of entry points for mandatory middlewares.
# fields:
# "name" (required) is the entry point name from setup.py.
# "after_fn" (optional) a function that takes a pipelinewrapper object as its
# single argument and returns a list of middlewares that this middleware
# should come after. any middlewares in the returned list that are not present
# in the pipeline will be ignored, so you can safely name optional middlewares
# to come after. for example, ["catch_errors", "bulk"] would install this
# middleware after catch_errors and bulk if both were present, but if bulk
# were absent, would just install it after catch_errors.
required_filters = [
{'name': 'catch_errors'},
{'name': 'gatekeeper',
'after_fn': lambda pipe: (['catch_errors']
if pipe.startswith("catch_errors")
else [])},
{'name': 'dlo', 'after_fn': lambda _junk: ['catch_errors', 'gatekeeper',
'proxy_logging']}]
class application(object):
"""wsgi application for the proxy server."""
def __init__(self, conf, memcache=none, logger=none, account_ring=none,
container_ring=none, object_ring=none):
if conf is none:
conf = {}
if logger is none:
self.logger = get_logger(conf, log_route='proxy-server')
self.logger = logger
swift_dir = conf.get('swift_dir', '/etc/swift')
self.node_timeout = int(conf.get('node_timeout', 10))
self.recoverable_node_timeout = int(
conf.get('recoverable_node_timeout', self.node_timeout))
self.conn_timeout = float(conf.get('conn_timeout', 0.5))
self.client_timeout = int(conf.get('client_timeout', 60))
self.put_queue_depth = int(conf.get('put_queue_depth', 10))
self.object_chunk_size = int(conf.get('object_chunk_size', 65536))
self.client_chunk_size = int(conf.get('client_chunk_size', 65536))
self.trans_id_suffix = conf.get('trans_id_suffix', '')
self.post_quorum_timeout = float(conf.get('post_quorum_timeout', 0.5))
self.error_suppression_interval = \
int(conf.get('error_suppression_interval', 60))
self.error_suppression_limit = \
int(conf.get('error_suppression_limit', 10))
self.recheck_container_existence = \
int(conf.get('recheck_container_existence', 60))
self.recheck_account_existence = \
int(conf.get('recheck_account_existence', 60))
self.allow_account_management = \
config_true_value(conf.get('allow_account_management', 'no'))
self.object_post_as_copy = \
config_true_value(conf.get('object_post_as_copy', 'true'))
self.object_ring = object_ring or ring(swift_dir, ring_name='object')
self.container_ring = container_ring or ring(swift_dir,
ring_name='container')
self.account_ring = account_ring or ring(swift_dir,
ring_name='account')
self.memcache = memcache
mimetypes.init(mimetypes.knownfiles +
[os.path.join(swift_dir, 'mime.types')])
self.account_autocreate = \
config_true_value(conf.get('account_autocreate', 'no'))
self.expiring_objects_account = \
(conf.get('auto_create_account_prefix') or '.') + \
(conf.get('expiring_objects_account_name') or 'expiring_objects')
self.expiring_objects_container_divisor = \
int(conf.get('expiring_objects_container_divisor') or 86400)
self.max_containers_per_account = \
int(conf.get('max_containers_per_account') or 0)
self.max_containers_whitelist = [
a.strip()
for a in conf.get('max_containers_whitelist', '').split(',')
if a.strip()]
self.deny_host_headers = [
host.strip() for host in
conf.get('deny_host_headers', '').split(',') if host.strip()]
self.rate_limit_after_segment = \
int(conf.get('rate_limit_after_segment', 10))
self.rate_limit_segments_per_sec = \
int(conf.get('rate_limit_segments_per_sec', 1))
self.log_handoffs = config_true_value(conf.get('log_handoffs', 'true'))
self.cors_allow_origin = [
for a in conf.get('cors_allow_origin', '').split(',')
self.strict_cors_mode = config_true_value(
conf.get('strict_cors_mode', 't'))
self.node_timings = {}
self.timing_expiry = int(conf.get('timing_expiry', 300))
self.sorting_method = conf.get('sorting_method', 'shuffle').lower()
self.max_large_object_get_time = float(
conf.get('max_large_object_get_time', '86400'))
value = conf.get('request_node_count', '2 * replicas').lower().split()
if len(value) == 1:
value = int(value[0])
self.request_node_count = lambda replicas: value
elif len(value) == 3 and value[1] == '*' and value[2] == 'replicas':
self.request_node_count = lambda replicas: value * replicas
raise valueerror(
'invalid request_node_count value: %r' % ''.join(value))
self._read_affinity = read_affinity = conf.get('read_affinity', '')
self.read_affinity_sort_key = affinity_key_function(read_affinity)
except valueerror as err:
# make the message a little more useful
raise valueerror("invalid read_affinity value: %r (%s)" %
(read_affinity, err.message))
write_affinity = conf.get('write_affinity', '')
self.write_affinity_is_local_fn \
= affinity_locality_predicate(write_affinity)
raise valueerror("invalid write_affinity value: %r (%s)" %
(write_affinity, err.message))
value = conf.get('write_affinity_node_count',
'2 * replicas').lower().split()
self.write_affinity_node_count = lambda replicas: value
self.write_affinity_node_count = lambda replicas: value * replicas
'invalid write_affinity_node_count value: %r' % ''.join(value))
# swift_owner_headers are stripped by the account and container
# controllers; we should extend header stripping to object controller
# when a privileged object header is implemented.
swift_owner_headers = conf.get(
'swift_owner_headers',
'x-container-read, x-container-write, '
'x-container-sync-key, x-container-sync-to, '
'x-account-meta-temp-url-key, x-account-meta-temp-url-key-2, '
'x-account-access-control')
self.swift_owner_headers = [
name.strip().title()
for name in swift_owner_headers.split(',') if name.strip()]
# initialization was successful, so now apply the client chunk size
# parameter as the default read / write buffer size for the network
# sockets.
#
# note well: this is a class setting, so until we get set this on a
# per-connection basis, this affects reading and writing on all
# sockets, those between the proxy servers and external clients, and
# those between the proxy servers and the other internal servers.
# ** because it affects the client as well, currently, we use the
# client chunk size as the govenor and not the object chunk size.
socket._fileobject.default_bufsize = self.client_chunk_size
self.expose_info = config_true_value(
conf.get('expose_info', 'yes'))
self.disallowed_sections = list_from_csv(
conf.get('disallowed_sections'))
self.admin_key = conf.get('admin_key', none)
register_swift_info(
version=swift_version,
strict_cors_mode=self.strict_cors_mode,
**constraints.effective_constraints)
def check_config(self):
check the configuration for possible errors
if self._read_affinity and self.sorting_method != 'affinity':
self.logger.warn("sorting_method is set to '%s', not 'affinity'; "
"read_affinity setting will have no effect." %
self.sorting_method)
def get_controller(self, path):
def __call__(self, env, start_response):
def update_request(self, req):
if 'x-storage-token' in req.headers and \
'x-auth-token' not in req.headers:
req.headers['x-auth-token'] = req.headers['x-storage-token']
return req
def handle_request(self, req):
def sort_nodes(self, nodes):
def set_node_timing(self, node, timing):
if self.sorting_method != 'timing':
return
timing = round(timing, 3) # sort timings to the millisecond
self.node_timings[node['ip']] = (timing, now + self.timing_expiry)
def error_limited(self, node):
def error_limit(self, node, msg):
mark a node as error limited. this immediately pretends the
node received enough errors to trigger error suppression. use
this for errors like insufficient storage. for other errors
use :func:`error_occurred`.
:param node: dictionary of node to error limit
:param msg: error message
node['errors'] = self.error_suppression_limit + 1
node['last_error'] = time()
self.logger.error(_('%(msg)s %(ip)s:%(port)s/%(device)s'),
{'msg': msg, 'ip': node['ip'],
'port': node['port'], 'device': node['device']})
def error_occurred(self, node, msg):
handle logging, and handling of errors.
:param node: dictionary of node to handle errors for
node['errors'] = node.get('errors', 0) + 1
def iter_nodes(self, ring, partition, node_iter=none):
yields nodes for a ring partition, skipping over error
limited nodes and stopping at the configurable number of
nodes. if a node yielded subsequently gets error limited, an
extra node will be yielded to take its place.
note that if you're going to iterate over this concurrently from
multiple greenthreads, you'll want to use a
swift.common.utils.greenthreadsafeiterator to serialize access.
otherwise, you may get valueerrors from concurrent access. (you also
may not, depending on how logging is configured, the vagaries of
socket io and eventlet, and the phase of the moon.)
:param ring: ring to get yield nodes from
:param partition: ring partition to yield nodes for
:param node_iter: optional iterable of nodes to try. useful if you
want to filter or reorder the nodes.
part_nodes = ring.get_part_nodes(partition)
if node_iter is none:
node_iter = itertools.chain(part_nodes,
ring.get_more_nodes(partition))
num_primary_nodes = len(part_nodes)
# use of list() here forcibly yanks the first n nodes (the primary
# nodes) from node_iter, so the rest of its values are handoffs.
primary_nodes = self.sort_nodes(
list(itertools.islice(node_iter, num_primary_nodes)))
handoff_nodes = node_iter
nodes_left = self.request_node_count(len(primary_nodes))
for node in primary_nodes:
if not self.error_limited(node):
yield node
if not self.error_limited(node):
nodes_left -= 1
if nodes_left <= 0:
return
handoffs = 0
for node in handoff_nodes:
handoffs += 1
if self.log_handoffs:
self.logger.increment('handoff_count')
self.logger.warning(
'handoff requested (%d)' % handoffs)
if handoffs == len(primary_nodes):
self.logger.increment('handoff_all_count')
def exception_occurred(self, node, typ, additional_info):
handle logging of generic exceptions.
:param node: dictionary of node to log the error for
:param typ: server type
:param additional_info: additional information to log
self.logger.exception(
_('error with %(type)s server %(ip)s:%(port)s/%(device)s re: '
'%(info)s'),
{'type': typ, 'ip': node['ip'], 'port': node['port'],
'device': node['device'], 'info': additional_info})
def modify_wsgi_pipeline(self, pipe):
def app_factory(global_conf, **local_conf):
"""paste.deploy app factory for creating wsgi proxy apps."""
conf = global_conf.copy()
conf.update(local_conf)
app = application(conf)
app.check_config()
return app