如何從零開始編寫一個 Kubernetes CRD

做者：宋欣建（晝夢），螞蟻金服中間件團隊開發工程師

本文首先向你簡單介紹了 Kubernetes，而後教你從零開始構建一個 Kubernetes CRD。若是你已經對 Kubernetes 十分了解的話能夠跳過本文前半部分的 Kubernetes 介紹，直接從 Controller 部分開始閱讀。

快速入門Kubernetes

Kubernetes是一個容器管理系統。

具體功能：

• 基於容器的應用部署、維護和滾動升級

• 負載均衡和服務發現

• 跨機器和跨地區的集羣調度

• 自動伸縮

• 無狀態服務和有狀態服務

• 普遍的 Volume 支持

• 插件機制保證擴展性

經過閱讀Kubernetes指南和Kubernetes HandBook以及官方文檔 或者 閱讀 Kubernetes權威指南能夠得到更好的學習體驗。

在開始安裝Kubernetes以前，咱們須要知道：

一、Docker與Kubernetes

Docker是一個容器運行時的實現，Kubernetes依賴於某種容器運行時的實現。

二、Pod

Kubernetes中最基本的調度單位是Pod，Pod從屬於Node（物理機或虛擬機），Pod中能夠運行多個Docker容器，會共享 PID、IPC、Network 和 UTS namespace。Pod在建立時會被分配一個IP地址，Pod間的容器能夠互相通訊。

三、Yaml

Kubernetes中有着不少概念，它們都算作是一種對象，如Pod、Deployment、Service等，均可以經過一個yaml文件來進行描述，並能夠對這些對象進行CRUD操做（對應REST中的各類HTTP方法）。

下面一個Pod的yaml文件示例：

apiVersion: v1
  kind: Pod
  metadata:
    name: my-nginx-app
    labels:
      app: my-nginx-app
  spec:
    containers:
    - name: nginx
      image: nginx:1.7.9
      ports:
      - containerPort: 80複製代碼

kind：對象的類別

metadata：元數據，如Pod的名稱，以及標籤Label【用於識別一系列關聯的Pod，可使用 Label Selector 來選擇一組相同 label 的對象】

spec：但願Pod能達到的狀態，在此體現了Kubernetes的聲明式的思想，咱們只須要定義出指望達到的狀態，而不須要關心如何達到這個狀態，這部分工做由Kubernetes來完成。這裏咱們定義了Pod中運行的容器列表，包括一個nginx容器，該容器對外暴露了80端口。

四、Node

Node 是 Pod 真正運行的主機，能夠是物理機，也能夠是虛擬機。爲了管理 Pod，每一個 Node 節點上至少要運行 container runtime、kubelet 和 kube-proxy 服務。

五、Deployment

Deployment用於管理一個無狀態應用，對應一個Pod的集羣，每一個Pod的地位是對等的，對Deployment來講只是用於維護必定數量的Pod，這些Pod有着相同的Pod模板。

 apiVersion: apps/v1
 kind: Deployment
 metadata:
 name: my-nginx-app
 spec:
 replicas: 3
 selector:
 matchLabels:
 app: my-nginx-app
 template:
 metadata:
 labels:
 app: my-nginx-app
 spec:
 containers:
 - name: nginx
 image: nginx:1.7.9
 ports:
 - containerPort: 80複製代碼

能夠對Deployment進行部署、升級、擴縮容等操做。

六、Service

Service用於將一組Pod暴露爲一個服務。

在 kubernetes 中，Pod 的 IP 地址會隨着 Pod 的重啓而變化，並不建議直接拿 Pod 的 IP 來交互。那如何來訪問這些 Pod 提供的服務呢？使用 Service。Service 爲一組 Pod（經過 labels 來選擇）提供一個統一的入口，併爲它們提供負載均衡和自動服務發現。

 apiVersion: v1
 kind: Service
 metadata:
 name: my-nginx-app
 labels:
 name: my-nginx-app
 spec:
 type: NodePort      #這裏表明是NodePort類型的
 ports:
 - port: 80          # 這裏的端口和clusterIP(10.97.114.36)對應，即10.97.114.36:80,供內部訪問。
 targetPort: 80    # 端口必定要和container暴露出來的端口對應
 protocol: TCP
 nodePort: 32143   # 每一個Node會開啓，此端口供外部調用。
 selector:
 app: my-nginx-app複製代碼

