实现一个简单的Kubernetes controller

controller是什么

简单来说，controller就是监听kubernetes的配置/状态变化，自动对资源进行调节。比如kubernetes内置的controller有Deployment、StatefulSet、DaemonSet。

实现controller

这里我们实现一个简单的loadbalancer controller，使用kubernetes的client-go。

先创建一个新的Serivce,并且指定Type为LoadBalancer。

apiVersion: v1
kind: Service
metadata:
  name: my-service
spec:
  selector:
    app.kubernetes.io/name: MyApp
  ports:
    - protocol: TCP
      port: 80
      targetPort: 80
  clusterIP: 10.0.171.239
  type: LoadBalancer

这时Service的状态还是pending,因为还没有controller为其分配IP地址。

创建新的client

config, err := rest.InClusterConfig()
if err != nil {
    panic(err)
}

cli, err := kubernetes.NewForConfig(config)
if err != nil {
    panic(err)
}

创建一个新的对Service状态变化的监听

wch, err := cli.CoreV1().Services(corev1.NamespaceAll).Watch(context.TODO(), metav1.ListOptions{Watch: true})
if err != nil {
    panic(err)
}
defer wch.Stop()

如果是新建的Serivce,并且Type为LoadBalancer，则自动为其分配一个IP地址。

// 端口分配范围
var portRange = atomic.Uint32{}
func init() {
    portRange.Store(30000)
}


for {
        select {
        case <-signChannel:
            return
        case obj, ok := <-wch.ResultChan():
            if !ok {
                return
            }

            svc, ok := obj.Object.(*corev1.Service)
            if !ok {
                continue
            }

            switch obj.Type {
            case watch.Added:
                // 判断是否是LoadBalancer
                // 生成环境中还应结合Annotations进行判断，防止误更改错误的Service.
                if svc.Spec.Type != corev1.ServiceTypeLoadBalancer {
                    continue
                }

                // 生成端口，并创建转发（只会打印，不会真正创建规则）
                sport := portRange.Add(1)
                forward(svc, sport)

                // 为service分配IP地址
                svc.Status.LoadBalancer.Ingress = []corev1.LoadBalancerIngress{
                    {
                        IP: svc.Spec.ClusterIP,
                        Ports: []corev1.PortStatus{
                            {
                                Port: int32(sport),
                            },
                        },
                    },
                }

                // 更新service
                _, err := cli.
                    CoreV1().
                    Services(svc.Namespace).
                    UpdateStatus(context.TODO(), svc, metav1.UpdateOptions{})
                if err != nil {
                    slog.Error(err.Error())
                } else {
                    slog.Info("service added", svc.Name, svc.Namespace, svc.Spec.ClusterIP, svc.Spec.ClusterIPs)
                }

            case watch.Modified:
                slog.Info("service modified", svc.Name, svc.Namespace, svc.Spec.ClusterIP, svc.Spec.ClusterIPs)
            case watch.Deleted:
                slog.Info("service deleted", svc.Name, svc.Namespace, svc.Spec.ClusterIP, svc.Spec.ClusterIPs)
            }
        }
}

// ...
// ...

// 模拟实现iptables自动创建转发
func forward(svc *corev1.Service, sport uint32) {
    ip, err := netip.ParseAddr(svc.Spec.ClusterIP)
    if err != nil {
        panic(err)
    }

    for _, v := range svc.Spec.Ports {
        var ipt string
        var mask string

        if ip.Is4() {
            ipt = "iptables"
            mask = "32"
        } else {
            ipt = "ip6tables"
            mask = "128"
        }

        fmt.Println(
            ipt, "-t", "filter", "-A", "FORWARD",
            "-d", svc.Spec.ClusterIP+"/"+mask,
            "-p", string(v.Protocol),
            "--dport", strconv.Itoa(int(v.Port)),
            "-j", "DROP",
        )

        fmt.Println(
            ipt, "-t", "nat", "-I", "PRETROUTING", 0,
            "-p", string(v.Protocol),
            "--dport", strconv.Itoa(int(sport)),
            "-j", "DNAT",
            "--to", net.JoinHostPort(svc.Spec.ClusterIP, strconv.Itoa(int(v.Port))),
        )

        fmt.Println(
            ipt, "-t", "nat", "-I", "POSTROUTING",
            "-d", svc.Spec.ClusterIP+"/"+mask,
            "-p", v.Protocol,
            "-j", "MASQUERADE",
        )
    }
}

运行这个简单程序，再观察Serivce, 就会发现Serivice已经不再处于pending状态，并也为其分配了一个IP地址，虽然不能真正使用此地址进行访问，因为我们并没有创建真正的转发规则。

apiVersion: v1
kind: Service
metadata:
  name: my-service
spec:
  selector:
    app.kubernetes.io/name: MyApp
  ports:
    - protocol: TCP
      port: 80
      targetPort: 80
  clusterIP: 10.0.171.239
  type: LoadBalancer
status:
  loadBalancer:
    ingress:
    - ip: 10.0.171.239

k3s的klipper-lb是一个很简单loadbalancer实现，代码只是几行脚本。

start_proxy() {
    for src_range in ${SRC_RANGES//,/ }; do
        if echo ${src_range} | grep -Eq ":"; then
            ip6tables -t filter -I FORWARD -s ${src_range} -p ${DEST_PROTO} --dport ${DEST_PORT} -j ACCEPT
        else
            iptables -t filter -I FORWARD -s ${src_range} -p ${DEST_PROTO} --dport ${DEST_PORT} -j ACCEPT
        fi
    done

    for dest_ip in ${DEST_IPS//,/ }; do
        if echo ${dest_ip} | grep -Eq ":"; then
            [ $(cat /proc/sys/net/ipv6/conf/all/forwarding) == 1 ] || exit 1
            ip6tables -t filter -A FORWARD -d ${dest_ip}/128 -p ${DEST_PROTO} --dport ${DEST_PORT} -j DROP
            ip6tables -t nat -I PREROUTING -p ${DEST_PROTO} --dport ${SRC_PORT} -j DNAT --to [${dest_ip}]:${DEST_PORT}
            ip6tables -t nat -I POSTROUTING -d ${dest_ip}/128 -p ${DEST_PROTO} -j MASQUERADE
        else
            [ $(cat /proc/sys/net/ipv4/ip_forward) == 1 ] || exit 1
            iptables -t filter -A FORWARD -d ${dest_ip}/32 -p ${DEST_PROTO} --dport ${DEST_PORT} -j DROP
            iptables -t nat -I PREROUTING -p ${DEST_PROTO} --dport ${SRC_PORT} -j DNAT --to ${dest_ip}:${DEST_PORT}
            iptables -t nat -I POSTROUTING -d ${dest_ip}/32 -p ${DEST_PROTO} -j MASQUERADE
        fi
    done
}