tcpHandler | 南风未起｜云原生/Go技术分享博客

# 1.入口

# 1.1.entrypoint

setupServer()初始化traefik服务时，会调用server.NewTCPEntryPoints()构建tcp入口，以监听及处理指定入口的流量。

// NewTCPEntryPoints creates a new TCPEntryPoints.
func NewTCPEntryPoints(...) (TCPEntryPoints, error) {
	...
	serverEntryPointsTCP := make(TCPEntryPoints)
    // 遍历启动入口
	for entryPointName, config := range entryPointsConfig {
        // 获取entrypoint协议
		protocol, err := config.GetProtocol()
		...

		if protocol != "tcp" {
			continue
		}
		...

        // 构建tcp entrypoint入口
		serverEntryPointsTCP[entryPointName], err = NewTCPEntryPoint(...)
		...
	}
	return serverEntryPointsTCP, nil
}

// NewTCPEntryPoint creates a new TCPEntryPoint.
func NewTCPEntryPoint(...) (*TCPEntryPoint, error) {
    ...

    // 构建TCP监听器
	listener, err := buildListener(ctx, configuration)
	...

    // 默认的tcp路由
	rt := &tcprouter.Router{}

    // 请求装饰器
	reqDecorator := requestdecorator.New(hostResolverConfig)

    // 构建http服务器
	httpServer, err := createHTTPServer(ctx, listener, configuration, true, reqDecorator)
	...

    // 注册http forwarder(tcp router转发流量)
	rt.SetHTTPForwarder(httpServer.Forwarder)

    // 构建https服务器
	httpsServer, err := createHTTPServer(ctx, listener, configuration, false, reqDecorator)
	...
    
    // 构建http3服务器(QUIC)
	h3Server, err := newHTTP3Server(ctx, configuration, httpsServer)
	...

    // 注册https forwarder及https muxter路由
	rt.SetHTTPSForwarder(httpsServer.Forwarder)

    // tcp switcher
	tcpSwitcher := &tcp.HandlerSwitcher{}
    
    // 切换到默认router
	tcpSwitcher.Switch(rt)

	return &TCPEntryPoint{
		listener:               listener,
		switcher:               tcpSwitcher,
		transportConfiguration: configuration.Transport,
		tracker:                tracker,
		httpServer:             httpServer,
		httpsServer:            httpsServer,
		http3Server:            h3Server,
	}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

注意

1.tcp listener监听连接，调用tcp switcher进行路由切换

2.tcp switcher作为统一入口，挂载tcp router，router根据路由获取及执行handler

3.tcp router挂载tcp muxer、https muxer和tcp tls muxer以匹配路由，同时挂载http forward转发流量给http server

4.http3服务独立运行，监听udp连接，但会复用https handler

# 1.2.start

eps.Start()会依次启动tcp入口，接受流量及转发到tcp switcher，switcher获取router进行路由匹配，将分发请求到handler。

// Start starts the TCP server.
func (e *TCPEntryPoint) Start(ctx context.Context) {
	...

    // 启动http3服务器
	if e.http3Server != nil {
		go func() { _ = e.http3Server.Start() }()
	}

	for {
        // 监听到连接
		conn, err := e.listener.Accept()
		...

        // 创建writeCloser
		writeCloser, err := writeCloser(conn)
		...

        // 异步响应
		safe.Go(func() {
			...
            // 基于tcp switcher响应
			e.switcher.ServeTCP(newTrackedConnection(writeCloser, e.tracker))
		})
	}
}

// ServeTCP forwards the TCP connection to the current active handler.
func (s *HandlerSwitcher) ServeTCP(conn WriteCloser) {
    // 获取tcp router
	handler := s.router.Get()
    // 转为handler
	h, ok := handler.(Handler)
	if ok {
        // 处理请求
		h.ServeTCP(conn)
	} else {
		conn.Close()
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

注意

tcp switcher类似switcher router的上层代理，配合watcher更新的router实现处理器切换

# 2.路由

# 2.1.switchRouter

watcher监听到配置变更会推送到注册的listener，switchRouter就是用于切换路由信息的listener，由配置变更驱动执行。

func switchRouter(routerFactory *server.RouterFactory, serverEntryPointsTCP server.TCPEntryPoints, serverEntryPointsUDP server.UDPEntryPoints) func(conf dynamic.Configuration) {
	return func(conf dynamic.Configuration) {
		...
		// tcp routers构造
		routers, udpRouters := routerFactory.CreateRouters(rtConf)

        // 切换tcp router
		serverEntryPointsTCP.Switch(routers)
		...
	}
}

// CreateRouters creates new TCPRouters and UDPRouters.
func (f *RouterFactory) CreateRouters(rtConf *runtime.Configuration) (map[string]*tcprouter.Router, map[string]udp.Handler) {
	...
    
	// TCP
	rtTCPManager := tcprouter.NewManager(rtConf, svcTCPManager, middlewaresTCPBuilder, handlersNonTLS, handlersTLS, f.tlsManager)
	routersTCP := rtTCPManager.BuildHandlers(ctx, f.entryPointsTCP)
	...
    
	return routersTCP, routersUDP
}

// Switch the TCP routers.
func (eps TCPEntryPoints) Switch(routersTCP map[string]*tcprouter.Router) {
	for entryPointName, rt := range routersTCP {
        // tcp router切换
		eps[entryPointName].SwitchRouter(rt)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

注意

switcher本质上是加了锁的handler value，封装一层是希望线程安全获取及切换handler及不影响旧的流量

# 2.2.buildHandlers

buildHandlers根据entrypoint生成tcp router，tcp router挂载tcp/http组件，根据匹配策略获取及执行handler.ServeTCP()。

// BuildHandlers builds the handlers for the given entrypoints.
func (m *Manager) BuildHandlers(rootCtx context.Context, entryPoints []string) map[string]*Router {
    // tcp路由
	entryPointsRouters := m.getTCPRouters(rootCtx, entryPoints)
    // http路由(https tls)
	entryPointsRoutersHTTP := m.getHTTPRouters(rootCtx, entryPoints, true)
    
	....
    // 遍历入口
	for _, entryPointName := range entryPoints {
        // 获取tcp路由定义
		routers := entryPointsRouters[entryPointName]
		...
        // 构造tcp router
		handler, err := m.buildEntryPointHandler(ctx, routers, entryPointsRoutersHTTP[entryPointName], m.httpHandlers[entryPointName], m.httpsHandlers[entryPointName])
		...
		entryPointHandlers[entryPointName] = handler
	}
	return entryPointHandlers
}

// tcp router
func (m *Manager) buildEntryPointHandler(ctx context.Context, configs map[string]*runtime.TCPRouterInfo, configsHTTP map[string]*runtime.RouterInfo, handlerHTTP, handlerHTTPS http.Handler) (*Router, error) {
	// 创建入口级别tcp router
	router, err := NewRouter()
	...

    // 设置http handler
	router.SetHTTPHandler(handlerHTTP)
    ...
    
    // sni检查器包装https handler
	sniCheck := snicheck.New(tlsOptionsForHost, handlerHTTPS)

	// 设置https handler
	router.SetHTTPSHandler(sniCheck, defaultTLSConf)
	...

    // 注册tcp handler
	m.addTCPHandlers(ctx, configs, router)

	return router, nil
}

// addTCPHandlers creates the TCP handlers defined in configs, and adds them to router.
func (m *Manager) addTCPHandlers(ctx context.Context,configs map[string]*runtime.TCPRouterInfo,router *Router) {
	// 遍历tcp路由配置
    for routerName, routerConfig := range configs {
		...
		// 解析路由配置的SNI域名
		domains, err := tcpmuxer.ParseHostSNI(routerConfig.Rule)
		...

		// 普通TCP或越过tls的tcp路由
		if routerConfig.TLS == nil || routerConfig.TLS.Passthrough {
            // 构造tcp handler
			handler, err = m.buildTCPHandler(ctxRouter, routerConfig)
			...
		}

		// 普通TCP路由注册    
		if routerConfig.TLS == nil {
			// 向muxerTCP注册tcp muxer
            router.muxerTCP.AddRoute(routerConfig.Rule, routerConfig.RuleSyntax, routerConfig.Priority, handler)
			...
		}

        // TLS PassThrough模式,加密流原封不动转到后端
		if routerConfig.TLS.Passthrough {
			// 向muxerTCPTLS注册tcp muxer
            router.muxerTCPTLS.AddRoute(routerConfig.Rule,routerConfig.RuleSyntax,routerConfig.Priority,handler)
            ...
		}
		...

        // 获取tls配置
		tlsConf, err := m.tlsManager.Get(traefiktls.DefaultTLSStoreName, tlsOptionsName)
		...
		// 构造tcp handler
		handler, err = m.buildTCPHandler(ctxRouter, routerConfig)
		...

        // 包装为tls handler
		handler = &tcp.TLSHandler{
			Next:   handler,
			Config: tlsConf,
		}

        // 向muxerTCPTLS注册tcp tls muxer
		router.muxerTCPTLS.AddRoute(routerConfig.Rule, routerConfig.RuleSyntax, routerConfig.Priority, handler);
        ...
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

注意

tcp route加载会注册http/https handler及tcp handler，以根据muxer维护的AST规则树处理请求

# 2.3.buildTCPHandler

buildTCPHandler()会创建service handler，外层包装middleware chain以结合中间件处理流量，handler的类型为balancer。

func (m *Manager) buildTCPHandler(ctx context.Context, router *runtime.TCPRouterInfo) (tcp.Handler, error) {
	...
    // 构造service handler
	sHandler, err := m.serviceManager.BuildTCP(ctx, router.Service)
	...
	// 构建middleware链
	mHandler := m.middlewaresBuilder.BuildChain(ctx, router.Middlewares)

    // 组合middleware和service
	return tcp.NewChain().Extend(*mHandler).Then(sHandler)
}

// tcp service handler
func (m *Manager) BuildTCP(rootCtx context.Context, serviceName string) (tcp.Handler, error) {
	...
	// 获取service配置
	conf, ok := m.configs[serviceQualifiedName]
	...

	switch {
    // LB类型
	case conf.LoadBalancer != nil:
		loadBalancer := tcp.NewWRRLoadBalancer()
		...

        // 打乱遍历后端池
		for index, server := range shuffle(conf.LoadBalancer.Servers, m.rand) {
			...

            // 向连接池申请tcp连接
			dialer, err := m.dialerManager.Get(conf.LoadBalancer.ServersTransport, server.TLS)
			...

            // 构造tcp proxy
			handler, err := tcp.NewProxy(server.Address, conf.LoadBalancer.ProxyProtocol, dialer)
			...

            // 注册到LB
			loadBalancer.AddServer(handler)
		}

		return loadBalancer, nil

    // 加权负载模式(LB+LB)
	case conf.Weighted != nil:
		loadBalancer := tcp.NewWRRLoadBalancer()

        // 打乱遍历service
		for _, service := range shuffle(conf.Weighted.Services, m.rand) {
            // 递归构造LB
			handler, err := m.BuildTCP(ctx, service.Name)
			...

            // 注册到负载均衡器
			loadBalancer.AddWeightServer(handler, service.Weight)
		}

		return loadBalancer, nil

	default:
		return nil, err
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

注意

针对loadbalancer service会构造balancer handler，针对weighted service会构造balancer+balancer handler

# 2.4.addRoute

addRoute()会关联AST抽象树matcher及handler注册route，后续的路由处理就是根据matcher匹配及执行handler.ServeTCP()。

// AddRoute adds a new route, associated to the given handler, at the given priority, to the muxer.
func (m *Muxer) AddRoute(rule string, syntax string, priority int, handler tcp.Handler) error {
	...

	switch syntax {
	case "v2":
		parse, err = m.parserV2.Parse(rule)
		...

		matcherFuncs = tcpFuncsV2
	default:
		parse, err = m.parser.Parse(rule)
		...
        
		matcherFuncs = tcpFuncs
	}

    // 构造AST抽象语法树
	buildTree, ok := parse.(rules.TreeBuilder)
	...
	ruleTree := buildTree()
	...
	
    // 语法树注册为matcher
	matchers.addRule(ruleTree, matcherFuncs)
	...

	newRoute := &route{
		handler:  handler,
		matchers: matchers,
		catchAll: catchAll,
		priority: priority,
	}
    
	m.routes = append(m.routes, newRoute)

	sort.Sort(m.routes)

	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

注意

tcp router会注册http muxer 和tcp muxer，tcp muxer本质上对应一个route处理器，维护matcher及service handler

# 3.流量处理

# 3.1.router

这里提到的router不是入口初始化的默认值，而是基于watcher推送的配置生成的最新tcp router，router根据路由匹配handler处理流量。

// tcp/http连接处理
func (r *Router) ServeTCP(conn tcp.WriteCloser) {
	// 部分纯TCP协议(Redis)会等服务器先发点什么再动,这种情况clientHelloInfo()会卡死,因此提前处理
	if r.muxerTCP.HasRoutes() && !r.muxerTCPTLS.HasRoutes() && !r.muxerHTTPS.HasRoutes() {
        // 提取连接的SNI/IP/ALPN协议列表
		connData, err := tcpmuxer.NewConnData("", conn, nil)
		...

        // 匹配TCP规则
		handler, _ := r.muxerTCP.Match(connData)
		// 匹配到service handler
		if handler != nil {
			// 取消超时限制
			conn.SetDeadline(time.Time{})
            ...

            // 处理普通tcp请求
			handler.ServeTCP(conn)
			return
		}
	}

    // 下面的处理默认都是client-first协议,否则会卡死
    // 检测Postgres协议(PostgreSQL客户端发起)
	br := bufio.NewReader(conn)
	postgres, err := isPostgres(br)
	...
	if postgres {
		// 取消超时限制
		conn.SetDeadline(time.Time{})
        ...

        // Postgres请求处理
		r.servePostgres(r.GetConn(conn, getPeeked(br)))
		return
	}

    // 获取TLS握手首包
	hello, err := clientHelloInfo(br)
	...

	// 取消超时限制
	conn.SetDeadline(time.Time{})
    ...

    // 提取连接的SNI/IP/ALPN协议列表
	connData, err := tcpmuxer.NewConnData(hello.serverName, conn, hello.protos)
	...

    // 明文请求
	if !hello.isTLS {
       	// 匹配tcp router handler
		handler, _ := r.muxerTCP.Match(connData)
		switch {
        // TCP请求处理
		case handler != nil:
			handler.ServeTCP(r.GetConn(conn, hello.peeked))
        // HTTP请求,由http转发器处理
		case r.httpForwarder != nil:
			r.httpForwarder.ServeTCP(r.GetConn(conn, hello.peeked))
		default:
			conn.Close()
		}
		return
	}

	// Handling ACME-TLS/1 challenges.
	if slices.Contains(hello.protos, tlsalpn01.ACMETLS1Protocol) {
		r.acmeTLSALPNHandler().ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

	// 匹配https路由
	handlerHTTPS, catchAllHTTPS := r.muxerHTTPS.Match(connData)
	if handlerHTTPS != nil && !catchAllHTTPS {
		// https请求处理
		handlerHTTPS.ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

	// TCP TLS
	handlerTCPTLS, catchAllTCPTLS := r.muxerTCPTLS.Match(connData)
	if handlerTCPTLS != nil && !catchAllTCPTLS {
        // tcp tls透传
		handlerTCPTLS.ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

    // https通配路由(*)
	if handlerHTTPS != nil {
		handlerHTTPS.ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

    // tcp tls通配路由(*)
	if handlerTCPTLS != nil {
		handlerTCPTLS.ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

    // https全局路由(走默认证书)
	if r.httpsForwarder != nil {
		r.httpsForwarder.ServeTCP(r.GetConn(conn, hello.peeked))
		return
	}

	conn.Close()
}

注意

这里可以看出，http server较为特殊，依赖tcp router转发流量、切换http handler

# 3.2.nonTLS

muxerTCP.Match()根据请求SNI匹配tcp.handler，执行handler.ServeTCP()处理tcp请求，handler可能由middleware包装。

// ServeTCP forwards the connection to the right service.
func (b *WRRLoadBalancer) ServeTCP(conn WriteCloser) {
	b.lock.Lock()
    // 获取后端(wrrLB/proxy)
	next, err := b.next()
	b.lock.Unlock()
	...

	next.ServeTCP(conn)
}

func (b *WRRLoadBalancer) next() (Handler, error) {
	...

	// 获取服务最大权重
	max := b.maxWeight()
	...

	// 所有服务权重最大公约数
	gcd := b.weightGcd()

	for {
        // 轮询下一个服务器(index初始为-1)
		b.index = (b.index + 1) % len(b.servers)
        // 执行一轮
		if b.index == 0 {
            // 阈值减GCD
			b.currentWeight -= gcd
            // 阈值每轮重置到最大
			if b.currentWeight <= 0 {
				b.currentWeight = max
			}
		}
        // 获取该轮后端
		srv := b.servers[b.index]
        // 满足阈值则返回
		if srv.weight >= b.currentWeight {
			return srv, nil
		}
	}
}

// WRRLB的后端是Proxy/WRRLB,这里以最底层的Proxy分析
// ServeTCP forwards the connection to a service.
func (p *Proxy) ServeTCP(conn WriteCloser) {
	...
	defer conn.Close()

    // 建立到目标后端服务器的TCP连接
	connBackend, err := p.dialBackend()
	...

	// maybe not needed, but just in case
	defer connBackend.Close()
	errChan := make(chan error)

    // HAProxy的proxy protocol支持,将原始客户端传递到后端协议(src,dest)
	if p.proxyProtocol != nil && p.proxyProtocol.Version > 0 && p.proxyProtocol.Version < 3 {
        // 构造proxy header
		header := proxyproto.HeaderProxyFromAddrs(byte(p.proxyProtocol.Version), conn.RemoteAddr(), conn.LocalAddr())
        // 写入
        header.WriteTo(connBackend)
		...
	}

    // 流数据拷贝
	go p.connCopy(conn, connBackend, errChan)
	go p.connCopy(connBackend, conn, errChan)
	...

	<-errChan
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

注意

负载均衡模式balancer后端是tcp proxy，权重模式balancer后端还是balancer，本质上都会执行到proxy.ServeTCP()

# 3.3.withTLS

postgres/tcp tls较为特殊，外层的tls handler用于构造tls conn以进行流量解析，最终还是依赖balancer处理解析后的流量。

// servePostgres serves a connection with a Postgres client negotiating a STARTTLS session.
// It handles TCP TLS routing, after accepting to start the STARTTLS session.
func (r *Router) servePostgres(conn tcp.WriteCloser) {
    // TLS握手提示
	_, err := conn.Write(PostgresStartTLSReply)
	...

    // 读取TLS握手的字节
	br := bufio.NewReader(conn)
	...
	_, err = br.Read(b)
	...

    // 解析TLS握手字节
	hello, err := clientHelloInfo(br)
	...

    // 非TLS连接关闭
	if !hello.isTLS {
		conn.Close()
		return
	}

    // TCP连接元信息
	connData, err := tcpmuxer.NewConnData(hello.serverName, conn, hello.protos)
	...

	// 匹配TCP TLS路由
	handlerTCPTLS, _ := r.muxerTCPTLS.Match(connData)
	...

	// 恢复conn的peek字节
	proxiedConn := r.GetConn(conn, hello.peeked)
	if _, ok := handlerTCPTLS.(*tcp.TLSHandler); !ok {
		proxiedConn = &postgresConn{WriteCloser: proxiedConn}
	}

    // 连接处理
	handlerTCPTLS.ServeTCP(proxiedConn)
}

// ServeTCP terminates the TLS connection.
func (t *TLSHandler) ServeTCP(conn WriteCloser) {
    // TLS终止由tls.Server完成,其内部重写Read和Write以解析TLS流为明文
	t.Next.ServeTCP(tls.Server(conn, t.Config))
}

// ServeTCP forwards the connection to the right service.
func (b *WRRLoadBalancer) ServeTCP(conn WriteCloser) {
	b.lock.Lock()
    // 获取handler
	next, err := b.next()
	b.lock.Unlock()
	...

	next.ServeTCP(conn)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

注意

postgres流量处理会先获取TLS握手首包，根据握手信息匹配路由处理器，TLS数据解析由tlsHandler的serverConn包装连接处理