七、Kubernetes組件

• etcd 保存了整個集羣的狀態；

• apiserver 提供了資源操做的惟一入口，並提供認證、受權、訪問控制、API 註冊和發現等機制；

• controller manager 負責維護集羣的狀態，好比故障檢測、自動擴展、滾動更新等；

• scheduler 負責資源的調度，按照預約的調度策略將 Pod 調度到相應的機器上；

• kubelet 負責維護容器的生命週期，同時也負責 Volume（CVI）和網絡（CNI）的管理；

• Container runtime 負責鏡像管理以及 Pod 和容器的真正運行（CRI）；

• kube-proxy 負責爲 Service 提供 cluster 內部的服務發現和負載均衡

安裝Kubernetes【Minikube】

minikube爲開發或者測試在本地啓動一個節點的kubernetes集羣，minikube打包了和配置一個linux虛擬機、docker與kubernetes組件。

Kubernetes集羣是由Master和Node組成的，Master用於進行集羣管理，Node用於運行Pod等workload。而minikube是一個Kubernetes集羣的最小集。

一、安裝virtualbox

www.virtualbox.org/wiki/Downlo…

二、安裝minikube

curl -Lo minikube https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64 && chmod +x minikube && sudo mv minikube /usr/local/bin/複製代碼

三、啓用dashboard（web console）【可選】

minikube addons enable dashboard複製代碼

開啓dashboard

四、啓動minikube

minikube start

start以後能夠經過minikube status來查看狀態，若是minikube和cluster都是Running，則說明啓動成功。

五、查看啓動狀態

kubectl get pods

kubectl體驗【以一個Deployment爲例】

kubectl是一個命令行工具，用於向API Server發送指令。

咱們以部署、升級、擴縮容一個Deployment、發佈一個Service爲例體驗一下Kubernetes。

命令的一般格式爲：

kubectl

object_type(單數or複數) $object_name other params

• operation如get,replace,create,expose,delete等。

• object_type是操做的對象類型，如pods,deployments,services

• object_name是對象的name

• 後面能夠加一些其餘參數

kubectl命令表：

kubernetes.io/docs/refere…

docs.kubernetes.org.cn/490.html

Deployment的文檔：

kubernetes.io/docs/concep…

kubernetes.feisky.xyz/he-xin-yuan…

一、建立一個Deployment

可使用kubectl run來運行，也能夠基於現有的yaml文件來create。

kubectl run --image=nginx:1.7.9 nginx-app --port=80

或者

kubectl create -f my-nginx-deployment.yaml

my-nginx-deployment是下面這個yaml文件的名稱

 apiVersion: apps/v1
 kind: Deployment
 metadata:
 name: my-nginx-app
 spec:
 replicas: 3
 selector:
 matchLabels:
 app: my-nginx-app
 template:
 metadata:
 labels:
 app: my-nginx-app
 spec:
 containers:
 - name: nginx
 image: nginx:1.7.9
 ports:
 - containerPort: 80複製代碼

而後能夠經過kubectl get pods 來查看建立好了的3個Pod。

Pod的名稱是以所屬Deployment名稱爲前綴，後面加上惟一標識。

再經過kubectl get deployments來查看建立好了的deployment。

這裏有4列，分別是：

• DESIRED：Pod副本數量的指望值，即Deployment裏面定義的replicas

• CURRENT：當前Replicas的值

• UP-TO-DATE：最新版本的Pod的副本梳理，用於指示在滾動升級的過程當中，有多少個Pod副本已經成功升級

• AVAILABLE：集羣中當前存活的Pod數量

二、刪除掉任意一個Pod

Deployment自身擁有副本保持機制，會始終將其所管理的Pod數量保持爲spec中定義的replicas數量。

kubectl delete pods $pod_name

能夠看出被刪掉的Pod的關閉與代替它的Pod的啓動過程。

三、縮擴容

