ServiceMesh有关sidecar理解

理解sidecar的功能特性

在前文中分享了有关ServiceMesh的大概的一个演进过程，该过程都是大家对于服务管理与规划的解决方案。本文主要是通过简单的代码的编写来更加体会sidecar相关的功能与优缺点。

sidecar的功能点

sidecar实现的过程中，主要包含了sidecar与sidecar的通信，sidecar对应用流量的管控，对于不同的应用的限流、熔断等策略的实现。主要架构图如下；

大致功能流程如上所示，将以往在应用程序A或B中实现的服务发现、应用限流等功能都集成在了sidecar中，后续的升级管理都可以通过完成sidecar相关的管理就可以集中管理并平滑升级，并且当前所有的sidecar的管理都被接入到了管理功能中从而完成对所有的sidecar的管理与配置功能。

sidecar的示例代码

sidecar代码

sidecar示例代码

import random
import time
import sys
from uuid import uuid4from flask import Flask, jsonify, request
import etcd3
from threading import Lock, Thread
from etcd3.events import PutEvent, DeleteEvent
import requestsclass ServiceDisover(object):def __init__(self, host="127.0.0.1", port=2379, prefix="service"):self.host = hostself.port = portself.prefix = prefixself.sidecar_prefix = "/sidecar"self.etcd_client = etcd3.client(self.host, self.port)self.connects = {}self.sidecar_connects = {}self.lock = Lock()self.start_watch()self.start_watch_sidecar()def start_watch(self):t = Thread(target=self.watch_service)t.start()def start_watch_sidecar(self):t = Thread(target=self.watch_sidecar, args=(self.sidecar_prefix, ))t.start()def register_sidecar(self, sidecar_name, connect_msg):if not sidecar_name or not connect_msg:raise Exception(" args error")path = self.sidecar_prefix + "/" + sidecar_name + "/" + connect_msgself.etcd_client.put(path, "value")def unregister_sidecar(self, sidecar_name, service_name, connect_msg):if not service_name or not connect_msg:raise Exception(" args error")path = self.sidecar_prefix + "/" + sidecar_name + "/" + connect_msgself.etcd_client.delete_prefix(path)def register(self, sidecar_name, service_name, connect_msg):if not service_name or not connect_msg:raise Exception(" args error")path = self.prefix + "/" + sidecar_name + "/" + service_name + "/" + connect_msgself.etcd_client.put(path, "value")def unregister(self, sidecar_name, service_name, connect_msg):if not service_name or not connect_msg:raise Exception(" args error")path = self.prefix + "/" + sidecar_name + "/" + service_name + "/" + connect_msgself.etcd_client.delete_prefix(path)def get_services(self, service_name, selector):with self.lock:if service_name not in self.connects:returnreturn selector.choose(self.connects[service_name])def decode_service_key(self, event):print("event key  ", event.key, event.value)service = event.key.decode("utf-8")print("event key  ", service)services = service.split("/")if len(services) != 5:print("error  event  {0}".format(services))returnreturn services[2], services[3], services[4]def watch_service(self, prefix=None):if prefix is None:# prefix = self.prefix + "/" + SidecarNameprefix = self.prefixprint("start watch ", prefix)values = self.etcd_client.get_prefix(prefix)for v in values:key = v[1].key.decode("utf-8")res = key.split("/")if len(res) != 5:continuesidecar_name = res[2]service_name = res[3]service_connect_msg = res[4]if service_name not in self.connects:self.connects[service_name] = [[sidecar_name, service_connect_msg]]else:if [sidecar_name, service_connect_msg] not in self.connects[service_name]:self.connects[service_name].append([sidecar_name, service_connect_msg])print(" service  msg  ", self.connects)vents_iterator, cancel = self.etcd_client.watch_prefix(prefix)  # 监听etcd中aaa键 是否发生改变，for event in vents_iterator:if isinstance(event, PutEvent):res = self.decode_service_key(event)print("put event  ", res)if res is None:continuewith self.lock:sidecar_name = res[0]service_name = res[1]service_connect_msg = res[2]if service_name not in self.connects:self.connects[service_name] = [[sidecar_name, service_connect_msg]]else:if [sidecar_name, service_connect_msg] not in self.connects[service_name]:self.connects[service_name].append([sidecar_name, service_connect_msg])elif isinstance(event, DeleteEvent):res = self.decode_service_key(event)print("delete event  ", res)if res is None:continuewith self.lock:sidecar_name = res[0]service_name = res[1]service_connect_msg = res[2]if service_name not in self.connects:print("not found service_name : {0} in connects ".format(service_name))continueif [sidecar_name, service_connect_msg] in self.connects[service_name]:self.connects[service_name].remove([sidecar_name, service_connect_msg])else:print('not exists event  {0}'.format(event))print("finial  ", self.connects)def decode_sidecar_key(self, event):print("sidecar event key  ", event.key, event.value)service = event.key.decode("utf-8")print("sidecar event key  ", service)services = service.split("/")if len(services) != 4:print("error  event  {0}".format(services))returnreturn services[2], services[3]def watch_sidecar(self, prefix=None):if prefix is None:prefix = self.sidecar_prefixprint("watch sidecar  ", prefix)values = self.etcd_client.get_prefix(prefix)for v in values:key = v[1].key.decode("utf-8")res = key.split("/")if len(res) != 4:continueself.sidecar_connects[res[2]] = res[3]print(" sidecar  msg  ", self.sidecar_connects)vents_iterator, cancel = self.etcd_client.watch_prefix(prefix)  # 监听etcd中aaa键 是否发生改变，for event in vents_iterator:if isinstance(event, PutEvent):res = self.decode_sidecar_key(event)print("put event  ", res)if res is None:continuewith self.lock:sidecar_name = res[0]sidecar_connect_msg = res[1]self.sidecar_connects[sidecar_name] = sidecar_connect_msgelif isinstance(event, DeleteEvent):res = self.decode_sidecar_key(event)print("delete event  ", res)if res is None:continuewith self.lock:sidecar_name = res[0]sidecar_connect_msg = res[1]if sidecar_name not in self.sidecar_connects:print("not found sidecar_name : {0} in connects ".format(sidecar_name))continueself.sidecar_connects.pop(sidecar_name)else:print('not exists sidecar event  {0}'.format(event))print("finial  ", self.sidecar_connects)class RandomLoadBalance(object):"""负载均衡 简单的随机选择"""def choose(self, values):if len(values) == 0:returnreturn random.choice(values)class RateBucketLimit(object):"""熔断限流"""def __init__(self, limit_time=1, rate=1):self.historys = []# 安装秒级限流，如一秒钟一个令牌self.rate = rate# 限制时间 即多长的时间内产生总共多个的bucket 秒self.limit_time = limit_time# 桶总共多大self.bucket = int(self.limit_time*self.rate)self.lock = Lock()def allow(self, n=1):# 获取毫秒时间戳millis = int(round(time.time() * 1000))with self.lock:if len(self.historys) == 0:if self.bucket >= n:for i in range(n):self.historys.append(millis)return Truereturn Falsefor i, v in enumerate(self.historys):if millis - v <= self.limit_time*1000:breakcan_allow_n = int(int((millis - v)/1000)*self.rate)if can_allow_n >= n:for i in range(n):self.historys.append(millis)if len(self.historys) > self.bucket:self.historys = self.historys[-self.bucket:]return Truereturn Falseclass Service(object):def __init__(self, sidecar_name, connect_msg, host="127.0.0.1", port=2379, prefix="/service"):self.discover = ServiceDisover(host, port, prefix)self.limit = RateBucketLimit(limit_time=5)self.selector = RandomLoadBalance()self.sidecar_name = sidecar_nameself.connect_msg = connect_msgself.retry = 5def get_service(self, service_name, uri):if self.limit.allow():select_msg = self.discover.get_services(service_name, self.selector)if select_msg is None or len(select_msg) != 2:print("not found select_msg or len error", select_msg)returnsidecar_name = select_msg[0]host = select_msg[1]if sidecar_name != self.sidecar_name:if sidecar_name not in self.discover.sidecar_connects:print("not found sidecar_name : {0}  connect {1}".format(sidecar_name, self.discover.sidecar_connects))returnhost = self.discover.sidecar_connects[sidecar_name]uri = "/service?uri={0}&service_name={1}".format(uri, service_name)retry_count = 0while retry_count < self.retry:try:resp = requests.get("http://" + host + "/" + uri)if resp.status_code == 200:return resp.json()print(" response status_code error ", resp.status_code)retry_count += 1except Exception as e:print("requests error  ", e)retry_count += 1else:print(" api limit  ")if len(sys.argv) > 1:SidecarName = sys.argv[1]SideCarConnect = sys.argv[2]
else:SidecarName = str(uuid4())SideCarConnect = "127.0.0.1:3000"print("sidecarname  ", SidecarName, SideCarConnect)
service = Service(SidecarName, SideCarConnect)
service.discover.register_sidecar(SidecarName, SideCarConnect)
print("over")
app = Flask(__name__)details = SideCarConnect.split(":")
host = details[0]
port = details[1]@app.route('/service')
def service_api():uri = request.args.get("uri")service_name = request.args.get("service_name")res = service.get_service(service_name, uri)if res is None:res = {}return jsonify(res)@app.route("/register")
def register():service_name = request.args.get("service_name")connect_msg = request.args.get("connect_msg")service.discover.register(SidecarName, service_name, connect_msg)return jsonify({"detail": "ok"})if __name__ == '__main__':app.run(host=host, port=port)

