利用 CRD 實現一個 mini-k8s-proxy

前言

=====

進入正文前，需要熟悉以下幾個概念定義：

Custom Resources： 資源（Resource）是 Kubernetes API 中的一個端點，其中存儲的是某個類別的 API 對象（Objects）的一個集合。例如內置的 pods 資源包含一組 Pod 對象。定製資源（Custom Resources）是對 Kubernetes API 的擴展，可以動態的在集羣內安裝刪除，對用戶而言，如同使用 pods 這種內置資源方式一樣，可以使用 kubectl 來創建和訪問定製資源裏面的定製對象（Custom Objects）。
CustomResourceDefinition（CRD）： CRD 定製資源定義是 Kubernetes 提供的向集羣中添加定製資源（Custom Resources）的一種方式，允許用戶定義定製資源。
Custom Controllers： 定製資源只能存取結構化的數據，而定製控制器（Custom Controllers）可以通過觀測分析將結構化的數據解釋爲用戶所期望狀態的記錄，並持續地維護該狀態。它可以用於任何類別的資源，若定製資源與定製控制器相結合，定製資源就能夠提供真正的聲明式 API（Declarative API）。例如，Traefik 的 IngressRoute，apache apisix 的 apisix-ingress-controller ，以及本文接下來的示例等，都是通過編寫 CRD 創建定製資源 + 使用 k8s.io/code-generator 實現定製控制器完成的。
Operator： Operator 是 Kubernetes 的擴展軟件，可以理解爲一種特殊的 Controller ，旨在捕獲（正在管理一個或一組服務的）運維人員的關鍵目標。原理和 Controller 一樣，都是通過編寫 CRD 創建定製資源然後監聽 CRD 相關變化並作出響應。區別就在於 Operator 可能會結合原生的 Controller（Deployment/Replicas），或者 Kubernetes 的網絡，存儲等來控制應用程序的狀態。通俗點講就是，Operator 是一個功能更強大，封裝了對於特定應用程序的運維經驗的 Controller 。例如，etcd operator，prometheus operator 等。

如果覺得上面的定義太枯燥了，那就由我來簡單的總結一遍吧：

Kubernetes 裏面呢，有很多內置的資源，資源中包括對象，例如 pods 資源包含着一組 Pod 對象
Kubernetes 爲了方便用戶，還允許用戶創建自定義資源，用戶創建出來的資源就取了個高大上的名字叫定製資源，同理，定製資源裏有定製對象，例如 Traefik 裏面的 ingressroutes 定製資源中包含着一組 IngressRoute 對象
用戶想要創建定製資源，可以通過編寫 CustomResourceDefinition（CRD，定製資源定義）來實現
用戶創建完定製資源後，還可以編寫代碼來監聽資源的創建、刪除、更新動作，這就是一個 Controller 控制器了
用戶編寫的控制器如果是專注於某個特定的應用程序，並賦予了特定的領域知識，就形成了一個 Operator

目標

實現一個可以通過配置 host 攔截到匹配的請求域名，將流量代理轉發到具體的 service 中（通過配置 serviceName，namespace，port，scheme）的極簡網絡代理工具。其中，配置通過 CRD 創建，代理程序可以通過控制器監聽配置變化，動態更新，無需重啓。（PS：其實就是簡單模擬了 Traefik IngressRoute 的實現）

創建 CRD

將下面的 CustomResourceDefinition 保存爲 crd.yaml 文件

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  # 名字必需與下面的 spec 字段匹配，並且格式爲 '<名稱的複數形式>.<組名>'
  name: proxyroutes.miniproxy.togettoyou.com
spec:
  # 組名
  group: miniproxy.togettoyou.com
  names:
    # kind 通常是單數形式的駝峯編碼（CamelCased）形式。你的資源清單會使用這一形式。
    kind: ProxyRoute
    # shortNames 允許你在命令行使用較短的字符串來匹配資源
    shortNames:
      - pr
    # 名稱的複數形式，用於 URL：/apis/<組>/<版本>/<名稱的複數形式>
    plural: proxyroutes
    # 名稱的單數形式，作爲命令行使用時和顯示時的別名
    singular: proxyroute
  # 可以是 Namespaced 或 Cluster
  scope: Namespaced
  # 列舉此 CustomResourceDefinition 所支持的版本
  versions:
    - name: v1alpha1
      # 每個版本都可以通過 served 標誌來獨立啓用或禁止
      served: true
      # 其中一個且只有一個版本必需被標記爲存儲版本
      storage: true
      # schema 是必需字段
      schema:
        # openAPIV3Schema 是用來檢查定製對象的模式定義
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                # 定義我們需要的幾個配置項
                host:
                  type: string
                serviceName:
                  type: string
                namespace:
                  type: string
                port:
                  type: integer
                scheme:
                  type: boolean

