简介

通过将 etcd 集群定义为一个 K8S CRD，etcd-operator 负责 etcd 集群的创建与运维。

它主要包含三个controller:

1. cluster-operator: 负责自动化创建，销毁，升级，自动扩缩容，故障迁移etcd集群

2. backup-operator: 负责对etcd的数据进行定时备份，备份后端支持远程存储，如aliyun oss存储

3. restore-operator: 负责通过备份数据恢复etcd集群

operator 流程分析

1. 创建 EtcdCluster CRD；

2. 创建 EtcdCluster 的 Informer 来处理 EtcdCluster 增删改事件；

3. 当用户提交 EtcdCluster 创建请求；创建 Etcd 集群初始节点；根据期望的 Etcd 集群 size 创建并加入成员节点；

4. 当用户提交 EtcdCluster 更新（ 镜像版本 / 集群 pod size 等更新）请求：调整集群到期望状态；

5. 当用户提交 EtcdCluster 删除请求：无需操作，由垃圾回收自动删除节点；

6. Etcd 集群数据的备份及恢复分别由 EtcdBackup 和 EtcdRestore 的 Operator 实现；

源码结构

├── cmd # 程序入口
│   ├── backup-operator # 备份集群用的 Operator
│   ├── operator # 集群 Operator
│   └── restore-operator # 恢复集群用的 Operator
├── example # 一些示例文件
│   ├── deployment.yaml # 部署 Operator
│   ├── etcd-backup-operator # 备份用 Operator 相关
│   ├── etcd-restore-operator # 恢复用 Operator 相关
│   ├── example-etcd-cluster-nodeport-service.json
│   ├── example-etcd-cluster.yaml # 部署 Etcd 集群
│   ├── rbac # 用于创建 RBAC 规则
│   └── tls # 部署 TLS 连接版的 Etcd 集群
├── hack # 提供一些开发相关的脚本
├── pkg # 主要源码
│   ├── apis # EtcdCluster API 组定义
│   ├── backup # 操作备份恢复源相关的实现
│   ├── chaos # 集群容灾测试相关
│   ├── cluster # 集群控制实现
│   │   ├── cluster.go # Etcd 集群的实际维护
│   │   ├── metrics.go # 监控相关
│   │   ├── reconcile.go # 节点的创建或删除
│   │   └── upgrade.go # 节点的升级
│   ├── controller # EtcdCluster Controller 实现
│   │   ├── backup-operator # 备份集群用的 Operator 实现
│   │   ├── restore-operator # 恢复集群用的 Operator 实现
│   │   ├── controller.go # 事件处理 Handler
│   │   ├── informer.go # 创建 Informer 监听 EtcdCluster 增删改事件
│   │   ├── metrics.go # Prometheus 数据统计
│   ├── generated # K8S 工具生成的代码
│   └── util # 一些工具，操作 Pod 对象，Etcd API 调用等

Controller 初始化

创建 EtcdCluster Controller：