sidecar的主要工作就是先注册监听所有的sidecar对应的地址列表，这样就可以知道不同的sidecar的地址在哪里，然后sidecar再监听自己对应的服务列表，这样就可以指定自己sidecar对应的服务的列表，这样当一个sidecar接受到一个请求的过程中就知道将该服务发送到哪个sidecar中去，进而等到返回数据。

应用程序A

import sys
import requestsfrom flask import Flask, jsonifyapp = Flask(__name__)SideCarName = "sidecarA"
SideCarConnect = "127.0.0.1:3000"
ServiceName = "serviceA"if len(sys.argv) > 1:connect_msg = sys.argv[1]
else:connect_msg = "127.0.0.1:5000"details = connect_msg.split(":")
host = details[0]
port = details[1]def register():url = "http://" + SideCarConnect + "/register?service_name={0}&connect_msg={1}".format(ServiceName, connect_msg)resp = requests.get(url)print("register  ", resp.status_code, resp.content)register()def get_sidecar(service_name, uri):url = "http://" + SideCarConnect + "/service?service_name={0}&uri={1}".format(service_name, uri)resp = requests.get(url)if resp is None:return {service_name: "get sidecar error"}if resp.status_code == 200:return {service_name: resp.json()}return {service_name: "get sidecar badrequest"}def get_service_b():res = {}print(" call service_b ")result = get_sidecar("serviceB", "service_b")if result:res.update(result)return res@app.route('/service_a')
def service_a():res = {"service_a": "servic_a_msg"}res.update(get_service_b())return jsonify(res)if __name__ == '__main__':app.run(host, port)