縮擴容有兩種實現方式，一種是修改yaml文件，將replicas修改成新的值，而後kubectl replace -f my-nginx-deployment.yaml；

另外一種是使用scale命令：kubectl scale deployment $deployment_name --replicas=5

四、更新

更新也是有兩種實現方式，Kubernetes的升級能夠實現無縫升級，即不須要進行停機。

一種是rolling-update方式，重建Pod，edit/replace/set image等都可以實現。好比說咱們能夠修改yaml文件，而後kubectl replace -f my-nginx-deployment.yaml，也能夠kubectl set image

resource_name $container_name=nginx:1.9.1

另外一種是patch方式，patch不會去重建Pod，Pod的IP能夠保持。

kubectl get pods -o yaml能夠以yaml的格式來查看Pod

這裏能夠看出容器的版本已經被更新到了1.9.1。

五、暴露服務

暴露服務也有兩種實現方式，一種是經過kubectl create -f my-nginx-service.yaml能夠建立一個服務：

 apiVersion: v1
 kind: Service
 metadata:
 name: my-nginx-app
 labels:
 name: my-nginx-app
 spec:
 type: NodePort      #這裏表明是NodePort類型的
 ports:
 - port: 80          # 這裏的端口和clusterIP(10.97.114.36)對應，即10.97.114.36:80,供內部訪問。
 targetPort: 80    # 端口必定要和container暴露出來的端口對應
 protocol: TCP
 nodePort: 32143   # 每一個Node會開啓，此端口供外部調用。
 selector:
 app: my-nginx-app複製代碼

ports中有三個端口，第一個port是Pod供內部訪問暴露的端口，第二個targetPort是Pod的Container中配置的containerPort，第三個nodePort是供外部調用的端口。

另外一種是經過kubectl expose命令實現。

minikube ip返回的就是minikube所管理的Kubernetes集羣所在的虛擬機ip。

minikube service my-nginx-app --url也能夠返回指定service的訪問URL。

CRD【CustomResourceDefinition】

CRD是Kubernetes爲提升可擴展性，讓開發者去自定義資源（如Deployment，StatefulSet等）的一種方法。

Operator=CRD+Controller。

CRD僅僅是資源的定義，而Controller能夠去監聽CRD的CRUD事件來添加自定義業務邏輯。

關於CRD有一些連接先貼出來：

• 官方文檔：kubernetes.io/docs/tasks/…

• CRD Yaml的Schema：kubernetes.io/docs/refere…

• kubernetes.feisky.xyz/cha-jian-ku…

• sq.163yun.com/blog/articl…

• book.kubebuilder.io/

若是說只是對CRD實例進行CRUD的話，不須要Controller也是能夠實現的，只是只有數據，沒有針對數據的操做。

一個CRD的yaml文件示例：

 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition
 metadata:
    # name must match the spec fields below, and be in the form: <plural>.<group>
 name: crontabs.stable.example.com
 spec:
    # group name to use for REST API: /apis/<group>/<version>
 group: stable.example.com
    # list of versions supported by this CustomResourceDefinition
 version: v1beta1
    # either Namespaced or Cluster
 scope: Namespaced
 names:
      # plural name to be used in the URL: /apis/<group>/<version>/<plural>
 plural: crontabs
      # singular name to be used as an alias on the CLI and for display
 singular: crontab
      # kind is normally the CamelCased singular type. Your resource manifests use this.
 kind: CronTab
      # shortNames allow shorter string to match your resource on the CLI
 shortNames:
 - ct複製代碼

經過kubectl create -f crd.yaml能夠建立一個CRD。

kubectl get CustomResourceDefinitions能夠獲取建立的全部CRD。

而後能夠經過kubectl create -f my-crontab.yaml能夠建立一個CRD的實例：

 apiVersion: "stable.example.com/v1beta1"
 kind: CronTab
 metadata:
 name: my-new-cron-object
 spec:
 cronSpec: "* * * * */5"
 image: my-awesome-cron-image複製代碼

Controller【Fabric8】

如何去實現一個Controller呢？

可使用Go來實現，而且不管是參考資料仍是開源支持都很是好，推薦有Go語言基礎的優先考慮用client-go來做爲Kubernetes的客戶端，用KubeBuilder來生成骨架代碼。一個官方的Controller示例項目是sample-controller。