func run(ctx context.Context) {  // 用于测试 Etcd 集群容灾状况，仅用于测试环境  startChaos(context.Background(), cfg.KubeCli, cfg.Namespace, chaosLevel)  // 创建 Controller 并开始控制循环  c := controller.New(cfg)  err := c.Start()  logrus.Fatalf("controller Start() failed: %v", err)}

创建 EtcdCluster CRD：

func (c *Controller) Start() error {  ...  // 等待 EtcdCluster CRD 创建完成  for {    err := c.initResource()    if err == nil {      break    }    ...    time.Sleep(initRetryWaitTime)  }  ...  c.run()  ...}

创建 Informer 处理 EctdCluster 事件：

func (c *Controller) run() {  ...  // EctdCluster 对象的增删改都会调用 handleClusterEvent 来处理  _, informer := cache.NewIndexerInformer(source, &api.EtcdCluster{}, 0, cache.ResourceEventHandlerFuncs{    AddFunc:    c.onAddEtcdClus,    UpdateFunc: c.onUpdateEtcdClus,    DeleteFunc: c.onDeleteEtcdClus,  }, cache.Indexers{})  ...  // TODO：以后可以使用 Queue 来避免阻塞  informer.Run(ctx.Done())}

处理事件类型循环:

func (c *Controller) handleClusterEvent(event *Event) (bool, error) {  ...  // 集群失效后，从维护的集群集合中删除  if clus.Status.IsFailed() {    ...    if event.Type == kwatch.Deleted {      delete(c.clusters, getNamespacedName(clus))      return false, nil    }    ...  }  ...  switch event.Type {  case kwatch.Added:    ...    // 创建 Etcd 集群    nc := cluster.New(c.makeClusterConfig(), clus)    ...    c.clusters[getNamespacedName(clus)] = nc    ...  case kwatch.Modified:    ...    // 更新 Etcd 集群    c.clusters[getNamespacedName(clus)].Update(clus)    ...  case kwatch.Deleted:    ...    // 删除 Etcd 集群    c.clusters[getNamespacedName(clus)].Delete()    delete(c.clusters, getNamespacedName(clus))    ...  }  return false, nil}

Controller 控制循环

1. 收到 EtcdCluster 创建事件，则创建一个新的集群

func New(config Config, cl *api.EtcdCluster) *Cluster {  c := &Cluster{    ...  }  go func() {    // 初始化集群    if err := c.setup(); err != nil {      ...    }    // 开始集群节点控制循环    c.run()  }()  return c}

1. 为集群创建初始节点

func (c *Cluster) startSeedMember() error {  m := &etcdutil.Member{    Name:         k8sutil.UniqueMemberName(c.cluster.Name),    Namespace:    c.cluster.Namespace,    SecurePeer:   c.isSecurePeer(),    SecureClient: c.isSecureClient(),  }  ms := etcdutil.NewMemberSet(m)  if err := c.createPod(ms, m, "new"); err != nil {  ...  _, err := c.eventsCli.Create(k8sutil.NewMemberAddEvent(m.Name, c.cluster))  ...}

1. 为集群创建两个 Headless Service

• CreateClientService：用于 Etcd 客户端的访问；

• CreatePeerService：用于节点间访问；

func (c *Cluster) setupServices() error {  err := k8sutil.CreateClientService(c.config.KubeCli, c.cluster.Name, c.cluster.Namespace, c.cluster.AsOwner())  ...  return k8sutil.CreatePeerService(c.config.KubeCli, c.cluster.Name, c.cluster.Namespace, c.cluster.AsOwner())}

1. 在控制循环中创建或删除成员节点以达到集群期望状态

每隔 reconcileInterval  秒调整一次集群状态 reconcile()：

• 将集群节点数调整到期望的 size；reconcileMembers()

• 删除所有不属于集群中的成员节点；

• 创建集群中缺失的成员节点；

• 如果集群未达到法定节点数，则退出并报错；

• 将集群节点调整到期望的 Etcd 版本；upgradeOneMember()

• 修改 Pod 对象的 Image 后请求 patch；

Etcd 节点的启动方式

• 在启动节点 Pod 时，其中的 init container 会等到 Pod DNS 解析可用后才会启动 Etcd 容器；

• 从 MemberSet 中找到其他成员的 DNS 名称并配置 --initial-cluster 参数后启动 Etcd 容器；

• Etcd Operator 部署 Etcd 集群，采用的是静态集群（Static）的方式。

可以看到，在 etcd 集群启动参数（比如：initial-cluster）里，Etcd Operator 只会使用 Pod 的 DNS 记录，而不是它的 IP 地址。

这当然是因为，在 Operator 生成上述启动命令的时候，Etcd 的 Pod 还没有被创建出来，它的 IP 地址自然也无从谈起。

每个 Cluster 对象，都会事先创建一个与该 EtcdCluster 同名的 Headless Service。这样，Etcd Operator 在接下来的所有创建 Pod 的步骤里，就都可以使用 Pod 的 DNS 记录来代替它的 IP 地址了。

func newEtcdPod(m *etcdutil.Member, initialCluster []string, clusterName, state, token string, cs api.ClusterSpec) *v1.Pod {  ...  pod := &v1.Pod{    ...    Spec: v1.PodSpec{      InitContainers: []v1.Container{{        ...        Command: []string{"/bin/sh", "-c", fmt.Sprintf(`          TIMEOUT_READY=%d          while ( ! nslookup %s )          do            # If TIMEOUT_READY is 0 we should never time out and exit            TIMEOUT_READY=$(( TIMEOUT_READY-1 ))                        if [ $TIMEOUT_READY -eq 0 ];                then                    echo "Timed out waiting for DNS entry"                    exit 1                fi            sleep 1          done`, DNSTimeout, m.Addr())},      }},      Containers:    []v1.Container{container},      ...    },  }  ...}

ACK 集群部署实战

选用 ACK 版本：v1.20.11-aliyun.1

安装etcd operator

CoreDNS 无法解析 etcd pod 的 IP，部署在 init container 等待 Pod DNS  解析 hang  住了。

附录：使用 Kubebuilder demo

kubebuilder 是生成 k8s operator 的脚手架工具，本 demo 介绍使用 kubebuilder 快速搭建一个 operator 项目。

1. 初始化项目：

kubebuilder init --domain my.domain --repo my.domain/guestbook

kubebuilder create api --group webapp --version v1 --kind Guestbook

➜  guestbook tree
.
├── Dockerfile
├── Makefile
├── PROJECT
├── README.md
├── api
│   └── v1
│       ├── groupversion_info.go
│       ├── guestbook_types.go
│       └── zz_generated.deepcopy.go
├── bin
│   └── controller-gen
├── config
│   ├── crd
│   │   ├── kustomization.yaml
│   │   ├── kustomizeconfig.yaml
│   │   └── patches
│   │       ├── cainjection_in_guestbooks.yaml
│   │       └── webhook_in_guestbooks.yaml
│   ├── default
│   │   ├── kustomization.yaml
│   │   ├── manager_auth_proxy_patch.yaml
│   │   └── manager_config_patch.yaml
│   ├── manager
│   │   ├── controller_manager_config.yaml
│   │   ├── kustomization.yaml
│   │   └── manager.yaml
│   ├── prometheus
│   │   ├── kustomization.yaml
│   │   └── monitor.yaml
│   ├── rbac
│   │   ├── auth_proxy_client_clusterrole.yaml
│   │   ├── auth_proxy_role.yaml
│   │   ├── auth_proxy_role_binding.yaml
│   │   ├── auth_proxy_service.yaml
│   │   ├── guestbook_editor_role.yaml
│   │   ├── guestbook_viewer_role.yaml
│   │   ├── kustomization.yaml
│   │   ├── leader_election_role.yaml
│   │   ├── leader_election_role_binding.yaml
│   │   ├── role_binding.yaml
│   │   └── service_account.yaml
│   └── samples
│       └── webapp_v1_guestbook.yaml
├── controllers
│   ├── guestbook_controller.go
│   └── suite_test.go
├── go.mod
├── go.sum
├── hack
│   └── boilerplate.go.txt
└── main.go

13 directories, 38 files

1. 部署 CRD

➜  guestbook make install

➜  guestbook kubectl get crd |grep my.domain
guestbooks.webapp.my.domain                                      2022-05-04T08:58:36Z

可以看到 CRD 已经创建成功，但是 CR  实例还没创建。

1. 创建 CR 实例：

➜  guestbook kubectl apply -f config/samples/
guestbook.webapp.my.domain/guestbook-sample created

4.本地运行 operator ：

➜  guestbook make run...go fmt ./...go vet ./...go run ./main.goI0504 16:59:38.598884   16635 request.go:665] Waited for 1.020466713s due to client-side throttling, not priority and fairness, request: GET:https://59.110.25.213:6443/apis/metrics.alibabacloud.com/v1alpha1?timeout=32s1.6516547788561852e+09  INFO  controller-runtime.metrics  Metrics server is starting to listen  {"addr": ":8090"}1.6516547788572779e+09  INFO  setup  starting manager1.651654778857989e+09  INFO  Starting server  {"path": "/metrics", "kind": "metrics", "addr": "[::]:8090"}1.6516547788579981e+09  INFO  Starting server  {"kind": "health probe", "addr": "[::]:8089"}1.651654778858467e+09  INFO  controller.guestbook  Starting EventSource  {"reconciler group": "webapp.my.domain", "reconciler kind": "Guestbook", "source": "kind source: *v1.Guestbook"}1.651654778858505e+09  INFO  controller.guestbook  Starting Controller  {"reconciler group": "webapp.my.domain", "reconciler kind": "Guestbook"}1.6516547799614458e+09  INFO  controller.guestbook  Starting workers  {"reconciler group": "webapp.my.domain", "reconciler kind": "Guestbook", "worker count": 1}

后续可以继续将 operator 部署到 k8s 上