Flask应用示例6 - 集成微服务调用链Zipkin

2020-06-14 本文已影响0人红薯爱帅

1. 前言

在传统单体服务或少数微服务架构中，debug相对简单，微服务之间的依赖关系也比较容易梳理。
但是，在较大的微服务架构中，如何更好地解决上述问题，需要引入微服务调用链，来记录和排查问题，比如Zipkin。
主要解决的问题：

微服务系统瓶颈分析，通过分析API调用链，了解哪个是核心service（需提高健壮性），那个service的耗时较长（需拆分或代码优化），等
Debug，troubleshoot latency problems，or find Exception in a data flow.
微服务依赖关系，zipkin提供服务依赖关系分析功能

微服务架构

2. zipkin架构

zipkin架构

2.1. components

collector

Once the trace data arrives at the Zipkin collector daemon, it is validated, stored, and indexed for lookups by the Zipkin collector.

storage

Zipkin was initially built to store data on Cassandra since Cassandra is scalable, has a flexible schema, and is heavily used within Twitter. However, we made this component pluggable. In addition to Cassandra, we natively support ElasticSearch and MySQL. Other back-ends might be offered as third party extensions.

search

Once the data is stored and indexed, we need a way to extract it. The query daemon provides a simple JSON API for finding and retrieving traces. The primary consumer of this API is the Web UI.
https://zipkin.io/zipkin-api/

web UI

We created a GUI that presents a nice interface for viewing traces. The web UI provides a method for viewing traces based on service, time, and annotations.
Note: there is no built-in authentication in the UI!

2.2. Terms

transport

Spans sent by the instrumented library must be transported from the services being traced to Zipkin collectors. There are three primary transports: HTTP, Kafka and Scribe.

trace

trace表示一个api调用链，即经过哪些service，每个service消耗的时间，等

span

span表示每次调用，即参与api调用链的一个service。主要包含三个基本元素，span_id, parenet_id, trace_id，其他信息还包含，name、duration、tags、localEndpoint等

{
      "traceId": "5982fe77008310cc80f1da5e10147517",
      "name": "query",
      "id": "be2d01e33cc78d97",
      "parentId": "ebf33e1a81dc6f71",
      "timestamp": 1458702548786000,
      "duration": 13000,
      "localEndpoint": {
        "serviceName": "zipkin-query",
        "ipv4": "192.168.1.2",
        "port": 9411
      },
      "remoteEndpoint": {
        "serviceName": "spanstore-jdbc",
        "ipv4": "127.0.0.1",
        "port": 3306
      },
      "annotations": [
        {
          "timestamp": 1458702548786000,
          "value": "cs"
        },
        {
          "timestamp": 1458702548799000,
          "value": "cr"
        }
      ],
      "tags": {
        "jdbc.query": "select distinct `zipkin_spans`.`trace_id` from `zipkin_spans` join `zipkin_annotations` on (`zipkin_spans`.`trace_id` = `zipkin_annotations`.`trace_id` and `zipkin_spans`.`id` = `zipkin_annotations`.`span_id`) where (`zipkin_annotations`.`endpoint_service_name` = ? and `zipkin_spans`.`start_ts` between ? and ?) order by `zipkin_spans`.`start_ts` desc limit ?",
        "sa": "true"
      }
    }

2.3. 工作原理

如何使用zipkin

注意：需要异步发送span，否则，会影响业务API
支持的语言或框架
https://zipkin.io/pages/tracers_instrumentation

3. 启动zipkin

zipkin提供了多种启动方式，java、docker、docker-compose等

docker run -d -p 9411:9411 openzipkin/zipkin

访问zipkin ui, http://localhost:9411

zipkin ui
更多启动方式
https://hub.docker.com/r/openzipkin/zipkin/
https://github.com/openzipkin/zipkin/tree/master/docker

4. flask+zipkin

start multiple service

