编写一个k8s的webhook来实现deployment的拦截

Admission Webhook

Admission Webhook 是 api-server 对外提供的一个扩展能力,api-server 作为 kubernetes 的核心,几乎所有组件都需要跟他打交道,基本可以说掌控了 k8s 的 api-server,你就可以控制 k8s 的行为。

在早期的版本 api-server 并没有提供 admissionresgistration 的能力(v1.9之前),当我们要对 k8s 进行控制的时候,只能重新编译 api-server。比如你想阻止某个控制器的行为,或拦截某个控制器的资源修改。admission webhook 就是提供了这样的能力,比如你希望某个特定 label 标签的 pod 再创建的时候都注入 sidercar,或者阻止不合规的资源。

CRD解析

Admission Webhook 包涵两种 CRD:mutatingwebhookconfigurationvalidatingwebhookconfiguration

下面是一个 mutatingwebhookconfiguration 的CRD文件:

apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
name: mutating-test.shikanon.com
webhooks:
- admissionReviewVersions: # admissionReviewVersions 请求的版本
- v1beta1
- v1
clientConfig: # 客户端配置
caBundle: # ca证书
service: # 调用服务相关配置,这里是一个k8s的service,访问地址是<name>.<namespace>.svc:<port>/<path>
name: mutating-test
namespace: testing-tools
path: /mutation-deployment
port: 8000
failurePolicy: Ignore # 调用失败策略,Ignore为忽略错误, failed表示admission会处理错误
matchPolicy: Exact
name: mutating-test.shikanon.com # webhook名称
namespaceSelector: {} # 命名空间过滤条件
objectSelector: # 对象过滤条件
matchExpressions:
- key: mutating-test-webhook
operator: In
values:
- enabled
- "true"
# reinvocationPolicy表示再调度策略,因为webhook本身没有顺序性,因此每个修改后可能又被其他webhook修改,所以提供
# 一个策略表示是否需要被多次调用,Never 表示只会调度一次,IfNeeded 表示资源被修改后会再调度这个webhook
reinvocationPolicy: Never
rules: # 规则
- apiGroups:
- apps
apiVersions:
- v1
operations:
- CREATE
- UPDATE
resources:
- deployments
scope: '*' # 匹配范围,"*" 匹配所有资源,但不包括子资源,"*/*" 匹配所有资源,包括子资源
sideEffects: None # 这个表示webhook是否存在副作用,主要针对 dryRun 的请求
timeoutSeconds: 30

Webhook 的流程和格式

Admission Webhook 本质是 api-server 的一个 webhook 调用,下面是 api-server 的处理流程:

api-server 通过读取 mutatingwebhookconfigurationvalidatingwebhookconfiguration 的 CR 文件的目标地址,然后回调用户自定义的服务。

                                            ┌──────────────────────────────────┐
┌─────────────────┐ │ │
apply │ │ read │ validatingwebhookconfiguration │
────────────►│ api-server │◄───────────┤ │
│ │ │ mutatingwebhookconfiguration │
└────────┬────────┘ │ │
│ └──────────────────────────────────┘

│ 回调


┌────────▼────────┐
│ │
│ webhookservice │
│ │
└─────────────────┘

api-server 发起的请求是一串json数据格式,header需要设置content-typeapplication/json, 我们看看请求的 body :

curl -X POST \
http://webhook-url \
-H 'content-type: application/json' \
-d '{
"apiVersion": "admission.k8s.io/v1",
"kind": "AdmissionReview",
"request": {
...
}
}'

返回的结果:

{
"kind": "AdmissionReview",
"apiVersion": "admission.k8s.io/v1",
"response": {
"uid": "b955fb34-0135-4e78-908e-eeb2f874933f",
"allowed": true,
"status": {
"metadata": {},
"code": 200
},
"patch": "W3sib3AiOiJyZXBsYWNlIiwicGF0aCI6Ii9zcGVjL3JlcGxpY2FzIiwidmFsdWUiOjJ9XQ==",
"patchType": "JSONPatch"
}
}

这里的 patch 是用base64编码的一个json,我们解码看看,是一个 json patch:

