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

// NodePoolStatus defines the observed state of NodePool
type NodePoolStatus struct {
	// status=200 说明正常，其他情况为异常情况
	Status int `json:"status"`
	// 节点的数量
	NodeCount int `json:"nodeCount"`
	// 允许被调度的容量
	Allocatable corev1.ResourceList `json:"allocatable,omitempty" protobuf:"bytes,2,rep,name=allocatable,casttype=ResourceList,castkey=ResourceName"`
}
//+kubebuilder:object:root=true
//+kubebuilder:resource:scope=Cluster
//+kubebuilder:subresource:status
// NodePool is the Schema for the nodepools API
type NodePool struct {

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
31
32
33
34

func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	// ......
	if len(nodes.Items) > 0 {
		r.Log.Info("find nodes, will merge data", "nodes", len(nodes.Items))
+		pool.Status.Allocatable = corev1.ResourceList{}
+		pool.Status.NodeCount = len(nodes.Items)
		for _, n := range nodes.Items {
			n := n
			// 更新节点的标签和污点信息
			err := r.Update(ctx, pool.Spec.ApplyNode(n))
			if err != nil {
				return ctrl.Result{}, err
			}
+			for name, quantity := range n.Status.Allocatable {
+				q, ok := pool.Status.Allocatable[name]
+				if ok {
+					q.Add(quantity)
+					pool.Status.Allocatable[name] = q
+					continue
+				}
+				pool.Status.Allocatable[name] = quantity
+			}
		}
	}
  // ......
  +	pool.Status.Status = 200
+	err = r.Status().Update(ctx, pool)
	return ctrl.Result{}, err
}

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

apiVersion: nodes.lailin.xyz/v1
kind: NodePool
metadata:
  name: worker
spec:
  taints:
    - key: node-pool.lailin.xyz
      value: worker
      effect: NoSchedule
  labels:
    "node-pool.lailin.xyz/worker": "10"
  handler: runc

1
2
3
4
5

▶ kubectl get no 
NAME                 STATUS   ROLES                  AGE   VERSION
kind-control-plane   Ready    control-plane,master   29m   v1.20.2
kind-worker          Ready    worker                 28m   v1.20.2
kind-worker2         Ready    worker                 28m   v1.20.2

1
2
3
4
5
6
7
8
9
10

status:
  allocatable:
    cpu: "8"
    ephemeral-storage: 184026512Ki
    hugepages-1Gi: "0"
    hugepages-2Mi: "0"
    memory: 6129040Ki
    pods: "220"
  nodeCount: 2
  status: 200

1
2
3
4
5

+//+kubebuilder:printcolumn:JSONPath=".status.status",name=Status,type=integer
+//+kubebuilder:printcolumn:JSONPath=".status.nodeCount",name=NodeCount,type=integer
//+kubebuilder:object:root=true
//+kubebuilder:resource:scope=Cluster
//+kubebuilder:subresource:status

1
2
3

▶ kubectl get NodePool 
NAME     STATUS   NODECOUNT
worker   200      2

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

// NodePoolReconciler reconciles a NodePool object
type NodePoolReconciler struct {
	client.Client
	Log      logr.Logger
	Scheme   *runtime.Scheme
+	Recorder record.EventRecorder
}
func (r *NodePoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	+	// 添加测试事件
+	r.Recorder.Event(pool, corev1.EventTypeNormal, "test", "test")
	pool.Status.Status = 200
	err = r.Status().Update(ctx, pool)
	return ctrl.Result{}, err
}

1
2
3
4
5
6
7
8
9

if err = (&controllers.NodePoolReconciler{
		Client:   mgr.GetClient(),
		Log:      ctrl.Log.WithName("controllers").WithName("NodePool"),
		Scheme:   mgr.GetScheme(),
+		Recorder: mgr.GetEventRecorderFor("NodePool"),
	}).SetupWithManager(mgr); err != nil {
		setupLog.Error(err, "unable to create controller", "controller", "NodePool")
		os.Exit(1)
	}

1
2
3
4

Events:
  Type    Reason  Age   From      Message
  ----    ------  ----  ----      -------
  Normal  test    4s    NodePool  test

1
2
3
4
5
6

// SetupWithManager sets up the controller with the Manager.
func (r *NodePoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&nodesv1.NodePool{}).
		Complete(r)
}

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
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

// SetupWithManager sets up the controller with the Manager.
func (r *NodePoolReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&nodesv1.NodePool{}).
		Watches(&source.Kind{Type: &corev1.Node{}}, handler.Funcs{UpdateFunc: r.nodeUpdateHandler}).
		Complete(r)
}
func (r *NodePoolReconciler) nodeUpdateHandler(e event.UpdateEvent, q workqueue.RateLimitingInterface) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	oldPool, err := r.getNodePoolByLabels(ctx, e.ObjectOld.GetLabels())
	if err != nil {
		r.Log.Error(err, "get node pool err")
	}
	if oldPool != nil {
		q.Add(reconcile.Request{
			NamespacedName: types.NamespacedName{Name: oldPool.Name},
		})
	}
	newPool, err := r.getNodePoolByLabels(ctx, e.ObjectNew.GetLabels())
	if err != nil {
		r.Log.Error(err, "get node pool err")
	}
	if newPool != nil {
		q.Add(reconcile.Request{
			NamespacedName: types.NamespacedName{Name: newPool.Name},
		})
	}
}
func (r *NodePoolReconciler) getNodePoolByLabels(ctx context.Context, labels map[string]string) (*nodesv1.NodePool, error) {
	pool := &nodesv1.NodePool{}
	for k := range labels {
		ss := strings.Split(k, "node-role.kubernetes.io/")
		if len(ss) != 2 {
			continue
		}
		err := r.Client.Get(ctx, types.NamespacedName{Name: ss[1]}, pool)
		if err == nil {
			return pool, nil
		}
		if client.IgnoreNotFound(err) != nil {
			return nil, err
		}
	}
	return nil, nil
}