部署在了sidecarA处的服务，并且其中有个api调用了service_b的接口服务，该接口就通过sidecarA来调用sidecarB，然后sidecarB再调用service_b的服务来提供服务。

应用程序B

import sys
import requests
from flask import Flask, jsonifySideCarName = "sidecarB"
SideCarConnect = "127.0.0.1:3001"
ServiceName = "serviceB"app = Flask(__name__)if len(sys.argv) > 1:connect_msg = sys.argv[1]
else:connect_msg = "127.0.0.1:5001"details = connect_msg.split(":")
host = details[0]
port = details[1]def register():url = "http://" + SideCarConnect + "/register?service_name={0}&connect_msg={1}".format(ServiceName, connect_msg)resp = requests.get(url)print("register  ", resp.status_code, resp.content)register()@app.route('/service_b')
def service_b():res = {"service_b": "service_b_msg"}return jsonify(res)if __name__ == '__main__':app.run(host, port)

部署位于sidecarB处的应用程序B，该服务对外提供服务，并且在运行的时候就注册到sidecar中的信息，然后通过各个sidecar来提供服务。

运行示例

首先运行sidecarA的示例代码；

(venv) 192:servicemesh_t wuzi$ python sidecar.py sidecarA 127.0.0.1:3000
sidecarname   sidecarA 127.0.0.1:3000
start watch  /service
watch sidecar   /sidecarsidecar  msg   {'sidecarA': '127.0.0.1:3000', 'sidecarB': '127.0.0.1:3001'}
overservice  msg   {'testService': [['sidecar1', '127.0.0.0:1111']], 'serviceA': [['sidecarA', '127.0.0.1:5000']], 'serviceB': [['sidecarB', '127.0.0.1:5001'], ['sidecarB', '127.0.0.1:5002']]}* Serving Flask app "sidecar" (lazy loading)* Environment: productionWARNING: This is a development server. Do not use it in a production deployment.Use a production WSGI server instead.* Debug mode: off* Running on http://127.0.0.1:3000/ (Press CTRL+C to quit)