$ SERVICE_NAME=app3 SERVICE_PORT=6002 \
python app.py
$ SERVICE_NAME=app2 SERVICE_PORT=6001 NEXT_SERVICE_API=http://localhost:6002 \
python app.py
$ SERVICE_NAME=app1 SERVICE_PORT=6000 NEXT_SERVICE_API=http://localhost:6001 \
python app.py

call api portal

$ http :6000/hello
HTTP/1.0 200 OK
Content-Length: 5
Content-Type: text/html; charset=utf-8
Date: Sun, 14 Jun 2020 05:14:33 GMT
Server: Werkzeug/1.0.1 Python/3.6.9

hello

check zipkin ui
http://localhost:9411
image.png
image.png

5. code

app.py

from flask import Flask, request, abort, g
from py_zipkin.zipkin import create_http_headers_for_new_span
import requests
from time import sleep
from random import randint

import config as global_config
from logging_helper import logger
from zipkin import flask_start_zipkin, flask_stop_zipkin, set_tags


def create_app(config=None):
    app = Flask(__name__)

    config_app(app, config)
    init_controller(app)

    @app.before_request
    def before():
        flask_start_zipkin()

    @app.teardown_request
    def teardown(e):
        flask_stop_zipkin()

    return app


def init_controller(app):
    @app.route('/<string:name>', methods=['GET'])
    def do_stuff(name):
        logger.info(f'{global_config.SERVICE_NAME} - {name}')
        sleep(randint(100, 1000)/1000)
        if global_config.NEXT_SERVICE_API:
            headers = create_http_headers_for_new_span()
            requests.get(f'{global_config.NEXT_SERVICE_API}/{name}', headers=headers)
            set_tags(dict(api_name=name))
        return name, 200


def config_app(app, config):
    app.config.from_object(global_config)

    if config is not None and isinstance(config, dict):
        app.config.update(config)


if __name__ == '__main__':
    app = create_app()
    app.run(host="0.0.0.0", port=global_config.SERVICE_PORT, debug=True)

config.py

import os

env = os.environ.get
SERVICE_NAME = env('SERVICE_NAME', 'try_zipkin')
SERVICE_PORT = env('SERVICE_PORT', 6000)
NEXT_SERVICE_API = env('NEXT_SERVICE_API')

# ===== Zipkin ========
ZIPKIN_HOST = env('ZIPKIN_HOST', '127.0.0.1')
ZIPKIN_PORT = env('ZIPKIN_PORT', 9411)
ZIPKIN_SAMPLE_RATE = env('ZIPKIN_SAMPLE_RATE', 100.0)
ZIPKIN_DISABLE = env("ZIPKIN_DISABLE", False)
ZIPKIN_SERVICE_NAME = SERVICE_NAME

zipkin.py

import random
import functools
from flask import request, g
import requests
from py_zipkin.zipkin import zipkin_span, ZipkinAttrs
from py_zipkin.transport import BaseTransportHandler

import config as config
from logging_helper import logger


def gen_hex_str(length=16):
    charset = (
        'abcdef'
        '0123456789'
    )
    return ''.join([random.choice(charset) for _ in range(length)])


class HttpTransport(BaseTransportHandler):
    def get_max_payload_bytes(self):
        return None

    def send(self, encoded_span):
        # The collector expects a thrift-encoded list of spans.
        try:
            requests.post(
                f'http://{config.ZIPKIN_HOST}:{config.ZIPKIN_PORT}/api/v1/spans',
                data=encoded_span,
                headers={'Content-Type': 'application/x-thrift'},
                timeout=(1, 5),
            )
        except Exception as e:
            logger.exception("Failed to send to zipkin: %s", str(e))


class KafkaTransport(BaseTransportHandler):

    def get_max_payload_bytes(self):
        # By default Kafka rejects messages bigger than 1000012 bytes.
        return 1000012

    def send(self, message):
        from kafka import SimpleProducer, KafkaClient
        kafka_client = KafkaClient('{}:{}'.format('localhost', 9092))
        producer = SimpleProducer(kafka_client)
        producer.send_messages('kafka_topic_name', message)


