实现一个简单的Kubernetes controller

controller是什么

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

实现controller

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

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

apiVersion: v1kind: Servicemetadata:  name: my-servicespec:  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: v1kind: Servicemetadata:  name: my-servicespec:  selector:    app.kubernetes.io/name: MyApp  ports:    - protocol: TCP      port: 80      targetPort: 80  clusterIP: 10.0.171.239  type: LoadBalancerstatus:  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}