對於Java來講，目前Kubernetes的JavaClient有兩個，一個是Jasery，另外一個是Fabric8。後者要更好用一些，由於對Pod、Deployment都有DSL定義，並且構建對象是以Builder模式作的，寫起來比較舒服。

Fabric8的資料目前只有github.com/fabric8io/k…，注意看目錄下的examples。

這些客戶端本質上都是經過REST接口來與Kubernetes API Server通訊的。

Controller的邏輯實際上是很簡單的：監聽CRD實例（以及關聯的資源）的CRUD事件，而後執行相應的業務邏輯

MyDeployment

基於Fabric8來開發一個較爲完整的Controller的示例目前我在網絡上是找不到的，下面MyController的實現也是一步步摸索出來的，不免會有各類問題=.=，歡迎大佬們捉蟲。

用例

代碼在 github.com/songxinjian…

MyDeployment是用來模擬Kubernetes提供的Deployment的簡化版實現，目前能夠作到如下功能：

• 啓動後自動建立出一個MyDeployment的CRD

  • 【觸發】啓動應用

  • 【指望】能夠看到建立出來的CRD

  • 【測試】kubectl get CustomResourceDefinition -o yaml

• 建立一個MyDeployment: Nginx實例

  • 【觸發】kubectl create -f my-deployment-instance.yaml

  • 【指望】能夠看到級聯建立出來的3個pod

  • 【測試】kubectl get pods

• 手工刪掉一個pod

  • 【觸發】kubectl delete pods $pod_name

  • 【指望】pod被重建

  • 【測試】kubectl get pods -w

• 暴露一個服務

  • 【觸發】kubectl create -f my-deployment-service.yaml

  • 【指望】能夠經過curl來訪問nginx服務

  • 【測試】minikube service my-nginx-app --url 而後 curl

• 更新鏡像

  • 【觸發】kubectl replace -f my-deployment-instance-update-image-1.9.1.yaml

  • 【指望】pod的nginx版本被更新爲1.9.1

  • 【測試】kubectl get pods -o yaml

• 擴容

  • 【觸發】kubectl replace -f my-deployment-instance-update-scaleup-1.9.1.yaml

  • 【指望】pod被擴容到5個

  • 【測試】kubectl get pods

• 縮容

  • 【觸發】kubectl replace -f my-deployment-instance-update-scaledown-1.9.1.yaml

  • 【指望】pod被縮容到2個

  • 【測試】kubectl get pods

• 擴容並更新鏡像

  • 【觸發】kubectl replace -f my-deployment-instance-update-image-and-scaleup-1.14.yaml

  • 【指望】pod被擴容5個，且nginx版本被更新爲1.14

  • 【測試】kubectl get pods 而後 kubectl get pods -o yaml

• 刪除一個MyDeployment

  • 【觸發】kubectl delete mydeployments my-nginx-app

  • 【指望】MyDeployment被刪掉，而且關聯的pod也被級聯刪掉

  • 【測試】kubectl get mydeployments 而後 kubectl get pods

此外還有一些功能還沒有開發，其中狀態是很是重要的，很惋惜時間不夠沒有開發完成：

• 查看狀態(TODO)

• 回滾(TODO)

• 狀態更新【current，update-to-date，available】(TODO)

• describe EVENTS(TODO)

運行

一、搭建好上面的minikube環境後

二、拉下deployment-controller的代碼，是一個SpringBoot工程。

三、啓動kube-proxy

kubectl proxy --port=12000

這一步是爲了繞開Kubernetes API Server的權限認證。開啓proxy以後就能夠經過localhost:12000來鏈接Kubernetes了。

四、運行DeploymentControllerApplication的main方法便可。

五、此時能夠根據上述用例來進行測試。

MyDeployment實現

項目工程結構

CRD定義

按照Fabric8的邏輯，定義一個CRD須要至少定義如下內容：

• CustomResourceDefinition，須要繼承CustomResource，CRD資源定義

• CustomResourceDefinitionList，須要繼承CustomResourceList，CRD資源列表