def with_zipkin_span(span_name, **kwargs):
    """https://github.com/Yelp/py_zipkin/issues/96
    """
    def decorate(f):
        @functools.wraps(f)
        def inner(*args, **kw):
            zipkin_kwargs = dict(
                span_name=span_name,
                service_name=config.ZIPKIN_SERVICE_NAME
            )
            zipkin_kwargs.update(**kwargs)
            with zipkin_span(**zipkin_kwargs):
                return f(*args, **kw)
        return inner
    return decorate


def set_tags(extra_annotations):
    if getattr(g, '_zipkin_span', None):
        logger.info("Set tags")
        try:
            span = g._zipkin_span
            if span:
                logger.info("update: %s", extra_annotations)
                span.update_binary_annotations(extra_annotations)
        except Exception as e:
            logger.log_critical_error("Failed to set zipkin: %s", str(e))


def get_zipkin_attrs(headers):
    trace_id = headers.get('X-B3-TraceID') or gen_hex_str()
    span_id = headers.get('X-B3-SpanID') or gen_hex_str()
    parent_span_id = headers.get('X-B3-ParentSpanID')
    flags = headers.get('X-B3-Flags')
    is_sampled = headers.get('X-B3-Sampled') == '1'

    return ZipkinAttrs(
        trace_id=trace_id,
        span_id=span_id,
        parent_span_id=parent_span_id,
        flags=flags,
        is_sampled=is_sampled,
    )


def flask_start_zipkin():
    """
    Put it to before_request event.
    :return:
    """
    if config.ZIPKIN_DISABLE:
        return
    zipkin_attrs = get_zipkin_attrs(request.headers)
    if request.headers.get('X-B3-TraceID') or request.headers.get('X-B3-SpanID'):
        logger.info(f'Zipkin Debug: received header: {request.headers}')
    span = zipkin_span(
        service_name=config.ZIPKIN_SERVICE_NAME,
        span_name=f'{request.endpoint}.{request.method}',
        transport_handler=HttpTransport(),
        host=config.ZIPKIN_HOST,
        port=config.ZIPKIN_PORT,
        sample_rate=config.ZIPKIN_SAMPLE_RATE,
        zipkin_attrs=zipkin_attrs,
    )
    g._zipkin_span = span
    g._zipkin_span.start()


def flask_stop_zipkin():
    """
    Put it to tear_request event.
    :return:
    """
    if config.ZIPKIN_DISABLE:
        return

    if getattr(g, '_zipkin_span', None):
        print(f'stop zipkin span {g._zipkin_span}')
        g._zipkin_span.stop()

logging_helper.py

import logging
import types
import sys

import config as config


def init_logger_handler():
    logger_ = logging.getLogger(config.SERVICE_NAME)  # type: logging.Logger

    logger_.setLevel(logging.INFO)

    handler = logging.StreamHandler()
    handler.setLevel(logging.INFO)
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    logger_.addHandler(handler)
    return logger_


def log_response_error(self, resp):
    if sys.exc_info() != (None, None, None):
        self.exception('request failed')

    if resp is None:
        self.error('request hasn\'t been sent out')
    else:
        try:
            self.error(resp.json())
        except Exception:
            try:
                self.error(resp.text)
            except Exception:
                self.error(f'unknown resp content (status code {resp.status_code})')


logger = init_logger_handler()
logger.log_response_error = types.MethodType(log_response_error, logger)

requirements.txt

-i https://mirrors.aliyun.com/pypi/simple
flask
requests
py_zipkin

6. references

一文搞懂基于zipkin的分布式追踪系统原理与实现
https://juejin.im/post/5c3d4df0f265da61307517ad
zipkin
https://zipkin.io/