如何从零开始编写一个 Kubernetes CRD

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

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

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

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

minikube addons enable dashboard

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

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

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

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

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();

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);
}

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);
        }
    });
}

@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;
        }
    }
}

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);
        }
    });
}

@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);
       }
    }
}

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;
}

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));
            }
        }
    }

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!");
}

@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);
    }
}