• CustomResourceDefinitionSpec，須要實現KubernetesResource接口，CRD資源的Spec

• DoneableCustomResourceDefinition，須要繼承CustomResourceDoneable，CRD資源的修改Builder

• 【可選】CustomResourceDefinitionStatus（須要說明一點的是，CRD支持使用SubResource，包括scale和status，在1.11+以後能夠直接使用，在1.10中須要修改API Server的啓動參數來啓用；minikube的最新版本是能夠支持到Kubernetes的1.10的）

在CRD定義中一般是須要持有一個Spec的【注意，上面提到的全部類定義持有的成員變量最好都加一個@JsonProperty註解，不加的話在get資源時對JSON反序列化時用到的名字就是屬性名了】

下面是基於Fabric8的API構建出了一個CRD，以後能夠調用API將其建立到Kubernetes，效果和kubectl create -f 是同樣的。

但Controller須要作到的是啓動時自動建立一個CRD出來，因此用kubectl建立不夠自動化。

public static final String CRD_GROUP = "cloud.alipay.com";
  public static final String CRD_SINGULAR_NAME = "mydeployment";
  public static final String CRD_PLURAL_NAME = "mydeployments";
  public static final String CRD_NAME = CRD_PLURAL_NAME + "." + CRD_GROUP;
  public static final String CRD_KIND = "MyDeployment";
  public static final String CRD_SCOPE = "Namespaced";
  public static final String CRD_SHORT_NAME = "md";
  public static final String CRD_VERSION = "v1beta1";
  public static final String CRD_API_VERSION = "apiextensions.k8s.io/" + CRD_VERSION;    
  ​
  public static CustomResourceDefinition MY_DEPLOYMENT_CRD = new CustomResourceDefinitionBuilder()
              .withApiVersion(CRD_API_VERSION)
              .withNewMetadata()
              .withName(CRD_NAME)
              .endMetadata()
  ​
              .withNewSpec()
              .withGroup(CRD_GROUP)
              .withVersion(CRD_VERSION)
              .withScope(CRD_SCOPE)
              .withNewNames()
              .withKind(CRD_KIND)
              .withShortNames(CRD_SHORT_NAME)
              .withSingular(CRD_SINGULAR_NAME)
              .withPlural(CRD_PLURAL_NAME)
              .endNames()
              .endSpec()
  ​
              .withNewStatus()
              .withNewAcceptedNames()
              .addToShortNames(new String[]{"availableReplicas", "replicas", "updatedReplicas"})
              .endAcceptedNames()
              .endStatus()
              .build();複製代碼

Controller

入口處須要去爲咱們的CRD註冊一個反序列化器。

入口處

static {
      KubernetesDeserializer.registerCustomKind(MyDeployment.CRD_GROUP + "/" + MyDeployment.CRD_VERSION, MyDeployment.CRD_KIND, MyDeployment.class);
  }
  ​
  /** * 入口 */
  public void run() {
      // 建立CRD
      CustomResourceDefinition myDeploymentCrd = createCrdIfNotExists();
      // 監聽Pod的事件
      watchPod();
      // 監聽MyDeployment的事件
      watchMyDeployment(myDeploymentCrd);
  }複製代碼

watchPod是監聽MyDeployment所管理的Pod的CRUD事件。

private void watchPod() {
      delegate.client().pods().watch(new Watcher<Pod>() {
          @Override
          public void eventReceived(Action action, Pod pod) {
              // 若是是被MyDeployment管理的Pod
              if(pod.getMetadata().getOwnerReferences().stream().anyMatch(ownerReference -> ownerReference.getKind().equals(MyDeployment.CRD_KIND))) {
                  unifiedPodWatcher.eventReceived(action, pod);
              }
          }
  ​
          @Override
          public void onClose(KubernetesClientException e) {
              log.error("watching pod {} caught an exception {}", e);
          }
      });
  }複製代碼

UnifiedPodWatcher是處理了全部Pod的事件，而後在收到事件時去通知Pod事件的訂閱者，這裏用到了一個觀察者模式。

