概述
Kubernetes 中主要通过 List-Watch 机制实现组件间的异步消息通信，List-Watch 机制的实现原理值得深入分析下 。
在 Kubernetes client-go 源码分析 - ListWatcher 中我们讲到 client-go 的 ListWatcher 是如何实现 List() 和 Watch() 两个方法的，在那篇文章里我们一路跟到 RESTClient 调用 apiserver 后没有就继续往下挖掘了，因为下面的技术相对独立，要聊清楚也需要不小的篇幅，于是我们单独在“原理篇”里分析 List-Watch 的实现。
今天我们先从 http 协议层面来分析 watch 的实现机制，抓包看下调用 watch 接口时数据包流向是怎样的。
http chunked
Kubernetes 里的 watch 长链接是通过 http 协议 chunked 机制实现的，在响应头里加一个 Transfer-Encoding: chunked 就可以实现分段响应。我们用 golang 来模拟一下这个过程，从而先理解 chunked 是什么。
写个 demo 程序，server 端代码如下：
1
2
3
4
5
6
7
8
9
10
11
|
func Server() {
http.HandleFunc("/name", func(w http.ResponseWriter, r *http.Request) {
flusher := w.(http.Flusher)
for i := 0; i < 2; i++ {
fmt.Fprintf(w, "Daniel Hu\n")
flusher.Flush()
<-time.Tick(1 * time.Second)
}
})
log.Fatal(http.ListenAndServe(":8080", nil))
}
|
这里的逻辑是当客户端请求 localhost:8080/name 的时候，服务器端响应两段：“Daniel Hu\n”，然后直接运行，再随便用什么工具访问一下，比如 curl localhost:8080/name，抓个包可以看到如下响应体：
chunked 类型的 response 由一个个 chunk 组成，每个 chunk 都是格式都是 Chunk size + Chunk data + Chunk boundary，也就是块大小+数据+边界标识。chunk 的结尾是一个大小为0的 chunk，也就是"0\r\n"。串在一起整体格式类似这样：
- [Chunk size][Chunk data][Chunk boundary][Chunk size][Chunk data][Chunk boundary][Chunk size=0][Chunk boundary]
在上图的例子中，服务器端响应的内容是两个相同的字符串 “Daniel Hu\n”，客户端拿到的也就是是 “10Daniel Hu\n\r\n10Daniel Hu\n\r\n0\r\n”
这种类型的数据怎么接收呢？可以这样玩：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
func Client() {
resp, err := http.Get("http://127.0.0.1:8080/name")
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
fmt.Println(resp.TransferEncoding)
reader := bufio.NewReader(resp.Body)
for {
line, err := reader.ReadString('\n')
if len(line) > 0 {
fmt.Print(line)
}
if err == io.EOF {
break
}
if err != nil {
log.Fatal(err)
}
}
}
|
输出内容如下（两个字符串中间会间隔1s）：
1
2
3
4
|
[chunked]
Daniel Hu
Daniel Hu
|
http 协议的 chunked 类型响应数据方式大概就是这个玩法，接下来我们看下调用 Kubernetes api 的时候，能不能找到里面的 chunked 痕迹。
watch api
一般现在搭建集群都是 https 暴露 api 了，而且是双向 TLS，所以我们需要先代理一道，方便调用和抓包。
本地通过 kubectl 来代理 apiserver 暴露外部访问：
1
2
|
# kubectl proxy
Starting to serve on 127.0.0.1:8001
|
然后开始 watch 一个资源，比如我这里选择 coredns 的 configmap：
curl localhost:8001/api/v1/watch/namespaces/kube-system/configmaps/coredns
1
2
|
# curl localhost:8001/api/v1/watch/namespaces/kube-system/configmaps/coredns
{"type":"ADDED","object":{"kind":"ConfigMap","apiVersion":"v1","metadata":{"name":"coredns","namespace":"kube-system","uid":"f2eb4080-ce86-436a-9cfb-5bfdd3f59433","resourceVersion":"747388","creationTimestamp":"2021-09-06T02:49:56Z","managedFields":[{"manager":"kubeadm","operation":"Update","apiVersion":"v1","time":"2021-09-06T02:49:56Z","fieldsType":"FieldsV1","fieldsV1":{"f:data":{}}},{"manager":"kubectl-edit","operation":"Update","apiVersion":"v1","time":"2021-10-09T10:16:13Z","fieldsType":"FieldsV1","fieldsV1":{"f:data":{"f:Corefile":{}}}}]},"data":{"Corefile":".:53 {\n errors\n health {\n lameduck 5s\n }\n ready\n kubernetes cluster.local in-addr.arpa ip6.arpa {\n pods insecure\n fallthrough in-addr.arpa ip6.arpa\n ttl 30\n }\n prometheus :9153\n forward . /etc/resolv.conf {\n max_concurrent 1000\n }\n cache 30\n loop\n reload\n loadbalance\n}\n"}}}
|
这时候可以马上拿到一个响应，然后我们通过 kubectl 命令去编辑一下这个 configmap，可以看到 watch 端继续收到一条消息：
1
|
{"type":"MODIFIED","object":{"kind":"ConfigMap","apiVersion":"v1","metadata":{"name":"coredns","namespace":"kube-system","uid":"f2eb4080-ce86-436a-9cfb-5bfdd3f59433","resourceVersion":"747834","creationTimestamp":"2021-09-06T02:49:56Z","managedFields":[{"manager":"kubeadm","operation":"Update","apiVersion":"v1","time":"2021-09-06T02:49:56Z","fieldsType":"FieldsV1","fieldsV1":{"f:data":{}}},{"manager":"kubectl-edit","operation":"Update","apiVersion":"v1","time":"2021-10-09T10:16:13Z","fieldsType":"FieldsV1","fieldsV1":{"f:data":{"f:Corefile":{}}}}]},"data":{"Corefile":".:53 {\n errors\n health {\n lameduck 5s\n }\n ready\n kubernetes cluster.local in-addr.arpa ip6.arpa {\n pods insecure\n fallthrough in-addr.arpa ip6.arpa\n ttl 30\n }\n prometheus :9153\n forward . /etc/resolv.conf {\n max_concurrent 1000\n }\n cache 31\n loop\n reload\n loadbalance\n}\n"}}}
|
原来 apiserver 是这样将资源变更通知到 watcher 的。
这时候如果我们去抓包，依旧可以看到这两个响应信息的具体数据包格式，第一个响应体如下（截取了中间关键信息）：
1
2
3
4
5
6
7
8
9
10
11
|
// ……
0030 fe d7 b6 90 af fc 85 ac 48 54 54 50 2f 31 2e 31 ........HTTP/1.1
0040 20 32 30 30 20 4f 4b 0d 0a 41 75 64 69 74 2d 49 200 OK..Audit-I
// ……
0160 2d 62 66 38 38 66 37 66 39 66 33 66 61 0d 0a 54 -bf88f7f9f3fa..T
0170 72 61 6e 73 66 65 72 2d 45 6e 63 6f 64 69 6e 67 ransfer-Encoding
0180 3a 20 63 68 75 6e 6b 65 64 0d 0a 0d 0a 33 61 38 : chunked....3a8
0190 0d 0a 7b 22 74 79 70 65 22 3a 22 41 44 44 45 44 ..{"type":"ADDED
01a0 22 2c 22 6f 62 6a 65 63 74 22 3a 7b 22 6b 69 6e ","object":{"kin
01b0 64 22 3a 22 43 6f 6e 66 69 67 4d 61 70 22 2c 22 d":"ConfigMap","
// ……
|
可以看到这里的 http 头有一个 Transfer-Encoding: chunked，下面的内容是 {“type”:“ADDED…
继续看第二个包，第二个简单很多，少了 http 头信息，只是简单的第二个 chunk，长这样：
1
2
3
4
5
6
|
// ……
0030 fe d7 fb 0b af fc c0 4a 33 61 62 0d 0a 7b 22 74 .......J3ab..{"t
0040 79 70 65 22 3a 22 4d 4f 44 49 46 49 45 44 22 2c ype":"MODIFIED",
0050 22 6f 62 6a 65 63 74 22 3a 7b 22 6b 69 6e 64 22 "object":{"kind"
0060 3a 22 43 6f 6e 66 69 67 4d 61 70 22 2c 22 61 70 :"ConfigMap","ap
// ……
|
这里可以看到 0d 0a，出于程序员的职业敏感，得想到这个就是 \r\n，至于前面的 3ab，猜到了吗？转十进制就是 939，对应这个 chunk 的长度，这里和前面我们自己写的 http server 请求-响应格式就一致了。