$ echo "W3sib3AiOiJyZXBsYWNlIiwicGF0aCI6Ii9zcGVjL3JlcGxpY2FzIiwidmFsdWUiOjJ9XQ==" | base64 -d
[
{
"op": "replace",
"path": "/spec/replicas",
"value": 2
}
]

编写Admission Webhook服务

处理函数:

func (h *MutatingHandler) Handle(ctx context.Context, req kubeadmission.Request) kubeadmission.Response {
reqJson, err := json.Marshal(req.AdmissionRequest)
if err != nil {
return kubeadmission.Errored(http.StatusBadRequest, err)
}
fmt.Println(string(reqJson))
obj := &appsv1.Deployment{}

err = h.Decoder.Decode(req, obj)
if err != nil {
return kubeadmission.Errored(http.StatusBadRequest, err)
}
fmt.Println(obj.Name)
originObj, err := json.Marshal(obj)
if err != nil {
return kubeadmission.Errored(http.StatusBadRequest, err)
}
// 将新的资源副本数量改为1
newobj := obj.DeepCopy()
replicas := int32(1)
newobj.Spec.Replicas = &replicas
currentObj, err := json.Marshal(newobj)
if err != nil {
return kubeadmission.Errored(http.StatusBadRequest, err)
}
// 对比之前的资源类型和之后的资源类型的差异生成返回数据
resp := kubeadmission.PatchResponseFromRaw(originObj, currentObj)
respJson, err := json.Marshal(resp.AdmissionResponse)
if err != nil {
return kubeadmission.Errored(http.StatusBadRequest, err)
}
return resp
}

主程序:

func main() {
scheme := kuberuntime.NewScheme()
decoder, err := kubeadmission.NewDecoder(scheme)
if err != nil {
log.Fatalln(err)
}
mutation := MutatingHandler{
Decoder: decoder,
}
var logger = klogr.New()
webhook := kubeadmission.Webhook{
Handler: &mutation,
}

logger.Info("test", "unable to decode the request")
_, err = inject.LoggerInto(logger, &webhook)
if err != nil {
log.Fatalln(err)
}
http.HandleFunc("/mutation-deployment", webhook.ServeHTTP)
// 这里需要用到TLS证书和密钥
err = http.ListenAndServeTLS(":8000", "cert.pem", "private.key", nil)
if err != nil {
log.Fatalln(err)
}
}

生成TLS密钥

生成一个私钥

openssl genrsa -out private.key 2048

基于私钥生成一个证书签名请求(Certificate Signing Request,CSR),目标地址的域名为:mutating-test.testing-tools.svc, csr的配置:

[req]
req_extensions = v3_req
distinguished_name = req_distinguished_name
[req_distinguished_name]
[ v3_req ]
basicConstraints = CA:FALSE
keyUsage = nonRepudiation, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
subjectAltName = @alt_names
[alt_names]
DNS.1 = mutating-test
DNS.2 = mutating-test.testing-tools
DNS.3 = mutating-test.testing-tools.svc
DNS.4 = mutating-test.testing-tools.svc.cluster.local

创建命令:

# 
openssl req -new -key private.key -days 36500 -subj "/CN=mutating-test.testing-tools.svc" -out service.csr -config csr.conf

用api-server的CA进行签发

基于csr创建 CertificateSigningRequest:

cat <<EOF | kubectl create -f -
apiVersion: certificates.k8s.io/v1beta1
kind: CertificateSigningRequest
metadata:
name: mutating-test
spec:
groups:
- system:authenticated
request: $(cat server.csr | base64 | tr -d '\n')
usages:
- digital signature
- key encipherment
- server auth
EOF

认证csr:

kubectl certificate approve mutating-test

认证完成可以查看:

$ kubectl get csr
NAME AGE SIGNERNAME REQUESTOR CONDITION
mutating-test 1m kubernetes.io/legacy-unknown user-f9mr4 Approved,Issued

生成证书:

kubectl get csr mutating-test -o jsonpath='{.status.certificate}' | openssl base64 -d -A -out cert.pem

获取api-server的CA证书:

kubectl get configmap -n kube-system extension-apiserver-authentication -o=jsonpath='{.data.client-ca-file}' | base64 | tr -d '\n'

将这个证书填入 Webhook 的 caBundle。

参考文献

shikanon wechat
欢迎您扫一扫,订阅我滴↑↑↑的微信公众号!