@Component
  public class UnifiedPodWatcher {
      @Autowired
      private List<PodAddedWatcher> podAddedWatchers;
      @Autowired
      private List<PodModifiedWatcher> podModifiedWatchers;
      @Autowired
      private List<PodDeletedWatcher> podDeletedWatchers;
  ​
      /** * 將Pod事件統一收到此處 * @param action * @param pod */
      public void eventReceived(Watcher.Action action, Pod pod) {
          log.info("Thread {}: PodWatcher: {} => {}, {}", Thread.currentThread().getId(), action, pod.getMetadata().getName(), pod);
          switch (action) {
              case ADDED:
                  podAddedWatchers.forEach(watcher -> watcher.onPodAdded(pod));
                  break;
              case MODIFIED:
                  podModifiedWatchers.forEach(watcher -> watcher.onPodModified(pod));
                  break;
              case DELETED:
                  podDeletedWatchers.forEach(watcher -> watcher.onPodDeleted(pod));
                  break;
              default:
                  break;
          }
      }
  }複製代碼

Fabric8的watcher是在代碼層面不會限制在多個地方去監聽同一個對象的，但經粗略測試，多處監聽只有在第一個地方會收到回調；從可維護性角度來講，散落在各個地方的watcher代碼也是不夠優雅的。因此將watcher統一收口到一個地方，而後在這個地方下發事件。

而後MyDeployment的監聽也是比較相似的：

private void watchMyDeployment(CustomResourceDefinition myDeploymentCrd) {
      MixedOperation<MyDeployment, MyDeploymentList, DoneableMyDeployment, Resource<MyDeployment, DoneableMyDeployment>> myDeploymentClient = delegate.client().customResources(myDeploymentCrd, MyDeployment.class, MyDeploymentList.class, DoneableMyDeployment.class);
      myDeploymentClient.watch(new Watcher<MyDeployment>() {
          @Override
          public void eventReceived(Action action, MyDeployment myDeployment) {
              log.info("myDeployment: {} => {}" , action , myDeployment);
              if(myDeploymentHandlers.containsKey(MyDeploymentActionHandler.RESOURCE_NAME + action.name())) {
                  myDeploymentHandlers.get(MyDeploymentActionHandler.RESOURCE_NAME + action.name()).handle(myDeployment);
              }
          }
  ​
          @Override
          public void onClose(KubernetesClientException e) {
              log.error("watching myDeployment {} caught an exception {}", e);
          }
      });
  }複製代碼

建立MyDeployment的實現邏輯

建立MyDeployment時會建立出replicas個Pod出來。此處邏輯在MyDeploymentAddedHandler實現。

@Component(value = MyDeploymentActionHandler.RESOURCE_NAME + CrdAction.ADDED)
  @Slf4j
  public class MyDeploymentAddedHandler implements MyDeploymentActionHandler {
      @Autowired
      private KubeClientDelegate delegate;
  ​
      @Override
      public void handle(MyDeployment myDeployment) {
          log.info("{} added", myDeployment.getMetadata().getName());
          // TODO 當第一次啓動項目時，現存的MyDeployment會回調一次Added事件，這裏會致使重複建立pod【可經過status解決】,目前解法是去查一下現存的pod[不可靠]
          // 有可能pod的狀態還沒來得及置爲not ready
          int existedReadyPodNumber = delegate.client().pods().inNamespace(myDeployment.getMetadata().getNamespace()).withLabelSelector(myDeployment.getSpec().getLabelSelector()).list().getItems()
                  .stream().filter(UnifiedPodWatcher::isPodReady).collect(Collectors.toList()).size();
          Integer replicas = myDeployment.getSpec().getReplicas();
          for (int i = 0; i < replicas - existedReadyPodNumber; i++) {
              Pod pod = myDeployment.createPod();
              log.info("Thread {}:creating pod[{}]: {} , {}", Thread.currentThread().getId(), i, pod.getMetadata().getName(), pod);
              delegate.client().pods().create(pod);
         }
      }
  }複製代碼

這裏須要解釋一下爲何建立pod的數量須要減去existedReadyPodNumber。經觀察發現，若是現存有CRD實例，而後啓動Controller，會當即收到CRD的added事件，即便Pod都是健康的，這會致使建立出雙倍的Pod。因此這裏須要判斷一下現存的Pod數量，若是夠了，就不去建立了。

