本文主要是介绍open yurt之yurt-tunel-server iptables规则源码分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
简介
yurt-tunel-server是open yurt开源的用于转发来自K8s API server的包含logs、exec和metrics等运维指令请求到边缘节点yurt-tunel-agent的反向代理。
源码分析
版本v0.4.1，入口函数中：
if cfg.EnableIptables {
//新建IptablesManager对象iptablesMgr := iptables.NewIptablesManager(cfg.Client, //KAS clientcfg.SharedInformerFactory.Core().V1().Nodes(), //node informercfg.ListenAddrForMaster, //https监听地址端口，默认端口是10250cfg.ListenInsecureAddrForMaster, //http监听地址端口，默认“127.0.0.1:10255”cfg.IptablesSyncPeriod, //同步更新iptalbes规则的时间间隔，默认60秒，最小15秒。)if iptablesMgr == nil {return fmt.Errorf("fail to create a new IptableManager")}wg.Add(1)go iptablesMgr.Run(stopCh, &wg) //启动规则同步任务}
// 定时更新规则
func (im *iptablesManager) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {defer wg.Done()// 等待节点缓存完成同步，也就是从k8s里获取到所有节点信息并缓存起来if !cache.WaitForCacheSync(stopCh,im.nodeInformer.Informer().HasSynced) {klog.Error("sync node cache timeout")return}// 服务启动后首次更新iptables规则im.syncIptableSetting()ticker := time.NewTicker(time.Duration(im.syncPeriod) * time.Second)defer ticker.Stop()for {select {case <-stopCh:klog.Info("stop the iptablesManager")im.cleanupIptableSetting()returncase <-ticker.C:im.syncIptableSetting() //定时更新iptables规则}}
}
//更新iptables规则
func (im *iptablesManager) syncIptableSetting() {//从k8s里指定configMap对象存里提取出除10250和10255之外还需要进行DNAT到insecurePort的边缘侧端口数组dnatPorts, err := util.GetConfiguredDnatPorts(im.kubeClient, im.insecurePort)if err != nil {klog.Errorf("failed to sync iptables rules, %v", err)return}//和上一次缓存的端口数组进行比对，分出哪些是被删除了的，内部源码略去portsChanged, deletedDnatPorts := im.getDeletedPorts(dnatPorts)//将端口20150和10255加入到端口数组中currentDnatPorts := append(dnatPorts, kubeletSecurePort, kubeletInsecurePort)// 从k8s查询出所有不带有agent标记的节点，也就是所有云端节点的internal IPnodesIP := im.getIPOfNodesWithoutAgent()//和上一次缓存的节点internal IP数组进行比对，分出哪些节点新增的，哪些节点是已删除了的，内部源码略去nodesChanged, addedNodesIP, deletedNodesIP := im.getAddedAndDeletedNodes(nodesIP)//将回环境地址127.0.0.1加入到端口数组中currentNodesIP := append(nodesIP, loopbackAddr)// 更新iptables规则err = im.ensurePortsIptables(currentDnatPorts, deletedDnatPorts, currentNodesIP, deletedNodesIP)if err != nil {klog.Errorf("failed to ensurePortsIptables: %v", err)return}// 如果端口有变化，更新端口数组缓存lastDnatPorts if portsChanged {im.lastDnatPorts = dnatPortsif len(deletedDnatPorts) != 0 {// 针对删除的DNAT端口，需要清除conntrack相关的规则im.clearConnTrackEntries(currentNodesIP, deletedDnatPorts)}klog.Infof("dnat ports changed, %v", dnatPorts)}// 如果云端节点internal IP有变化，更新云端节点internal IP数组lastNodesIPif nodesChanged {im.lastNodesIP = nodesIPim.clearConnTrackEntries(append(addedNodesIP, deletedNodesIP...), currentDnatPorts)klog.Infof("directly access nodes changed, %v for ports %v", nodesIP, currentDnatPorts)}
}
// 从k8s里指定命名空间里查询出指定名称的configMap对象，存有DNAT端口映射对，返回需要进行DANT的端口数组。也就是说，除10250和10255之外的边缘侧端口，如果还需要进行DNAT转发，就在这里配置
func GetConfiguredDnatPorts(client clientset.Interface, insecurePort string) ([]string, error) {ports := make([]string, 0)c, err := client.CoreV1().ConfigMaps(YurttunnelServerDnatConfigMapNs).Get(context.Background(), YurttunnelServerDnatConfigMapName, metav1.GetOptions{})if err != nil {if apierrors.IsNotFound(err) {return nil, fmt.Errorf("configmap %s/%s is not found",YurttunnelServerDnatConfigMapNs,YurttunnelServerDnatConfigMapName)} else {return nil, fmt.Errorf("fail to get configmap %s/%s: %v",YurttunnelServerDnatConfigMapNs,YurttunnelServerDnatConfigMapName, err)}}pairStr, ok := c.Data[yurttunnelServerDnatDataKey]if !ok || len(pairStr) == 0 {return ports, nil}portsPair := strings.Split(pairStr, ",")for _, pair := range portsPair {portPair := strings.Split(pair, "=")if len(portPair) == 2 && portPair[1] == insecurePort && len(portPair[0]) != 0 {if portPair[0] != "10250" && portPair[0] != "10255" {ports = append(ports, portPair[0])}}}return ports, nil
}
//从k8s查询出所有不带有agent标记的节点，也就是所有云端节点的internal IP
func (im *iptablesManager) getIPOfNodesWithoutAgent() []string {var nodesIP []stringnodes, err := im.nodeInformer.Lister().List(labels.Everything())if err != nil {klog.Errorf("failed to list nodes for iptables: %v", err)return nodesIP}for i := range nodes {if withoutAgent(nodes[i]) && isNodeReady(nodes[i]) {nodeIPs := getNodeInternalIPs(nodes[i])nodesIP = append(nodesIP, nodeIPs...)}}klog.V(4).Infof("nodes without %s: %s", projectinfo.GetAgentName(), strings.Join(nodesIP, ","))metrics.Metrics.ObserveCloudNodes(len(nodesIP))return nodesIP
}
var (tunnelCommentStr = strings.ReplaceAll(projectinfo.GetTunnelName(), "-", " ")//全局变量：iptables跳转规则链数组，初始化一条跳转规则：在DNAT表的OUTPUT链，tcp协议数据包跳到yurttunnelServerPortChain子链处理iptablesJumpChains = []iptablesJumpChain{{table: iptables.TableNAT,dstChain: yurttunnelServerPortChain,srcChain: iptables.ChainOutput,comment: fmt.Sprintf("%s server port", tunnelCommentStr),extraArgs: []string{"-p", "tcp"},},}
)func (im *iptablesManager) ensurePortsIptables(currentPorts, deletedPorts, currentIPs, deletedIPs []string) error {//针对每个端口,确保有一条从yurttunnelServerPortChain子链跳到该端口号作为后缀命名的二级子链jumpChains := iptablesJumpChainsfor _, port := range currentPorts {jumpChains = append(jumpChains, iptablesJumpChain{table: iptables.TableNAT,dstChain: iptables.Chain(fmt.Sprintf("%s%s", yurttunnelPortChainPrefix, port)),srcChain: yurttunnelServerPortChain,comment: fmt.Sprintf("jump to port %s", port),extraArgs: []string{"-p", "tcp", "--dport", port},})}if err := im.ensureJumpChains(jumpChains); err != nil {klog.Errorf("Failed to ensure jump chain, %v", err)return err}// 针对每个端口，确保二级子链中的规则，于是重点就在ensurePortIptables函数中for _, port := range currentPorts {err := im.ensurePortIptables(port, currentIPs, deletedIPs)if err != nil {return err}}if len(deletedPorts) == 0 {return nil}// 如果某个端口被删除，则清除相应的二级子链var deletedJumpChains []iptablesJumpChainfor _, port := range deletedPorts {deletedJumpChains = append(deletedJumpChains, iptablesJumpChain{table: iptables.TableNAT,dstChain: iptables.Chain(fmt.Sprintf("%s%s", yurttunnelPortChainPrefix, port)),srcChain: yurttunnelServerPortChain,comment: fmt.Sprintf("jump to port %s", port),extraArgs: []string{"-p", "tcp", "--dport", port},})}if err := im.deleteJumpChains(deletedJumpChains); err != nil {klog.Errorf("Failed to delete jump chain, %v", err)return err}return nil
}
func (im *iptablesManager) ensurePortIptables(port string, currentIPs, deletedIPs []string) error {//port 对应的二级子链名portChain := iptables.Chain(fmt.Sprintf("%s%s", yurttunnelPortChainPrefix, port))//如果没有云端节点，则清除该二级链下的所有规则，实际上这个条件应该不会出现true的情况if len(currentIPs) == 0 {_ = im.iptables.FlushChain(iptables.TableNAT, portChain)return nil}// 确保port对应的二级子链存在，没有则添加if _, err := im.iptables.EnsureChain(iptables.TableNAT, portChain); err != nil {klog.Errorf("could not ensure chain for tunnel server port(%s), %v", port, err)return err}//proxyDest用来放本地DNAT重定向目的地址。 如果API请求端口是10250，那么代理地址就是本地https监听地址，否则是本地http监听地址proxyDest := im.insecureDnatDestif port == kubeletSecurePort {proxyDest = im.secureDnatDest}// 针对所有现存云端节点的internal IP, tcp协议数据包全部直接RETURN操作,即不做DNAT处理。也就是说，如果API请求的目的IP是云端节点，那就不进行DNAT转发。for _, ip := range currentIPs {reqReturnPortIptablesArgs := reqReturnIptablesArgs(reqReturnComment, port, ip)_, err := im.iptables.EnsureRule(iptables.Prepend,iptables.TableNAT, portChain, reqReturnPortIptablesArgs...)if err != nil {klog.Errorf("could not ensure -j RETURN iptables rule for %s:%s: %v", ip, port, err)return err}}// 其它的tcp协议数据包，重定向到代理地址。也就是说，如果API请求的目的IP是边缘节点，那就进行DNAT转发到本地DNAT重定向目的地址，进而通过隧道转发到边缘dnatPortIptablesArgs := dnatIptablesArgs(dnatToTunnelComment, port, proxyDest)_, err := im.iptables.EnsureRule(iptables.Append,iptables.TableNAT, portChain, dnatPortIptablesArgs...)if err != nil {klog.Errorf("could not ensure dnat iptables rule for %s, %v", port, err)return err}//针对所有已删除云端节点的internal IP，删除对应的RETURN规则。for _, ip := range deletedIPs {deletedIPIptablesArgs := reqReturnIptablesArgs(reqReturnComment, port, ip)err = im.iptables.DeleteRule(iptables.TableNAT,portChain, deletedIPIptablesArgs...)if err != nil {klog.Errorf("could not delete old iptables rules for %s:%s: %v", ip, port, err)return err}}return nil
}
//公共规则参数，也就是匹配API请求的目的端口
func iptablesCommonArgs(msg, destPort string, destIP net.IP) []string {args := []string{"-p", "tcp","-m", "comment",}if len(msg) != 0 {args = append(args, "--comment", msg)}if len(destPort) != 0 {args = append(args, "--dport", destPort)}if destIP != nil {ip := toCIDR(destIP)args = append(args, "-d", ip)}return args
}//DNAT操作规则参数
func dnatIptablesArgs(msg, destPort, proxyDest string) []string {args := iptablesCommonArgs(msg, destPort, nil)args = append(args, "-j", "DNAT", "--to-destination", proxyDest)return args
}//RETURN操作规则参数，非DNAT
func reqReturnIptablesArgs(msg, destPort, ip string) []string {destIP := net.ParseIP(ip)args := iptablesCommonArgs(msg, destPort, destIP)args = append(args, "-j", "RETURN")return args
}
总结
- 定时从k8s获取云端节点信息（主要是节点的internal IP）列表和需要代理转发的目的端口（10250和10255不需要在configMap中配置，其它端口需要在在configMap中配置）列表，进行iptables规则更新。
- 凡是目的地址是云端节点的tcp包都不做DNAT处理，非云端节点的tcp包就做DNAT处理进而交由tunel转发到边缘节点。
这篇关于open yurt之yurt-tunel-server iptables规则源码分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
