kubernetes用户态TCP代理实现
在k8s中针对service的访问通常基于kube proxy实现负载均衡,今天我们来探索下基于用户态的TCP代理组件的工业级实现核心设计, 其中包括随机端口生成器、TCP流复制等技术的核心实现
1. 基础筑基
今天主要是聊用户态的转发,而基于内核态的先不聊
1.1 流量重定向
流量重定向通常是指通过内核的netfilter来对数据包进行拦截,将其定向到我们指定的端口,实现对流量的劫持,从而针对流量里面的一些数据包进行一些额外的处理,这个过程对应用来说是完全透明的
1.1.1 目的地址重定向
目的地址重定向是指将针对某个IP或者某个端口的流量,进行重定向,从而实现流量发送的处理,在kube proxy中主要是通过REDIRECT来实现
1.1.2 目标地址转换
目标地址转换主要是指针对REDIRECT出去返回的流量,需要做一个重定向操作,即将其地址返回给本地的代理服务,由本地的代理服务再去实现转发给真正的应用
1.2 TCP代理实现
1.2.1 随机端口
随机端口是指我们要为为对应的Service建立一个一个临时的代理服务器,该代理服务器需要随机选择一个本地端口进行监听
1.2.2 流复制
代理服务器需要将要本地服务发送的数据复制的目标服务, 同时接收目标服务返回的数据,复制给本地服务
2. 核心设计实现
2.1 随机端口分配器
2.1.1 核心数据结构
分配器的核心数据结构主要是通过rand提供随机数生成器来进行端口的随机获取,并通过采用位计数方式来进行端口使用状态的记录, 获取的随机端口则会放入到ports中提供给进行端口的获取
type rangeAllocator struct { net.PortRange ports chan int // 保存当前可用的随机端口 used big.Int // 位计数,记录端口的使用状态 lock sync.Mutex rand *rand.Rand // 随机数生成器,生成随机端口 }
2.1.2 构建随机分配器
随机分配器主要是构建rand随机数生成器通过当前的时间作为随机因子,并构建ports缓冲buffer,当前是16个
ra := &rangeAllocator{ PortRange: r, ports: make(chan int, portsBufSize), rand: rand.New(rand.NewSource(time.Now().UnixNano())), }
2.1.3 随机数发生器
随机数发生器主要是通过调用nextFreePort来进行随机端口的生成,并放入到ports chan中
func (r *rangeAllocator) fillPorts() { for { // 获取一个随机端口 if !r.fillPortsOnce() { return } } } func (r *rangeAllocator) fillPortsOnce() bool { port := r.nextFreePort() // 获取当前随机端口 if port == -1 { return false } r.ports <- port // 将获取的随机端口放入到缓冲buffer中 return true }
2.1.4 portrange
PortRange是指端口随机的范围,支持单个端口、min-max区间、min+offset区间三种设置方式, 通过其构建Base和Size参数,供算法使用
switch notation { case SinglePortNotation: var port int port, err = strconv.Atoi(value) if err != nil { return err } low = port high = port case HyphenNotation: low, err = strconv.Atoi(value[:hyphenIndex]) if err != nil { return err } high, err = strconv.Atoi(value[hyphenIndex+1:]) if err != nil { return err } case PlusNotation: var offset int low, err = strconv.Atoi(value[:plusIndex]) if err != nil { return err } offset, err = strconv.Atoi(value[plusIndex+1:]) if err != nil { return err } high = low + offset default: return fmt.Errorf("unable to parse port range: %s", value) } pr.Base = low pr.Size = 1 + high - low
2.1.5 随机端口生成算法
随机端口的生成主要是通过先生成随机数,如果端口已经被使用,则会按照高低两个区间进行顺序搜索,直到找到未被占用的端口
func (r *rangeAllocator) nextFreePort() int { r.lock.Lock() defer r.lock.Unlock() // 随机选择port j := r.rand.Intn(r.Size) if b := r.used.Bit(j); b == 0 { r.used.SetBit(&r.used, j, 1) return j + r.Base } // search sequentially // 如果当前端口已经被占用,则从当前的随机数顺序递增查找,如果找到就将对应的bit位设置为1 for i := j + 1; i < r.Size; i++ { if b := r.used.Bit(i); b == 0 { r.used.SetBit(&r.used, i, 1) return i + r.Base } } // 如果高端端口已经被占用则会从低地址开始顺序查找 for i := 0; i < j; i++ { if b := r.used.Bit(i); b == 0 { r.used.SetBit(&r.used, i, 1) return i + r.Base } } return -1 }
2.2 TCP代理实现
2.2.1 核心数据结构
tcp代理实现上基于net.Listener构建一个tcp的监听器, port则是监听的地址
type tcpProxySocket struct { net.Listener port int }
2.2.2 接收本地重定向流量构建转发器
tcp代理接收到链接请求,则会accept然后根据Service和loadbalancer算法选择一台后端pod来进行链接的建立,然后启动异步流量复制,实现流向的双向复制
func (tcp *tcpProxySocket) ProxyLoop(service proxy.ServicePortName, myInfo *ServiceInfo, loadBalancer LoadBalancer) { for { if !myInfo.IsAlive() { // The service port was closed or replaced. return } // 等待接收链接 inConn, err := tcp.Accept() if err != nil { if isTooManyFDsError(err) { panic("Accept failed: " + err.Error()) } if isClosedError(err) { return } if !myInfo.IsAlive() { // Then the service port was just closed so the accept failure is to be expected. return } klog.Errorf("Accept failed: %v", err) continue } klog.V(3).Infof("Accepted TCP connection from %v to %v", inConn.RemoteAddr(), inConn.LocalAddr()) // 根据目标地址的信息和负载均衡算法来选则后端一台server进行链接的建立 outConn, err := TryConnectEndpoints(service, inConn.(*net.TCPConn).RemoteAddr(), "tcp", loadBalancer) if err != nil { klog.Errorf("Failed to connect to balancer: %v", err) inConn.Close() continue } // 启动异步流量复制 go ProxyTCP(inConn.(*net.TCPConn), outConn.(*net.TCPConn)) } }
2.2.3 TCP流量双向复制
TCP流量双向复制则是将流量进行双向的拷贝,通过io.Copy在底层完成从输入流和输出流的流量复制
func ProxyTCP(in, out *net.TCPConn) { var wg sync.WaitGroup wg.Add(2) klog.V(4).Infof("Creating proxy between %v <-> %v <-> %v <-> %v", in.RemoteAddr(), in.LocalAddr(), out.LocalAddr(), out.RemoteAddr()) // 实现流量想的双向复制 go copyBytes("from backend", in, out, &wg) go copyBytes("to backend", out, in, &wg) wg.Wait() } func copyBytes(direction string, dest, src *net.TCPConn, wg *sync.WaitGroup) { defer wg.Done() klog.V(4).Infof("Copying %s: %s -> %s", direction, src.RemoteAddr(), dest.RemoteAddr()) // 实现底层的流量的拷贝 n, err := io.Copy(dest, src) if err != nil { if !isClosedError(err) { klog.Errorf("I/O error: %v", err) } } klog.V(4).Infof("Copied %d bytes %s: %s -> %s", n, direction, src.RemoteAddr(), dest.RemoteAddr()) dest.Close() src.Close() }
2.2.4 通过负载均衡与重试机制实现链接建立
建立链接时有两个核心的设计即sessionAffinity机制与超时重试机制 亲和性机制:如果之前的亲和性链接存在依然有效,则会使用,如果发生断开,则需要重置亲和性 超时重试机制:这里其实可以采用backoff机制来进行超时重试机制的实现
func TryConnectEndpoints(service proxy.ServicePortName, srcAddr net.Addr, protocol string, loadBalancer LoadBalancer) (out net.Conn, err error) { // 默认是不重置亲和性,即之前建立的亲和性链接此时依然有效 sessionAffinityReset := false // EndpointDialTimeouts = []time.Duration{250 * time.Millisecond, 500 * time.Millisecond, 1 * time.Second, 2 * time.Second} // 是一种延迟重试机制,是一种等待重试的机制,减少因为网络不稳定导致的瞬间重试全部失败的情况 for _, dialTimeout := range EndpointDialTimeouts { // 通过负载均衡算法和亲和性选择一台endpoint来进行链接 endpoint, err := loadBalancer.NextEndpoint(service, srcAddr, sessionAffinityReset) if err != nil { klog.Errorf("Couldn't find an endpoint for %s: %v", service, err) return nil, err } klog.V(3).Infof("Mapped service %q to endpoint %s", service, endpoint) // 建立底层的链接 outConn, err := net.DialTimeout(protocol, endpoint, dialTimeout) if err != nil { if isTooManyFDsError(err) { panic("Dial failed: " + err.Error()) } klog.Errorf("Dial failed: %v", err) // 如果发生链接失败,则之前的亲和性则可能失败,此时就要重新进行选择节点进行链接 sessionAffinityReset = true continue } return outConn, nil } return nil, fmt.Errorf("failed to connect to an endpoint.") }
好了今天的内容就到这里,文章核心介绍了一种随机端口选择算法的实现,然后剖析了TCP代理的底层实现机制,其核心包括建立链接的亲和性、超时重试,以及TCP流量的复制技术的实现,今天就到这里,希望大家帮忙分享传播,让作者有继续分享写作的动力,谢谢大家
> 微信号:baxiaoshi2020 > 关注公告号阅读更多源码分析文章
> 更多文章关注 www.sreguide.com > 本文由博客一文多发平台 OpenWrite 发布
原文地址:https://my.oschina.net/u/4131034/blog/3163907