可是這又會引入一個問題，假如我將MyDeployment刪掉了，此時會級聯刪除關聯的Pod，在沒來得及刪掉以前，又去建立一個新的MyDeployment，這時候會發現現存的Pod數量並不是爲0，因此新建的Pod數量就不能達到replicas。解決辦法是去判斷一下Pod的狀態，若是是NotReady，那麼就不算是正常的Pod。

但這種解決思路仍是有問題，在刪掉MyDeployment以後不會當即將Pod狀態置爲NotReady，須要必定時間，在這段時間內若是建立MyDeployment，那麼仍是有可能會出現少建立Pod的狀況。

目前尚未什麼無BUG的思路。

建立一個Pod的實現，將這段代碼放到了MyDeployment中，以表示從屬關係（代碼也好寫一些）。

若是Pod是從屬於某個MyDeployment，那麼咱們應該將OwnerReference傳入；

Pod的name必須以MyDeployment的name爲前綴，後面加上惟一ID；

Pod的spec必須與MyDeployment的spec中的pod template一致；

Pod的labels中包含MyDeployment的label，而且要加上一個pod-template哈希值，以在更新資源時判斷pod template是否改變，若是沒有變化，則不觸發modified事件。

Note: A Deployment’s rollout is triggered if and only if the Deployment’s pod template (that is, .spec.template) is changed, for example if the labels or container images of the template are updated. Other updates, such as scaling the Deployment, do not trigger a rollout.

public Pod createPod() {
      int hashCode = this.getSpec().getPodTemplateSpec().hashCode();
      Pod pod = new PodBuilder()
              .withNewMetadata()
              .withLabels(this.getSpec().getLabelSelector().getMatchLabels())
              .addToLabels("pod-template-hash", String.valueOf(hashCode > 0 ? hashCode : -hashCode))
              .withName(this.getMetadata().getName()
                      .concat("-")
                      .concat(UUID.randomUUID().toString()))
              .withNamespace(this.getMetadata().getNamespace())
              .withOwnerReferences(
                      new OwnerReferenceBuilder()
                              .withApiVersion(this.getApiVersion())
                              .withController(Boolean.TRUE)
                              .withBlockOwnerDeletion(Boolean.TRUE)
                              .withKind(this.getKind())
                              .withName(this.getMetadata().getName())
                              .withUid(this.getMetadata().getUid())
                              .build()
              )
              .withUid(UUID.randomUUID().toString())
              .endMetadata()
              .withSpec(this.getSpec().getPodTemplateSpec().getSpec())
              .build();
      return pod;
  }複製代碼

更新MyDeployment的實現邏輯

更新時主要考慮了兩種狀況：縮擴容和更新鏡像。

經過判斷目前Pod數量和MyDeployment中spec的replicas中是否相同來判斷是否須要縮擴容。

經過判斷是否存在Pod與MyDeployment的container列表是否相同來判斷是否須要更新鏡像。

若是僅須要更新鏡像，則進行rolling-update；

若是僅須要縮擴容，則進行scale；

若是都須要，則先對剩餘Pod進行rolling-update，再對縮擴容的Pod進行縮擴容。

rolling-update是滾動升級，一種簡單的實現是先擴容一個Pod（新的鏡像），再縮容一個Pod（老的鏡像），如此反覆，直到所有都是新的鏡像爲止。

