关键字：python sanic 微服务 异步 zipkin sanic-plugin 插件 Sanic-Plugins-Framewor Pypi发布

所需环境：python3.7, Docker, Linux or WSL

image.png

Sanic插件(Plugin/Extension) - sanic-zipkin已经ready，你可以轻松用pip安装啦：

喜欢的话，github点个赞吧：https://github.com/kevinqqnj/sanic-zipkin

pip install sanic-zipkin

# app.py

from sanic_zipkin import SanicZipkin, logger, sz_rpc

sz = SanicZipkin(app)

上一篇已经学会了如何在Sanic app里引入aiozipkin，来做分布式系统追踪。本篇，来讨论下，如何创建一个完整的Sanic插件，以方便自己或者分享给他人。

先来看看插件是怎么用的：

功能

adding "Request span" by default

if Request is from another micro-service endpoint, span will be attached (Inject/Extract) to that endpoint

use "logger" decorator to create span for "methods" calls

use "sz_rpc" method to create sub-span for RPC calls, attaching to parent span

run examples/servic_a.py and examples/service_b.py

use Docker to run zipkin or jaeger:

docker run -d -p9411:9411 openzipkin/zipkin:latest

or

docker run -d -e COLLECTOR_ZIPKIN_HTTP_PORT=9411 -p5775:5775/udp -p6831:6831/udp -p6832:6832/udp -p5778:5778 -p16686:16686 -p14268:14268 -p9411:9411 jaegertracing/all-in-one

access the endpoint:

最简应用：

curl localhost:8000/ to see plugin's basic usage

from sanic_zipkin import SanicZipkin, logger, sz_rpc

app = Sanic(__name__)

# initilize plugin, default parameters:

# zipkin_address = 'http://127.0.0.1:9411/api/v2/spans'

# service = __name__

# host = '127.0.0.1'

# port = 8000

sz = SanicZipkin(app, service='service-a')

@app.route("/")

async def index(request):

return response.json({"hello": "from index"})

This "/" endpoint will add trace span to zipkin automatically

使用装饰器装饰方法，以及链式trace

curl localhost:8000/2 to see how to decorate methods and chain-calls

@logger()

async def db_access(context, data):

await asyncio.sleep(0.1)

print(f'db_access done. data: {data}')

return

@sz.route("/2")

async def method_call(request, context):

await db_access(context, 'this is method_call data')

return response.json({"hello": 'method_call'})

Use "@logger" decorator to generate span for methods.

Note: in this case, you need to use "@sz.route" decorator, and pass contextparameter to method calls.

微服务之间通过PRC访问：

curl localhost:8000/3 to see how RPC calls working, both GET/POST is supported

@logger()

async def decorate_demo(context, data):

await db_access(context, data)

data = {'payload': 'rpc call data of decorate_demo'}

rsp = await sz_rpc(context, backend_service2, data, method='GET')

print(rsp.status, await rsp.text())

return

@sz.route("/3")

async def rpc_call(request, context):

await decorate_demo(context, 'this is index4 data')

data = {'payload': 'rpc call data of rpc_call'}

rsp = await sz_rpc(context, backend_service1, data) # default method='POST'

print(rsp.status, await rsp.text())

return response.json({"hello": 'rpc_call'})

method sz_rpc just wrapper span injection to RPC POST/GET calls. In peer server, span-context will be automatically extracted and generate a chain-view in zipkin.

在Zipkin/Jaeger UI里查看Trace:

image.png

Sanic插件开发过程

使用Sanic-Plugins-Framework开发，省时省力，而且充分利用Sanic异步框架的威力。

1. 插件初始化

用户可自定义的初始化变量

继承Sanic-Plugins-Framework(SPF) 的Contextualize类型，在on_before_registered方法引用时，加载用户自定义的初始变量。

目前支持：

zipkin server地址

微服务名称service

微服务IP, port

from spf.plugins.contextualize import Contextualize

class SanicZipkin(Contextualize):

def __init__(self, *args, **kwargs):

super(SanicZipkin, self).__init__(*args, **kwargs)

self.zipkin_address = None

self.service = None

self.host = None

self.port = None

def on_before_registered(self, context, *args, **kwargs):

self.zipkin_address = kwargs.get('zipkin_address', 'http://127.0.0.1:9411/api/v2/spans')

self.service = kwargs.get('service', __name__)

self.host = kwargs.get('host', '127.0.0.1')

self.port = kwargs.get('port', 8000)

_logger.info(f'SanicZipkin: before registered: service={self.service}')

创建aiozipkin服务

实例化sanic_zipkin，然后调用Sanic 'before_server_start'方法，初始化context.tracer、context.aio_session。

context是SPF全局可以访问的上下文变量，可以存储任何你想要共享的数据。

sanic_zipkin = instance = SanicZipkin()

@sanic_zipkin.listener('before_server_start')

async def setup_zipkin(app, loop, context):

endpoint = az.create_endpoint(sanic_zipkin.service, ipv4=sanic_zipkin.host,

port=sanic_zipkin.port)

context.tracer = await az.create(sanic_zipkin.zipkin_address, endpoint,

sample_rate=1.0)