之後創建它：

kubectl apply -f crd.yaml

現在，就可以創建定製對象啦，將下面的 YAML 保存爲 example.yaml

apiVersion: miniproxy.togettoyou.com/v1alpha1
kind: ProxyRoute
metadata:
  name: example-proxyroute
spec:
  # 監聽域名
  host: whoami.togettoyou.com
  # 假設你有一個 whomai 的 service，位於 default 命名空間，容器內部端口爲 80 ，http 協議
  serviceName: whoami
  namespace: default
  port: 80
  scheme: false

並執行創建命令：

kubectl apply -f example.yaml

結合上文的定義介紹，複習一遍，在這裏 proxyroutes 就是我們通過 CRD 創建的定製資源，其中包含着一組 ProxyRoute 對象。

現在可以使用 kubectl 來查看我們剛纔創建的定製對象：

$ kubectl get proxyroute
NAME                 AGE
example-proxyroute   49s

$ kubectl get pr
NAME                 AGE
example-proxyroute   50s

實現控制器

創建項目目錄如下：

├─pkg
│  └─apis
│      └─miniproxy
│          └─v1alpha1
│              └─doc.go
│              └─register.go
│              └─types.go
│          └─register.go
├─script
│  └─boilerplate.go.txt
│  └─code-gen.sh
│  └─codegen.Dockerfile

doc.go 代碼：

// +k8s:deepcopy-gen=package
// +groupName=miniproxy.togettoyou.com

// Package v1alpha1 is the v1alpha1 version of the API.
package v1alpha1

register.go 代碼：

package v1alpha1

import (
 "mini-k8s-proxy/pkg/apis/miniproxy"

 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/apimachinery/pkg/runtime"
 "k8s.io/apimachinery/pkg/runtime/schema"
)

// SchemeGroupVersion is group version used to register these objects
var SchemeGroupVersion = schema.GroupVersion{Group: miniproxy.GroupName, Version: "v1alpha1"}

var (
 // SchemeBuilder initializes a scheme builder
 SchemeBuilder = runtime.NewSchemeBuilder(addKnownTypes)
 // AddToScheme is a global function that registers this API group & version to a scheme
 AddToScheme = SchemeBuilder.AddToScheme
)

// Kind takes an unqualified kind and returns back a Group qualified GroupKind
func Kind(kind string) schema.GroupKind {
 return SchemeGroupVersion.WithKind(kind).GroupKind()
}

// Resource takes an unqualified resource and returns a Group qualified GroupResource
func Resource(resource string) schema.GroupResource {
 return SchemeGroupVersion.WithResource(resource).GroupResource()
}

// Adds the list of known types to Scheme.
func addKnownTypes(scheme *runtime.Scheme) error {
 scheme.AddKnownTypes(SchemeGroupVersion,
  // 主要在這裏導入我們的定製資源對象
  &ProxyRoute{},
  &ProxyRouteList{},
 )
 metav1.AddToGroupVersion(scheme, SchemeGroupVersion)
 return nil
}

types.go 代碼：

package v1alpha1