public void handle(MyDeployment myDeployment) {
          log.info("{} modified", myDeployment.getMetadata().getName());
          PodList pods = delegate.client().pods().inNamespace(myDeployment.getMetadata().getNamespace()).withLabelSelector(myDeployment.getSpec().getLabelSelector()).list();
          int podSize = pods.getItems().size();
          int replicas = myDeployment.getSpec().getReplicas();
          boolean needScale = podSize != replicas;
          boolean needUpdate = pods.getItems().stream().anyMatch(pod -> {
              return myDeployment.isPodTemplateChanged(pod);
          });
          log.info("needScale: {}", needScale);
          log.info("needUpdate: {}", needUpdate);
          // 僅更新podTemplate
          if (!needScale) {
              syncRollingUpdate(myDeployment, pods.getItems());
          } else if (!needUpdate) {
              // 僅擴縮容
              int diff = replicas - podSize;
              if (diff > 0) {
                  scaleUp(myDeployment, diff);
              } else {
                  // 把列表前面的縮容，後面的不動
                  scaleDown(pods.getItems().subList(0, -diff));
              }
          } else {
              // 同時scale&update
              // 對剩餘部分作rolling-update，而後對diff進行縮擴容
              syncRollingUpdate(myDeployment, pods.getItems().subList(0, Math.min(podSize, replicas)));
              int diff = replicas - podSize;
              if (diff > 0) {
                  scaleUp(myDeployment, diff);
              } else {
                  scaleDown(pods.getItems().subList(replicas, podSize));
              }
          }
      }複製代碼

值得注意的一點是，全部CRUD的APi均爲REST調用，是Kubernetes API Server將對象的指望狀態寫入到ETCD中，而後由kubelet監聽事件，去執行變動。

這一點在rolling-update過程當中要格外注意，」先擴容，後縮容「中後縮容的前提是擴容成功，即新的Pod建立成功，且狀態變爲Ready。因此僅僅調用一下create接口是不夠的，須要等待直至Ready；刪除同理。

private void syncRollingUpdate(MyDeployment myDeployment, List<Pod> pods) {
      pods.forEach(oldPod -> {
          Pod newPod = myDeployment.createPod();
          log.info("Thread {}: pod {} is creating", Thread.currentThread().getId(), newPod.getMetadata().getName());
          delegate.createPodAndWait(newPod, myDeployment);
          log.info("Thread {}: pod {} is deleting", Thread.currentThread().getId(), oldPod.getMetadata().getName());
          delegate.deletePodAndWait(oldPod);
      });
  }複製代碼

public void createPodAndWait(Pod pod, MyDeployment myDeployment) {
      client.pods().create(pod);
      CountDownLatch latch = new CountDownLatch(1);
      modifiedPodLatchMap.put(pod.getMetadata().getUid(), latch);
      try {
          latch.await(TIME_OUT, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
         log.error("{}", e);
      }
      log.info("createPodAndWait wait finished successfully!");
  }複製代碼

這裏是await阻塞等待，當前類也實現了PodModifiedWatcher接口，countDown是在Pod狀態變動時觸發。

當Pod被建立後，每每會觸發兩個事件，第一個是Added，狀態是NotReady，第二個是Modified，狀態是Ready。這裏咱們檢測到新增的Pod正常運行後，去喚醒執行rolling-update的線程，以實現createPodAndwait的效果。

@Override
  public void onPodModified(Pod pod) {
      if (UnifiedPodWatcher.isPodReady(pod)) {
          CountDownLatch latch = modifiedPodLatchMap.remove(pod.getMetadata().getUid());
          if(latch != null) {
              latch.countDown();
          }
      }
  }複製代碼

縮擴容的邏輯比較簡單，不須要作等待。

/** * @param myDeployment * @param count 擴容的pod數量 */
  private void scaleUp(MyDeployment myDeployment, int count) {
      for (int i = 0; i < count; i++) {
          Pod pod = myDeployment.createPod();
          log.info("scale up pod[{}]: {} , {}", i, pod.getMetadata().getName(), pod);
          delegate.client().pods().create(pod);
      }
  }
  ​
  /** * @param pods 待刪掉的pod列表 */
  private void scaleDown(List<Pod> pods) {
      for (int i = 0; i < pods.size(); i++) {
          Pod pod = pods.get(i);
          log.info("scale down pod[{}]: {} , {}", i, pod.getMetadata().getName(), pod);
          delegate.client().pods().delete(pod);
      }
  }複製代碼

刪除MyDeployment的實現邏輯

其實就是沒有邏輯=.=

Kubernetes的邏輯是刪掉了一個資源時，若是其餘資源的Metadata中的ownerReference中引用該資源，那麼這些資源會被級聯刪除，這個行爲是能夠配置的，而且默認爲true。

因此咱們不須要在此處作任何事情。