打开另外一个终端如下运行

(venv) 192:servicemesh_t wuzi$ python sidecar.py sidecarB 127.0.0.1:3001
sidecarname   sidecarB 127.0.0.1:3001
start watch  /service
watch sidecar   /sidecarsidecar  msg   {'sidecarA': '127.0.0.1:3000', 'sidecarB': '127.0.0.1:3001'}service  msg   {'testService': [['sidecar1', '127.0.0.0:1111']], 'serviceA': [['sidecarA', '127.0.0.1:5000']], 'serviceB': [['sidecarB', '127.0.0.1:5001'], ['sidecarB', '127.0.0.1:5002']]}
over* Serving Flask app "sidecar" (lazy loading)* Environment: productionWARNING: This is a development server. Do not use it in a production deployment.Use a production WSGI server instead.* Debug mode: off* Running on http://127.0.0.1:3001/ (Press CTRL+C to quit)

此时打开另外一个终端运行应用程序A

(venv) 192:servicemesh_t wuzi$ python service_a_sidecar.py
register   200 b'{"detail":"ok"}\n'* Serving Flask app "service_a_sidecar" (lazy loading)* Environment: productionWARNING: This is a development server. Do not use it in a production deployment.Use a production WSGI server instead.* Debug mode: off* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

此时打开另外一个终端运行应用程序B1

(venv) 192:servicemesh_t wuzi$ python service_b_sidecar.py
register   200 b'{"detail":"ok"}\n'* Serving Flask app "service_b_sidecar" (lazy loading)* Environment: productionWARNING: This is a development server. Do not use it in a production deployment.Use a production WSGI server instead.* Debug mode: off* Running on http://127.0.0.1:5001/ (Press CTRL+C to quit)

此时打开另外一个终端运行应用程序B2

(venv) 192:servicemesh_t wuzi$ python service_b_sidecar.py 127.0.0.1:5002
register   200 b'{"detail":"ok"}\n'* Serving Flask app "service_b_sidecar" (lazy loading)* Environment: productionWARNING: This is a development server. Do not use it in a production deployment.Use a production WSGI server instead.* Debug mode: off* Running on http://127.0.0.1:5002/ (Press CTRL+C to quit)

此时再通过浏览器访问应用程序A，http://127.0.0.1:5000/service_a此时返回的就是如下

{
"serviceB": {
"service_b": "service_b_msg"
},
"service_a": "servic_a_msg"
}

此时查看终端运行程序就可知，首先访问到了sidecarA然后再sidecarA访问了sidecarB，sidecarB通过随机算法访问了应用程序B1或者应用程序B2，至此基础的逻辑运行完成。

总结

本文只是简单的通过功能实现的角度，实现了一个最简单的有关ServiceMesh架构实现的一个流程，最要是通过功能性的实现来理解整个ServiceMesh的实现的流程。由于本人才疏学浅，如有错误请批评指正。