import (
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// +genclient
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// ProxyRoute is a specification for a ProxyRoute resource
type ProxyRoute struct {
 metav1.TypeMeta   `json:",inline"`
 metav1.ObjectMeta `json:"metadata,omitempty"`

 Spec ProxyRouteSpec `json:"spec"`
}

// ProxyRouteSpec is the spec for a ProxyRoute resource
type ProxyRouteSpec struct {
 Host        string `json:"host"`
 ServiceName string `json:"serviceName"`
 Namespace   string `json:"namespace,omitempty"`
 Port        int32  `json:"port,omitempty"`
 Scheme      bool   `json:"scheme,omitempty"`
}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// ProxyRouteList is a list of ProxyRoute resources
type ProxyRouteList struct {
 metav1.TypeMeta `json:",inline"`
 metav1.ListMeta `json:"metadata"`

 Items []ProxyRoute `json:"items"`
}

register.go 代碼：

package miniproxy

// GroupName is the group name used in this package
const (
 GroupName = "miniproxy.togettoyou.com"
)

代碼編寫完成後，就可以使用 k8s.io/code-generator 來生成控制器相關代碼了，腳本定義在 script 文件夾下，其中 boilerplate.go.txt 爲生成的代碼頭部協議註釋，codegen.Dockerfile 內容爲：

FROM golang:1.16

ARG KUBE_VERSION

ENV GO111MODULE=on
ENV GOPROXY https://goproxy.cn,direct

RUN go get k8s.io/code-generator@$KUBE_VERSION; exit 0
RUN go get k8s.io/apimachinery@$KUBE_VERSION; exit 0

RUN mkdir -p $GOPATH/src/k8s.io/{code-generator,apimachinery}
RUN cp -R $GOPATH/pkg/mod/k8s.io/code-generator@$KUBE_VERSION $GOPATH/src/k8s.io/code-generator
RUN cp -R $GOPATH/pkg/mod/k8s.io/apimachinery@$KUBE_VERSION $GOPATH/src/k8s.io/apimachinery
RUN chmod +x $GOPATH/src/k8s.io/code-generator/generate-groups.sh

WORKDIR $GOPATH/src/k8s.io/code-generator

code-gen.sh 腳本內容如下：

#!/bin/bash -e

set -e -o pipefail

PROJECT_MODULE="mini-k8s-proxy"
IMAGE_

echo "Building codegen Docker image..."
docker build --build-arg KUBE_VERSION=v0.20.2 -f "./script/codegen.Dockerfile" \
            -t "${IMAGE_NAME}" \
            "."

cmd="/go/src/k8s.io/code-generator/generate-groups.sh all \
    ${PROJECT_MODULE}/pkg/generated \
    ${PROJECT_MODULE}/pkg/apis \
    miniproxy:v1alpha1 \
    --go-header-file=/go/src/${PROJECT_MODULE}/script/boilerplate.go.txt"

echo "Generating clientSet code ..."
docker run --rm \
           -v "$(pwd):/go/src/${PROJECT_MODULE}" \
           -w "/go/src/${PROJECT_MODULE}" \
           "${IMAGE_NAME}" $cmd

執行腳本生成相關代碼：

$ ./script/code-gen.sh

......
Generating clientSet code ...
Generating deepcopy funcs
Generating clientset for miniproxy:v1alpha1 at mini-k8s-proxy/pkg/generated/clientset
Generating listers for miniproxy:v1alpha1 at mini-k8s-proxy/pkg/generated/listers
Generating informers for miniproxy:v1alpha1 at mini-k8s-proxy/pkg/generated/informers

實現業務邏輯

由於業務較簡單，我們直接在 main.go 完成業務邏輯，貼上代碼：

type ProxyRouteSpec struct {
 V map[string]v1alpha1.ProxyRouteSpec
 sync.RWMutex
}

var prs = &ProxyRouteSpec{
 V: make(map[string]v1alpha1.ProxyRouteSpec, 0),
}

type resourceEventHandler struct {
 Ev chan<- interface{}
}

func (reh *resourceEventHandler) OnAdd(obj interface{}) {
 eventHandlerFunc(reh.Ev, obj)
}

func (reh *resourceEventHandler) OnUpdate(oldObj, newObj interface{}) {
 eventHandlerFunc(reh.Ev, newObj)
}

func (reh *resourceEventHandler) OnDelete(obj interface{}) {
 eventHandlerFunc(reh.Ev, obj)
}

func eventHandlerFunc(events chan<- interface{}, obj interface{}) {
 select {
 case events <- obj:
 default:
 }
}

func main() {
 ctx := context.Background()

 eventCh := make(chan interface{}, 1)
 // 採用緩衝大小爲 1 的通道方式來處理 CRD 事件
 eventHandler := &resourceEventHandler{Ev: eventCh}

 // 作爲測試，可以直接使用 kubeconfig 連接 k8s，實際部署使用 InClusterConfig 模式
 //cfg, err := clientcmd.BuildConfigFromFlags("", "tmp/config")
 cfg, err := rest.InClusterConfig()
 if err != nil {
  panic(err)
 }
 client, err := clientset.NewForConfig(cfg)
 if err != nil {
  panic(err)
 }
 // 構建 k8s Crd Informer 實例
 factoryCrd := externalversions.NewSharedInformerFactoryWithOptions(
  client,
  10*time.Minute,
 )
 // 註冊 Informer 事件處理
 factoryCrd.Miniproxy().V1alpha1().ProxyRoutes().Informer().AddEventHandler(eventHandler)
 // 啓動 Informer
 factoryCrd.Start(ctx.Done())
 // 等待首次緩存同步
 for t, ok := range factoryCrd.WaitForCacheSync(ctx.Done()) {
  if !ok {
   panic(fmt.Errorf("timed out waiting for controller caches to sync %s", t.String()))
  }
 }
 go startServer()

 for {
  select {
  case _, ok := <-eventCh:
   if !ok {
    continue
   }
   // 從 Lister 緩存獲取 CRD 資源對象
   proxyRoutes, err := factoryCrd.Miniproxy().V1alpha1().ProxyRoutes().Lister().List(labels.Everything())
   if err != nil {
    log.Println(err.Error())
    continue
   }
   // 清空本地緩存並重新放入
   prs.Lock()
   prs.V = make(map[string]v1alpha1.ProxyRouteSpec, 0)
   for _, proxyRoute := range proxyRoutes {
    fmt.Printf("%+v\n", proxyRoute)
    prs.V[proxyRoute.Spec.Host] = proxyRoute.Spec
   }
   prs.Unlock()
  }
 }
}

原理比較粗暴，通過 Informer — Lister 機制監聽 CRD 資源的變化，並將資源對象存入本地 map 緩存中。

繼續添加代理轉發邏輯：

func startServer() {
 gin.SetMode(gin.ReleaseMode)
 r := gin.Default()
 r.Any("/*any", handler)
 log.Fatalln(r.Run(":80"))
}

func handler(c *gin.Context) {
 prs.RLock()
 defer prs.RUnlock()
 if proxyRouteSpec, ok := prs.V[c.Request.Host]; ok {
  u := ""
  if proxyRouteSpec.Scheme {
   u += "https://"
  } else {
   u += "http://"
  }
  if proxyRouteSpec.Namespace != "" {
   u += proxyRouteSpec.ServiceName + "." + proxyRouteSpec.Namespace
  } else {
   u += proxyRouteSpec.ServiceName
  }
  if proxyRouteSpec.Port != 0 {
   u += fmt.Sprintf(":%d", proxyRouteSpec.Port)
  }
  log.Println("代理地址: ", u)
  proxyUrl, err := url.Parse(u)
  if err != nil {
   c.AbortWithStatus(http.StatusInternalServerError)
  }
  proxyServer(c, proxyUrl)
 } else {
  c.String(http.StatusNotFound, "404")
 }
}

// 代理轉發
func proxyServer(c *gin.Context, proxyUrl *url.URL) {
 proxy := &httputil.ReverseProxy{
  Director: func(outReq *http.Request) {
   u := outReq.URL
   outReq.URL = proxyUrl
   if outReq.RequestURI != "" {
    parsedURL, err := url.ParseRequestURI(outReq.RequestURI)
    if err == nil {
     u = parsedURL
    }
   }

   outReq.URL.Path = u.Path
   outReq.URL.RawPath = u.RawPath
   outReq.URL.RawQuery = u.RawQuery
   outReq.RequestURI = "" // Outgoing request should not have RequestURI

   outReq.Proto = "HTTP/1.1"
   outReq.ProtoMajor = 1
   outReq.ProtoMinor = 1

   if _, ok := outReq.Header["User-Agent"]; !ok {
    outReq.Header.Set("User-Agent", "")
   }

   // Even if the websocket RFC says that headers should be case-insensitive,
   // some servers need Sec-WebSocket-Key, Sec-WebSocket-Extensions, Sec-WebSocket-Accept,
   // Sec-WebSocket-Protocol and Sec-WebSocket-Version to be case-sensitive.
   // https://tools.ietf.org/html/rfc6455#page-20
   outReq.Header["Sec-WebSocket-Key"] = outReq.Header["Sec-Websocket-Key"]
   outReq.Header["Sec-WebSocket-Extensions"] = outReq.Header["Sec-Websocket-Extensions"]
   outReq.Header["Sec-WebSocket-Accept"] = outReq.Header["Sec-Websocket-Accept"]
   outReq.Header["Sec-WebSocket-Protocol"] = outReq.Header["Sec-Websocket-Protocol"]
   outReq.Header["Sec-WebSocket-Version"] = outReq.Header["Sec-Websocket-Version"]
   delete(outReq.Header, "Sec-Websocket-Key")
   delete(outReq.Header, "Sec-Websocket-Extensions")
   delete(outReq.Header, "Sec-Websocket-Accept")
   delete(outReq.Header, "Sec-Websocket-Protocol")
   delete(outReq.Header, "Sec-Websocket-Version")
  },
  Transport: &http.Transport{
   TLSClientConfig: &tls.Config{
    InsecureSkipVerify: true,
   },
  },
  ErrorHandler: func(w http.ResponseWriter, request *http.Request, err error) {
   statusCode := http.StatusInternalServerError
   w.WriteHeader(statusCode)
   w.Write([]byte(http.StatusText(statusCode)))
  },
 }
 proxy.ServeHTTP(c.Writer, c.Request)
}

每次請求連接會從本地緩存讀取配置，判斷是否匹配，若匹配則轉發代理到配置的服務中去。

部署

爲了方便測試，我已經編譯好鏡像上傳到 Docker Hub 上，所以大家可以直接使用下面的 yaml 部署：

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: proxyroutes.miniproxy.togettoyou.com
spec:
  group: miniproxy.togettoyou.com
  names:
    kind: ProxyRoute
    shortNames:
      - pr
    plural: proxyroutes
    singular: proxyroute
  scope: Namespaced
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                host:
                  type: string
                serviceName:
                  type: string
                namespace:
                  type: string
                port:
                  type: integer
                scheme:
                  type: boolean
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mini-k8s-proxy
spec:
  selector:
    matchLabels:
      app: mini-k8s-proxy
  replicas: 1
  template:
    metadata:
      labels:
        app: mini-k8s-proxy
    spec:
      containers:
        - name: mini-k8s-proxy
          image: togettoyou/mini-k8s-proxy:latest
          ports:
            - containerPort: 80

---
apiVersion: v1
kind: Service
metadata:
  name: mini-k8s-proxy-service
spec:
  ports:
    - port: 80
      targetPort: 80
  selector:
    app: mini-k8s-proxy
  type: NodePort

部署：

$ kubectl apply -f mini-k8s-proxy.yaml
customresourcedefinition.apiextensions.k8s.io/proxyroutes.miniproxy.togettoyou.com created
deployment.apps/mini-k8s-proxy created
service/mini-k8s-proxy-service created

$ kubectl get svc
NAME                     TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)        AGE
mini-k8s-proxy-service   NodePort    10.111.139.14   <none>        80:32112/TCP   32s

