【转】如何写一个operator
cover
description: "Writing a Controller for Pod Labels"
date: 2021.09.25 10:26
categories:
- K8s
tags: [K8s, Go]
keywords: Operator, Controller, Reconcile
原文地址：https://www.jianshu.com/p/79476712575e
如何写一个operator
文章源地址请移步writing-a-controller-for-pod-labels
k8s中的operator是什么？
operator旨在简化基于k8s部署有状态服务（例如：ceph集群、skywalking集群）
可以利用Operator SDK 构建一个operator，
operator使扩展k8s及实现自定义调度变得更加简单。
尽管Operator SDK 适合构建功能齐全的operator，
但也可以使用它来编写单个控制器。
这篇文章将指导您在Go中编写一个Kubernetes控制器，该控制器将向具有特定注释的pod添加pod-name标签
为什么我们需要一个控制器呢？
最近我们项目中有这么个需求：通过一个service将流量路由至同一ReplicaSet中指定pod内（service对应一个或多个pod）
而原生k8s并不能实现该功能，因为原生service只能通过label与Pod匹配，并且同一ReplicaSet内，Pod具有相同标签。
上述需求有两种解决方案:
- 创建
service时不指定标签选择器，而是利用Endpoints或EndpointSlices关联pod
此时我们需要写一个自定义控制器，用于插入指定pod的端点地址至Endpoints或EndpointSlices对象 - 为每个
Pod添加具有唯一value的标签，接下来我们就可以利用标签选择器进行service与Pod的关联。
由于k8s中的控制器实质是个控制循环程序，控制器可以对k8s的资源（Resource，比如namespace、service等）进行监听追踪。
此时如果我们创建一个控制器，仅监听Pod资源，针对指定Pod进行label处理，就可实现上述需求。
当然k8s原生资源StatefulSets也是可以实现这一功能的，但假设我们不想/不能使用StatefulSets类型去实现呢？
一般情况下，我们很少直接创建Pod类型，而是通过Deployment, ReplicaSet间接创建Pod。
我们可以指定标签添加到PodSpec中的每个Pod，但不能使用动态值，因此无法复制StatefulSet的pod-name标签。
我们尝试使用mutating admission webhook
实现。
当任何人创建Pod时，webhook会自动注入一个包含Pod名称的标签对Pod进行修改。
遗憾的是这种方式并不能实现我们的需求： 并不是所有的Pod在创建前都有名字。
举个例子：当ReplicaSet控制器创建一个Pod时，他向kube-apiserver发送一个请求，获取一个namePrefix而非name
kubeapi-server在将新的Pod持久化到etcd之前生成一个唯一的名称，
这个过程发生于在调用我们的许可webhook之后。所以在大多数情况下，我们无法知道一个带有mutating webhook的Pod的名字
一旦Pod持久化至K8s集群中时，它几乎不会发生变更，但我们仍然可以通过以下方式，添加label
kubectl label my-pod my-label-key=my-label-value
我们需要观察Kubernetes API中任何Pod的变化，并添加我们想要的标签。
我们将编写一个控制器来为我们做这件事，而不是手动做这件事
利用Operator SDK构建一个控制器
控制器是一个协调循环，它从Kubernetes API中读取期望的资源状态，并采取行动使集群的实际状态达到期望状态
安装配置
1.安装
Operator SDK
- 下载二进制
sudo curl -LO https://github.com/operator-framework/operator-sdk/releases/download/v1.12.0/operator-sdk_linux_amd64
sudo mv operator-sdk_linux_amd64 /usr/local/bin/operator-sdk
2.构建工程
mkdir label-operator && cd label-operator
3.初始化工程
export GOPROXY=https://goproxy.cn
operator-sdk init --domain=weiliang.io --repo=github.com/weiliang-ms/label-operator
4.创建控制器
接下来我们创建一个控制器，这个控制器将会处理Pod资源，而非自定义资源，所以不需要生成资源代码。
operator-sdk create api --group=core --version=v1 --kind=Pod --controller=true --resource=false
初始化编码
controllers/pod_controller.go解析
现在我们拥有了一个新文件: controllers/pod_controller.go。
该文件包含了PodReconciler类型，该类型包含两个方法：
-
Reconcile函数：
func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
_ = log.FromContext(ctx)
// your logic here
return ctrl.Result{}, nil
}
当创建、更新、或删除Pod时会调用Reconcile方法，Pod名称与命名空间作为函数入参，存于ctrl.Request对象之中
-
SetupWithManager函数：
func (r *PodReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&corev1.Pod{}).
Complete(r)
}
operator会在启动时执行SetupWithManager函数，SetupWithManager函数用于生命监听资源类型
因为我们只想要监听Pod资源变化，所以监听资源这部分代码不动
RBAC设置
接下来为我们的控制器配置RBAC权限，代码生成器生成的默认权限如下：
//+kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=core,resources=pods/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=core,resources=pods/finalizers,verbs=update
显然我们并不需要以上全部权限，我们控制器从不会CRUD Pod的status与finalizers字段。
控制器需要的仅仅是对Pod的读权限与更新权限，本着最小原则，我们调整权限如下
// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;list;watch;update;patch
此时我们已经编写好了控制器的基本调用逻辑。
实现Reconcile函数
我们希望Reconcile实现以下功能：
- 通过入参
ctrl.Request中的Pod名称与命名空间字段，请求k8s api获取Pod对象 - 如果
Pod拥有add-pod-name-label注解，给这个Pod添加一个pod-name标签 - 将上一步
Pod的变更回写k8s中
接下来我们为注解与标签定义一些常量
const (
addPodNameLabelAnnotation = "padok.fr/add-pod-name-label"
podNameLabel = "padok.fr/pod-name"
)
根据入参获取
Pod
首先我们根据入参信息，去k8s api获取Pod实例
func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
l := log.FromContext(ctx)
var pod corev1.Pod
if err := r.Get(ctx, req.NamespacedName, &pod); err != nil {
l.Error(err, "unable to fetch Pod")
return ctrl.Result{}, err
}
return ctrl.Result{}, nil
}
异常处理
当创建、更新或删除一个Pod时，会触发我们控制器的Reconcile方法
但当事件为'删除事件'时，r.Get()会返回一个指定错误对象，接下来我们通过引用下面的包来处理这个异常。
package controllers
import (
// other imports...
apierrors "k8s.io/apimachinery/pkg/api/errors"
// other imports...
)
// other functions...
func (r *PodReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
l := log.FromContext(ctx)
var pod corev1.Pod
if err := r.Get(ctx, req.NamespacedName, &pod); err != nil {
if apierrors.IsNotFound(err) {
// we'll ignore not-found errors, since we can get them on deleted requests.
return ctrl.Result{}, nil
}
l.Error(err, "unable to fetch Pod")
return ctrl.Result{}, err
}
return ctrl.Result{}, nil
}
// other functions...
编辑
Pod，判断注解、标签是否存在
此时我们已经获取到了这个Pod对象（创建、更新事件），接下来我们获取Pod的注解元数据，判断是否需要添加标签
...
/*
Step 1: 添加或移除标签.
*/
// 判断Pod是否存在注解 -> padok.fr/add-pod-name-label: true
labelShouldBePresent := pod.Annotations[addPodNameLabelAnnotation] == "true"
// 判断Pod是否存在标签 -> padok.fr/pod-name: Pod名称
labelIsPresent := pod.Labels[podNameLabel] == pod.Name
// 如果期望状态与实际状态一致（含有上述标签、注解），返回
if labelShouldBePresent == labelIsPresent {
log.Info("no update required")
return ctrl.Result{}, nil
}
// 存在注解 -> padok.fr/add-pod-name-label: true
if labelShouldBePresent {
// 判断标签map是否为空
if pod.Labels == nil {
// 为空创建
pod.Labels = make(map[string]string)
}
// 添加标签 -> padok.fr/pod-name: Pod名称
pod.Labels[podNameLabel] = pod.Name
log.Info("adding label")
} else {
// 不存在注解 -> padok.fr/add-pod-name-label: true
// 移除标签
delete(pod.Labels, podNameLabel)
log.Info("removing label")
}
...
回写
Pod至k8s
/*
Step 2: Update the Pod in the Kubernetes API.
*/
if err := r.Update(ctx, &pod); err != nil {
l.Error(err, "unable to update Pod")
return ctrl.Result{}, err
}
当我们回写Pod变更至k8s时存在以下风险：集群内的Pod与我们获取到的Pod已经不一致（可能通过其他渠道变更了该Pod）
在编写一个k8s控制器时，我们应该明白一个问题：我们编写的控制器并不是唯一能操作k8s资源对象的实例(其他控制器、kubectl等亦能操作k8s资源对象)
当发生这种情况时，最好的做法是通过重新排队事件，从头开始处理。
if err := r.Update(ctx, &pod); err != nil {
if apierrors.IsConflict(err) {
// The Pod has been updated since we read it.
// Requeue the Pod to try to reconciliate again.
return ctrl.Result{Requeue: true}, nil
}
if apierrors.IsNotFound(err) {
// The Pod has been deleted since we read it.
// Requeue the Pod to try to reconciliate again.
return ctrl.Result{Requeue: true}, nil
}
log.Error(err, "unable to update Pod")
return ctrl.Result{}, err
}
在k8s集群内运行该控制器
本人本地开发环境为windows10 + Ubuntu 20
本地
ubuntu安装Kubectl并配置kube-config
集群信息
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ kubectl get node
NAME STATUS ROLES AGE VERSION
node1 Ready master,worker 62d v1.18.6
node2 Ready master,worker 62d v1.18.6
node3 Ready master,worker 62d v1.18.6
node4 Ready worker 62d v1.18.6
label-operator下执行
shell目录
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ pwd
/mnt/d/github/label-operator
运行operator
export GOPROXY=https://goproxy.cn
make run
运行一个
nginx服务Pod
新建一个ubuntu shell窗口执行
kubectl run --image=nginx:1.20.0 my-nginx
查看Pod信息
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ kubectl get pod
NAME READY STATUS RESTARTS AGE
my-nginx 1/1 Running 0 78s
此时运行operator的窗口会输出如下信息，说明监听成功
2021-09-24T11:52:10.588+0800 INFO controller-runtime.manager.controller.pod no update required {"reconciler group": "", "reconciler kind": "Pod", "name": "m
y-nginx", "namespace": "default"}
2021-09-24T11:52:10.597+0800 INFO controller-runtime.manager.controller.pod no update required {"reconciler group": "", "reconciler kind": "Pod", "name": "m
y-nginx", "namespace": "default"}
2021-09-24T11:52:10.630+0800 INFO controller-runtime.manager.controller.pod no update required {"reconciler group": "", "reconciler kind": "Pod", "name": "m
y-nginx", "namespace": "default"}
查看
Pod标签
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ kubectl get pod my-nginx --show-labels
NAME READY STATUS RESTARTS AGE LABELS
my-nginx 1/1 Running 0 4m38s run=my-nginx
此时我们给该
Pod打上以下注解，并查看是否已自动添加新的标签
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ kubectl annotate pod my-nginx padok.fr/add-pod-name-label=true
pod/my-nginx annotated
查看标签
weiliang@DESKTOP-O8QG6I5:/mnt/d/github/label-operator$ kubectl get pod my-nginx --show-labels
NAME READY STATUS RESTARTS AGE LABELS
my-nginx 1/1 Running 0 6m39s padok.fr/pod-name=my-nginx,run=my-nginx
成功了！ 我们成功的实现上面的需求