trace_config = az.make_trace_config(context.tracer)

context.aio_session = aiohttp.ClientSession(trace_configs=[trace_config])

context.span = []

context.zipkin_headers = []

这里context.span设计成数组，模拟堆栈FILO(先进后出)，是因为考虑到链式调用时，tracer需要以parent span为基础，创建child span。同理Inject/Extract用到的context.zipkin_headers也设成数组。

2. 创建middleware，给Request GET/POST自动添加span

Requst: 先用middleware装饰器来监听request消息，然后调用自定义方法request_span(request, context)来创建span。

@sanic_zipkin.middleware(priority=2, with_context=True)

def mw1(request, context):

context.log(DEBUG, f'mw-request: add span and headers before request')

span = request_span(request, context)

context.span.append(span)

context.zipkin_headers.append(span.context.make_headers())

这里，要考虑到，当前span是new span，还是child span。

span.append()压入堆栈。

自定义方法request_span()，通过读取request里是否有zipkin_headers信息，来判断当前是其它微服务的RPC call，还是用户发起的http访问。

def request_span(request, context):

context.log(DEBUG, f'REQUEST json: {request.json}, args: {request.args}')

headers = request.parsed_json.get('zipkin_headers', None) if request.json else \

request.args.get('zipkin_headers', None)

Response: 在一次http访问结束，返回response时，此次访问的context.span和context.zipkin_headers弹出堆栈，以免污染到其它http访问的trace：span.pop()

@sanic_zipkin.middleware(priority=8, attach_to='response', relative='post',

with_context=True)

def mw2(request, response, context):

context.span.pop()

context.zipkin_headers.pop()

context.log(DEBUG, 'mw-response: clear span/zipkin_headers after Response')

3. 创建zipkin_headers，用于RPC Inject/Extract

如果当前堆栈里有zipkin_headers，则方法request_span()创建上下文：

def request_span(request, context):

context.log(DEBUG, f'REQUEST json: {request.json}, args: {request.args}')

headers = request.parsed_json.get('zipkin_headers', None) if request.json else \

request.args.get('zipkin_headers', None)

if headers:

span_context = az.make_context(headers)

with context.tracer.new_child(span_context) as span:

...

如果无，则创建新的span:

with context.tracer.new_trace() as span:

span.name(f'{request.method} {request.path}')

...

4. methods函数，添加@logger装饰器

对于非http request的方法函数，因为没有middleware可以监听，则需要创建新的装饰器了。

使用@logger时，默认context作为第一个参数：

def logger(type=None, category=None, detail=None, description=None,

tracing=True, level=logging.INFO, *args, **kwargs):

def decorator(fn=None):

@functools.wraps(fn)

async def _decorator(*args, **kwargs):

# print('_decorator args: ', args, kwargs)

context = args[0] if len(args) > 0 and isinstance(args[0], ContextDict) else None

...

然后创建新的new span 或 child span。

同时，添加(Inject)新的上下文zipkin_headers：

span = gen_span(fn.__name__, context)

context.zipkin_headers.append(span.context.make_headers())

执行装饰器所装饰的函数fn()。

之后，必须清除(弹出最上面)本次装饰器新增的临时span和zipkin_headers。因为当前函数外部的其它引用函数(如果有)所需要的数据，还在堆栈下面。

try:

exce = False

res = await fn(*args, **kwargs)

return res

except Exception as e:

exce = True

raise e

finally:

...

# clean up tmp vars for this wrapper

context.span.pop()

context.zipkin_headers.pop()

5. 创建帮助函数sz_rpc()，简化RPC 访问其它微服务时的Inject操作

这个很简单的helper，负责把zipkin_headers，inject到POST/GET访问的data里，这样，对方收到http request时，就可以顺利的Extract，得到span上下文了。

async def sz_rpc(context, url, data, method='POST'):

data.update({'zipkin_headers': json.dumps(context.zipkin_headers[-1])})

if method.upper() == 'POST':

return await context.aio_session.post(url, json=data)

else:

return await context.aio_session.get(url, params=data)

6. 发布插件到Pypi

插件写好了，下面就是发布了。

修改当前目录的结构，以及添加一些必要的标注文件：

kevinqq@CN-00009841:/c/Users/xxx/git/sanic-zipkin$ tree -L 2

├── CHANGES.txt # 版本信息，必须

├── LICENSE # 必须

├── MANIFEST.in

├── README.md

├── dist # 发布时自动打的包

│ └── sanic-zipkin-0.1.2.tar.gz

├── examples

│ ├── requirements.txt

│ ├── service_a.py

│ └── service_b.py

├── requirements-dev.txt

├── sanic_zipkin # 包的目录

│ ├── __init__.py # 必须。含版本信息和可引用的对象

│ └── sanic_zipkin.py # 主文件

├── sanic_zipkin.egg-info # 发布时自动生成

└── setup.py # 发布用的程序

发布前检查：

python3 setup.py check

发布到Pypi:

python3 setup.py sdist upload

此时，会让你输入Pypi的密码。

如果收到200，则上传成功。

检查是否已经可用：

pip install sanic-zipkin