訪問集羣域名: 32112 ，看到 404 的話，恭喜部署成功。

驗證使用 ProxyRoute

現在我有一個名稱爲 test-service 的 service 處於 testns 命名空間下，容器內部端口爲 80（是一個 nginx 服務）

$ kubectl get svc -n testns
NAME           TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)   AGE
test-service   ClusterIP   10.97.89.250   <none>        80/TCP    2s

我想要當請求 host 爲 test.togettoyou.com:32112 的流量請求過來時，可以代理轉發到 test-service 上，怎麼做呢

按照心裏預期，創建一個 ProxyRoute 資源對象：

apiVersion: miniproxy.togettoyou.com/v1alpha1
kind: ProxyRoute
metadata:
  name: test-proxyroute
spec:
  host: test.togettoyou.com:32112
  serviceName: test-service
  namespace: testns
  port: 80
  scheme: false

瀏覽器訪問 test.togettoyou.com:32112 ，神奇的事情就會發生了

總結

本文通過 CRD + Controller 實現了一個簡易的 K8S 代理轉發工具，相關代碼均上傳到了 Github（https://github.com/togettoyou/mini-k8s-proxy）

實現思路來自 Traefik，強烈推薦

最後，感謝您的閱讀！

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/SXF8OX_i4FBqBI2BZCfaoQ

目標

創建 CRD

實現控制器

實現業務邏輯

部署

驗證使用 ProxyRoute

總結

猜你喜歡