open yurt之yurt-tunel-server iptables规则源码分析

2023-11-08 00:18
文章标签 分析 源码 规则 open server iptables yurt tunel

本文主要是介绍open yurt之yurt-tunel-server iptables规则源码分析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

简介

yurt-tunel-server是open yurt开源的用于转发来自K8s API server的包含logs、exec和metrics等运维指令请求到边缘节点yurt-tunel-agent的反向代理。
image.png

源码分析

版本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规则源码分析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/366916。

相关文章

python使用Akshare与Streamlit实现股票估值分析教程（图文代码）

python使用Akshare与Streamlit实现股票估值分析教程（图文代码）

《python使用Akshare与Streamlit实现股票估值分析教程（图文代码）》入职测试中的一道题,要求:从Akshare下载某一个股票近十年的财务报表包括,资产负债表,利润表,现金流量表,保存... 目录一、前言二、核心知识点梳理1、Akshare数据获取2、Pandas数据处理3、Matplotl
阅读更多...
python panda库从基础到高级操作分析

python panda库从基础到高级操作分析

《pythonpanda库从基础到高级操作分析》本文介绍了Pandas库的核心功能,包括处理结构化数据的Series和DataFrame数据结构,数据读取、清洗、分组聚合、合并、时间序列分析及大数据... 目录1. Pandas 概述2. 基本操作：数据读取与查看3. 索引操作：精准定位数据4. Group
阅读更多...
MySQL中EXISTS与IN用法使用与对比分析

MySQL中EXISTS与IN用法使用与对比分析

《MySQL中EXISTS与IN用法使用与对比分析》在MySQL中,EXISTS和IN都用于子查询中根据另一个查询的结果来过滤主查询的记录,本文将基于工作原理、效率和应用场景进行全面对比... 目录一、基本用法详解1. IN 运算符2. EXISTS 运算符二、EXISTS 与 IN 的选择策略三、性能对比
阅读更多...
SQL Server跟踪自动统计信息更新实战指南

SQL Server跟踪自动统计信息更新实战指南

《SQLServer跟踪自动统计信息更新实战指南》本文详解SQLServer自动统计信息更新的跟踪方法,推荐使用扩展事件实时捕获更新操作及详细信息,同时结合系统视图快速检查统计信息状态,重点强调修... 目录SQL Server 如何跟踪自动统计信息更新：深入解析与实战指南 核心跟踪方法1️⃣ 利用系统目录
阅读更多...
MySQL 内存使用率常用分析语句

MySQL 内存使用率常用分析语句

《MySQL内存使用率常用分析语句》用户整理了MySQL内存占用过高的分析方法,涵盖操作系统层确认及数据库层bufferpool、内存模块差值、线程状态、performance_schema性能数据... 目录一、 OS层二、 DB层1. 全局情况2. 内存占js用详情最近连续遇到mysql内存占用过高导致
阅读更多...
深度解析Nginx日志分析与499状态码问题解决

深度解析Nginx日志分析与499状态码问题解决

《深度解析Nginx日志分析与499状态码问题解决》在Web服务器运维和性能优化过程中,Nginx日志是排查问题的重要依据,本文将围绕Nginx日志分析、499状态码的成因、排查方法及解决方案展开讨论... 目录前言1. Nginx日志基础1.1 Nginx日志存放位置1.2 Nginx日志格式2. 499
阅读更多...
SQL Server 中的 WITH (NOLOCK) 示例详解

SQL Server 中的 WITH (NOLOCK) 示例详解

《SQLServer中的WITH(NOLOCK)示例详解》SQLServer中的WITH(NOLOCK)是一种表提示,等同于READUNCOMMITTED隔离级别,允许查询在不获取共享锁的情... 目录SQL Server 中的 WITH (NOLOCK) 详解一、WITH (NOLOCK) 的本质二、工作
阅读更多...
SQL Server安装时候没有中文选项的解决方法

SQL Server安装时候没有中文选项的解决方法

《SQLServer安装时候没有中文选项的解决方法》用户安装SQLServer时界面全英文,无中文选项,通过修改安装设置中的国家或地区为中文中国,重启安装程序后界面恢复中文,解决了问题,对SQLSe... 你是不是在安装SQL Server时候发现安装界面和别人不同，并且无论如何都没有中文选项？这个问题也
阅读更多...
Olingo分析和实践之EDM 辅助序列化器详解(最佳实践)

Olingo分析和实践之EDM 辅助序列化器详解(最佳实践)

《Olingo分析和实践之EDM辅助序列化器详解(最佳实践)》EDM辅助序列化器是ApacheOlingoOData框架中无需完整EDM模型的智能序列化工具,通过运行时类型推断实现灵活数据转换,适用... 目录概念与定义什么是 EDM 辅助序列化器？核心概念设计目标核心特点1. EDM 信息可选2. 智能类
阅读更多...
Olingo分析和实践之OData框架核心组件初始化(关键步骤)

Olingo分析和实践之OData框架核心组件初始化(关键步骤)

《Olingo分析和实践之OData框架核心组件初始化(关键步骤)》ODataSpringBootService通过初始化OData实例和服务元数据,构建框架核心能力与数据模型结构,实现序列化、URI... 目录概述第一步：OData实例创建1.1 OData.newInstance() 详细分析1.1.1
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2