1
2
3
4
5
6
7
8
9

用户提交 YAML (期望状态)
        ↓
    API Server (持久化到 Etcd)
        ↓
    控制器 Watch 到变更
        ↓
    执行 Reconcile：当前状态 ≠ 期望状态 → 调整
        ↓
    更新 Status 字段报告实际状态

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

Kubernetes API 树形结构：

/api                          ← 核心 API（Pod、Node、Service 等，Group="")
└── /v1
    ├── pods
    ├── nodes
    └── services

/apis                         ← 非核心 API
├── apps/v1
│   └── deployments
├── batch/v1
│   ├── jobs
│   └── cronjobs
├── networking.k8s.io/v1
│   └── networkpolicies
└── <自定义 Group>/v1
    └── <自定义 Resource>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

POST /apis/batch/v2alpha1/namespaces/default/cronjobs
  ↓
[1] 过滤 Filter：认证、授权、超时、审计
  ↓
[2] MUX & Routes：URL 匹配 → 找到 CronJob 的 Handler
  ↓
[3] 类型定义匹配：按 Group → Version → Resource 定位 CronJob 类型
  ↓
[4] Convert：将用户提交的 YAML 转换为 Super Version（该类型所有版本的字段全集）
  ↓
[5] Admission：准入控制（Admission Controller、Initializer）
  ↓
[6] Validation：字段校验
  ↓
[7] 写入 Etcd

1
2
3
4
5

main()                     ← 初始化 client、InformerFactory、启动控制器
  ↓
Controller 定义            ← Watch 回调注册、工作队列
  ↓
业务逻辑 (Reconcile)       ← 从队列取 key → 调谐逻辑 → 更新 Status

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func main() {
    // 1. 创建 Kubernetes client 和自定义资源的 client
    cfg, _ := clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
    kubeClient, _ := kubernetes.NewForConfig(cfg)
    networkClient, _ := clientset.NewForConfig(cfg)

    // 2. 创建 Informer 工厂
    informerFactory := informers.NewSharedInformerFactory(networkClient, time.Second*30)

    // 3. 创建控制器
    controller := NewController(kubeClient, networkClient,
        informerFactory.Samplecrd().V1().Networks())

    // 4. 启动 Informer 和控制器
    go informerFactory.Start(stopCh)
    controller.Run(2, stopCh)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

API Server
    │
    │  List & Watch
    ▼
Reflector ──→ Delta FIFO ──→ Indexer（本地缓存）
                │
                │ 事件分发
                ▼
           Informer
                │
                │ 回调
                ▼
         Controller (工作队列 → Reconcile)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

func (c *Controller) Run(workers int, stopCh <-chan struct{}) error {
    // 等待本地缓存同步完成
    if !cache.WaitForCacheSync(stopCh, c.informerSynced) {
        return fmt.Errorf("cache sync failed")
    }

    // 启动 workers
    for i := 0; i < workers; i++ {
        go wait.Until(c.runWorker, time.Second, stopCh)
    }
    <-stopCh
    return nil
}

func (c *Controller) runWorker() {
    for c.processNextWorkItem() {
    }
}

func (c *Controller) processNextWorkItem() bool {
    key, quit := c.workqueue.Get()
    defer c.workqueue.Done(key)
    err := c.syncHandler(key.(string))
    if err != nil {
        c.workqueue.AddRateLimited(key)  // 失败重试
    } else {
        c.workqueue.Forget(key)
    }
    return true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 仅作用于 mynamespace 下的 Pod
kind: Role
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  namespace: mynamespace
  name: pod-reader
rules:
  - apiGroups: [""]
    resources: ["pods"]
    verbs: ["get", "watch", "list"]

1
2
3
4
5
6
7
8
9

# 作用于全集群的 Node 对象
kind: ClusterRole
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: node-reader
rules:
  - apiGroups: [""]
    resources: ["nodes"]
    verbs: ["get", "watch", "list"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: read-pods
  namespace: mynamespace
subjects:
  - kind: ServiceAccount
    name: my-controller
    namespace: default
roleRef:
  kind: Role
  name: pod-reader
  apiGroup: rbac.authorization.k8s.io

1
2
3
4
5
6
7

CRD (CustomResourceDefinition)     ← 定义"什么是一个 EtcdCluster"
    +
Custom Controller (Informer 模式)  ← 持续调谐，驱动实际状态到期望状态
    +
领域知识 (编码在 Reconcile 中)     ← 知道如何创建/扩缩容/备份/恢复 Etcd
    =
Operator

1
2
3
4
5

# 1. 创建 RBAC 规则
kubectl create -f example/rbac/create_role.sh

# 2. 部署 Operator
kubectl create -f example/deployment.yaml

1
2
3

$ kubectl get crd
NAME                                    CREATED AT
etcdclusters.etcd.database.coreos.com   2023-09-18T11:42:55Z

1
2
3
4

Group:    etcd.database.coreos.com
Kind:     EtcdCluster
Version:  v1beta2
Scope:    Namespaced

1
2
3
4
5
6
7

apiVersion: etcd.database.coreos.com/v1beta2
kind: EtcdCluster
metadata:
  name: my-etcd
spec:
  size: 3
  version: "3.4.7"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

用户创建 EtcdCluster CR (期望: size=3, version=3.4.7)
        ↓
API Server 写入 Etcd
        ↓
Etcd Operator (Informer Watch 到新 CR)
        ↓
Reconcile:
  1. 查当前状态: 0 个 Etcd Pod 在运行
  2. 调谐:
     ├── 创建 Service (Headless)
     ├── 创建 Seed Member (第一个 Pod)
     ├── 创建 Member 2
     └── 创建 Member 3
  3. 更新 CR Status: size=3, phase=Running
        ↓
运行中: 一个 Pod 挂了
        ↓
Operator 收到 Pod 事件
        ↓
Reconcile:
  1. 查当前状态: 2 个 Etcd Pod 在运行
  2. 调谐: 删除失败 Pod，创建新的
  3. 更新 Status