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

# 1.主流程

# 1.1.原理

traefik配置动态更新依赖watcher、listener、provider和switcher，watcher用于关联listener和provider，listener负责具体配置更新，provider负责配置发现及变更检查，switcher抽象了http.Handler、tcp.Handler逻辑，变更时加锁保证线程安全。
注意

watcher作为中间层关联provider和listener，以感知配置变化及推送给listener处理

# 1.2.start

watcher.Start()核心是三个方法，receiveConfigurations用于接收配置变化发送到消费者，applyConfigurations用于合并及应用配置，startProviderAggregator()用于启动provider聚合器实现服务发现。

// Start the configuration watcher.
func (c *ConfigurationWatcher) Start() {
    // 配置接收及推送
	c.routinesPool.GoCtx(c.receiveConfigurations)
    // 消费修整及转给listener
	c.routinesPool.GoCtx(c.applyConfigurations)
    // 服务发现
	c.startProviderAggregator()
}

1
2
3
4
5
6
7
8
9

receiveConfigurations循环监听配置变更，新的配置会发送到output channel供外部消费，作为配置推送的中转角色进行一些前置处理。

// receiveConfigurations receives configuration changes from the providers.
func (c *ConfigurationWatcher) receiveConfigurations(ctx context.Context) {
	newConfigurations := make(dynamic.Configurations)
	var output chan dynamic.Configurations
	for {
		select {
		case <-ctx.Done():
			return
		// 推送端阻塞的延迟发送
		case output <- newConfigurations.DeepCopy():
			output = nil

		default:
			select {
			case <-ctx.Done():
				return
            // 收到provider配置
			case configMsg, ok := <-c.allProvidersConfigs:
				if !ok {
					return
				}
				...

                // 配置变化无变化
				if reflect.DeepEqual(newConfigurations[configMsg.ProviderName], configMsg.Configuration) {
					continue
				}

                // 有变化,缓存最新的
				newConfigurations[configMsg.ProviderName] = configMsg.Configuration.DeepCopy()

                // output channel指向外层消费端管道
				output = c.newConfigs

			// 最新配置推送
			case output <- newConfigurations.DeepCopy():
				output = 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

applyConfigurations收到最新的配置会合并及应用，其实就是依次推送给listener，由listener负责更具体的配置更新。

// applyConfigurations blocks on a RingChannel that receives and sent as soon as a provider change occurs.
func (c *ConfigurationWatcher) applyConfigurations(ctx context.Context) {
	var lastConfigurations dynamic.Configurations
	for {
		select {
		case <-ctx.Done():
			return
        // 收到新配置
		case newConfigs, ok := <-c.newConfigs:
			if !ok {
				return
			}

			// 等待必须依赖的配置提供者配置变化
			if _, ok := newConfigs[c.requiredProvider]; c.requiredProvider != "" && !ok {
				continue
			}

            // 配置无变化
			if reflect.DeepEqual(newConfigs, lastConfigurations) {
				continue
			}

            // 合并配置(http/tcp/udp/tls)
			conf := mergeConfiguration(newConfigs.DeepCopy(), c.defaultEntryPoints)
            // http model配置合并到route
			conf = applyModel(conf)

            // 推送到listener
			for _, listener := range c.configurationListeners {
				listener(conf)
			}

			lastConfigurations = newConfigs
		}
	}
}

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

startProviderAggregator()用于服务发现，根据声明启动不同的provider，以监听不同资源变化及推送最新状态。

func (c *ConfigurationWatcher) startProviderAggregator() {
	...

	safe.Go(func() {
        // 启动provider监听(不同provider实现差异)
		c.providerAggregator.Provide(c.allProvidersConfigs, c.routinesPool)
		...
	})
}

// Provide calls the provide method of every providers.
func (p ProviderAggregator) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
    // 启动file provider文件监听
	if p.fileProvider != nil {
		p.launchProvider(configurationChan, pool, p.fileProvider)
	}

    // 异步启动其它provider
	for _, prd := range p.providers {
		safe.Go(func() {
			p.launchProvider(configurationChan, pool, prd)
		})
	}

	// 内部provider,用于标记所有provider加载完成
	if p.internalProvider != nil {
		p.launchProvider(configurationChan, pool, p.internalProvider)
	}

	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

注意

watcher关联provider和listener，基于receive和apply动作处理及转发provider监听配置，交给listener更新

# 1.3.launchProvider

launchProvider负责启动provider，监听traefik资源及渲染traefik的动态配置，实时响应及推送资源变更。

func (p ProviderAggregator) launchProvider(configurationChan chan<- dynamic.Message, pool *safe.Pool, prd provider.Provider) {
	...
	// 节流启动crd provider
	maybeThrottledProvide(prd, p.providersThrottleDuration)(configurationChan, pool)
    ...
}

// maybeThrottledProvide returns the Provide method of the given provider.
func maybeThrottledProvide(prd provider.Provider, defaultDuration time.Duration) func(chan<- dynamic.Message, *safe.Pool) error {
    // 节流时间配置(默认2s)
	providerThrottleDuration := defaultDuration
    // 部分provider支持自定义配置节流时间
	if throttled, ok := prd.(throttled); ok {
		// per-provider throttling
		providerThrottleDuration = throttled.ThrottleDuration()
	}

    // 未设置节流,返回原始provider
	if providerThrottleDuration == 0 {
		// throttling disabled
		return prd.Provide
	}

    // 设置节流,返回包装provider
	return func(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
        // 环形channel
		rc := newRingChannel()
        // 开启协程,根据节流间隔推送
		pool.GoCtx(func(ctx context.Context) {
			for {
				select {
				case <-ctx.Done():
					return
				case msg := <-rc.out():
					configurationChan <- msg
					time.Sleep(providerThrottleDuration)
				}
			}
		})
		
        // 启动provider
		return prd.Provide(rc.in(), pool)
	}
}

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

注意

launchProvider是通用加载器，会根据节流配置实时或延迟推送

# 2.提供者

# 2.1.crd实现

crd.Provide()会初始化k8s client客户端，基于客户端创建事件管道，watchAll所有配置的变化，根据信号加载监听资源的最新状态。

// Provide allows the k8s provider to provide configurations to traefik
// using the given configuration channel.
func (p *Provider) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
	...

    // 初始化k8s客户端
	k8sClient, err := p.newK8sClient(ctxLog)
	...

    // 启动协程异步监听
	pool.GoCtx(func(ctxPool context.Context) {
		operation := func() error {
            // 监听crd事件(基于不同informer)
			eventsChan, err := k8sClient.WatchAll(p.Namespaces, ctxPool.Done())
			...

            // 初始化节流管道
			throttleDuration := time.Duration(p.ThrottleDuration)
			throttledChan := throttleEvents(ctxLog, throttleDuration, pool, eventsChan)
            // 事件管道替换
			if throttledChan != nil {
				eventsChan = throttledChan
			}

			for {
				select {
				case <-ctxPool.Done():
					return nil
                // 监听到配置变更
				case event := <-eventsChan:
					// 加载资源最新状态
					conf := p.loadConfigurationFromCRD(ctxLog, k8sClient)
					// 计算配置hash
					confHash, err := hashstructure.Hash(conf, nil)
					switch {
					case err != nil:
						logger.Error().Err(err).Msg("Unable to hash the configuration")
                    // 配置一致不进行推送
					case p.lastConfiguration.Get() == confHash:
						logger.Debug().Msgf("Skipping Kubernetes event kind %T", event)
					default:
                        // 记录本次配置hash
						p.lastConfiguration.Set(confHash)
                        // 推送配置
						configurationChan <- dynamic.Message{
							ProviderName:  providerName,
							Configuration: conf,
						}
					}

                    // 节流等待(2s)
					time.Sleep(throttleDuration)
				}
			}
		}
		...
		
        // 启动监听(自动重试)
		backoff.RetryNotify(safe.OperationWithRecover(operation), backoff.WithContext(job.NewBackOff(backoff.NewExponentialBackOff()), ctxPool), notify)
		...
	})

	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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

loadConfigurationFromCRD()根据配置变更加载最新规则状态，渲染route-middleware-service-ServersTransport配置。

func (p *Provider) loadConfigurationFromCRD(ctx context.Context, client Client) *dynamic.Configuration {
	// 基于crd及secret构建tls store和tls config
    stores, tlsConfigs := buildTLSStores(ctx, client)
	...

    // 加载ingressRoute资源(http/tcp/udp)
	conf := &dynamic.Configuration{
		HTTP: p.loadIngressRouteConfiguration(ctx, client, tlsConfigs),
		TCP:  p.loadIngressRouteTCPConfiguration(ctx, client, tlsConfigs),
		UDP:  p.loadIngressRouteUDPConfiguration(ctx, client),
		TLS: &dynamic.TLSConfiguration{
			Options: buildTLSOptions(ctx, client),
			Stores:  stores,
		},
	}

	// 补充tls证书
	conf.TLS.Certificates = getTLSConfig(tlsConfigs)

    // 加载http中间件
	for _, middleware := range client.GetMiddlewares() {
		...

        // 加载http中间件
		conf.HTTP.Middlewares[id] = &dynamic.Middleware{
			...
		}
	}

	// 加载tcp中间件
	for _, middlewareTCP := range client.GetMiddlewareTCPs() {
		id := provider.Normalize(makeID(middlewareTCP.Namespace, middlewareTCP.Name))

		conf.TCP.Middlewares[id] = &dynamic.TCPMiddleware{
			InFlightConn: middlewareTCP.Spec.InFlightConn,
			IPWhiteList:  middlewareTCP.Spec.IPWhiteList,
			IPAllowList:  middlewareTCP.Spec.IPAllowList,
		}
	}
	...

    // 构建traefik service后端池
	for _, service := range client.GetTraefikServices() {
        // 注册http service
		cb.buildTraefikService(ctx, service, conf.HTTP.Services)
		...
	}

    // 加载http serversTransport配置(根据关联secret获取rootCA/cert/forwardTimeout)
	for _, serversTransport := range client.GetServersTransports() {
		...

        // 记录到http serversTransport
		conf.HTTP.ServersTransports[id] = &dynamic.ServersTransport{
			ServerName:          serversTransport.Spec.ServerName,
			InsecureSkipVerify:  serversTransport.Spec.InsecureSkipVerify,
			RootCAs:             rootCAs,
			Certificates:        certs,
			DisableHTTP2:        serversTransport.Spec.DisableHTTP2,
			MaxIdleConnsPerHost: serversTransport.Spec.MaxIdleConnsPerHost,
			ForwardingTimeouts:  forwardingTimeout,
			PeerCertURI:         serversTransport.Spec.PeerCertURI,
			Spiffe:              serversTransport.Spec.Spiffe,
		}
	}

    // 加载tcp serversTransport(dialTimeout/dialKeepAlive/terminationDelay/tls)
	for _, serversTransportTCP := range client.GetServersTransportTCPs() {
		...

        // tls配置
		if serversTransportTCP.Spec.TLS != nil {
			...
            
            // tcpServerTransport关联tls
			tcpServerTransport.TLS = &dynamic.TLSClientConfig{
				ServerName:         serversTransportTCP.Spec.TLS.ServerName,
				InsecureSkipVerify: serversTransportTCP.Spec.TLS.InsecureSkipVerify,
				RootCAs:            rootCAs,
				Certificates:       certs,
				PeerCertURI:        serversTransportTCP.Spec.TLS.PeerCertURI,
			}

            // 记录Spiffe配置
			tcpServerTransport.TLS.Spiffe = serversTransportTCP.Spec.TLS.Spiffe
		}

        // 记录到tcpServerTransport配置
		id := provider.Normalize(makeID(serversTransportTCP.Namespace, serversTransportTCP.Name))
		conf.TCP.ServersTransports[id] = &tcpServerTransport
	}

	return conf
}

注意

1.loadConfigurationFromCRD()触发会加载完整的路由配置

2.crd涉及到的资源都是从informer缓存获取

# 2.2.ingress实现

ingress provider会监听支持的ingress资源，根据ingress声明渲染出最新的配置进行推送，实现上与crd模式类型。

// Provide allows the k8s provider to provide configurations to traefik
// using the given configuration channel.
func (p *Provider) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
	...

	k8sClient, err := p.newK8sClient(ctxLog)
	...

    // 异步监听与渲染
	pool.GoCtx(func(ctxPool context.Context) {
		operation := func() error {
            // 基于informer监听资源
			eventsChan, err := k8sClient.WatchAll(p.Namespaces, ctxPool.Done())
			...
			
            // 限流管道包装
			throttleDuration := time.Duration(p.ThrottleDuration)
			throttledChan := throttleEvents(ctxLog, throttleDuration, pool, eventsChan)
			if throttledChan != nil {
				eventsChan = throttledChan
			}

			for {
				select {
				case <-ctxPool.Done():
					return nil
                // 配置变更
				case event := <-eventsChan:
					// 加载ingress配置
					conf := p.loadConfigurationFromIngresses(ctxLog, k8sClient)

                    // 计算配置hash
					confHash, err := hashstructure.Hash(conf, nil)
					switch {
					case err != nil:
						logger.Error().Msg("Unable to hash the configuration")
					case p.lastConfiguration.Get() == confHash:
						logger.Debug().Msgf("Skipping Kubernetes event kind %T", event)
                    // 配置不一致,推送
					default:
                        // 记录本次配置hash
						p.lastConfiguration.Set(confHash)
                        // 推送到外部
						configurationChan <- dynamic.Message{
							ProviderName:  "kubernetes",
							Configuration: conf,
						}
					}

					// 间隔2s触发下一次循环
					time.Sleep(throttleDuration)
				}
			}
		}
		...

        // 启动监听(带重试)
		backoff.RetryNotify(safe.OperationWithRecover(operation), backoff.WithContext(job.NewBackOff(backoff.NewExponentialBackOff()), ctxPool), notify)
        ...
	})

	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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

loadConfigurationFromIngresses()用于获取ingress，关联middleware和service，渲染最新的配置推送给listener同步。

func (p *Provider) loadConfigurationFromIngresses(ctx context.Context, client Client) *dynamic.Configuration {
	conf := &dynamic.Configuration{
		HTTP: &dynamic.HTTPConfiguration{
			Routers:     map[string]*dynamic.Router{},
			Middlewares: map[string]*dynamic.Middleware{},
			Services:    map[string]*dynamic.Service{},
		},
		TCP: &dynamic.TCPConfiguration{},
	}
	...

    // 获取允许的ingressClass
	if !p.DisableIngressClassLookup {
		ics, err := client.GetIngressClasses()
		...

		if p.IngressClass != "" {
			ingressClasses = filterIngressClassByName(p.IngressClass, ics)
		} else {
			ingressClasses = ics
		}
	}

	ingresses := client.GetIngresses()

	...
    // 遍历ingress
	for _, ingress := range ingresses {
		...

        // 检查ingress准入
		if !p.shouldProcessIngress(ingress, ingressClasses) {
			continue
		}

        // 解析ingress annotation配置
        // traefik.ingress.kubernetes.io/service.sticky.0.cookie.name --> traefik.service.sticky.0.cookie.name
        // traefik.service.sticky.0.cookie.name -> traefik.service.sticky[0].cookie.name
		rtConfig, err := parseRouterConfig(ingress.Annotations)
		...

        // 解析声明证书
		getCertificates(ctx, ingress, client, certConfigs)
		...

        // 默认后端处理
		if len(ingress.Spec.Rules) == 0 && ingress.Spec.DefaultBackend != nil {
            // 限制默认后端唯一
			if _, ok := conf.HTTP.Services["default-backend"]; ok {
				continue
			}

            // 加载后端
			service, err := p.loadService(client, ingress.Namespace, *ingress.Spec.DefaultBackend)
			...

            // 构造默认route
			rt := &dynamic.Router{
				Rule:       "PathPrefix(`/`)",
				RuleSyntax: "v3",
				Priority:   math.MinInt32,
				Service:    "default-backend",
			}

            // 更新默认路由配置
			if rtConfig != nil && rtConfig.Router != nil {
				rt.EntryPoints = rtConfig.Router.EntryPoints
				rt.Middlewares = rtConfig.Router.Middlewares
				rt.TLS = rtConfig.Router.TLS
			}
			...

			conf.HTTP.Routers["default-router"] = rt
			conf.HTTP.Services["default-backend"] = service
		}

		routers := map[string][]*dynamic.Router{}

        // rule路由加载
		for _, rule := range ingress.Spec.Rules {
            // 更新ingress地址
			p.updateIngressStatus(ingress, client)
			...

            // 匹配规则处理
			for _, pa := range rule.HTTP.Paths {
                // 加载后端池
				service, err := p.loadService(client, ingress.Namespace, pa.Backend)
				...
			
                // 加载port
				portString := pa.Backend.Service.Port.Name

				if len(pa.Backend.Service.Port.Name) == 0 {
					portString = strconv.Itoa(int(pa.Backend.Service.Port.Number))
				}

                // 生成service标识
				serviceName := provider.Normalize(ingress.Namespace + "-" + pa.Backend.Service.Name + "-" + portString)
                // 记录后端池
				conf.HTTP.Services[serviceName] = service

                // 生成route
				rt := loadRouter(rule, pa, rtConfig, serviceName)
				...
				routerKey := strings.TrimPrefix(provider.Normalize(ingress.Namespace+"-"+ingress.Name+"-"+rule.Host+pa.Path), "-")

                // 记录route
				routers[routerKey] = append(routers[routerKey], rt)
			}
		}

        // route记录到conf
		for routerKey, conflictingRouters := range routers {
            // 单负载
			if len(conflictingRouters) == 1 {
				conf.HTTP.Routers[routerKey] = conflictingRouters[0]
				continue
			}
			...

            // 多负载
			for _, router := range conflictingRouters {
				key, err := makeRouterKeyWithHash(routerKey, router.Rule)
				...
				conf.HTTP.Routers[key] = router
			}
		}
	}

    // 更新tls证书
	certs := getTLSConfig(certConfigs)
	if len(certs) > 0 {
		conf.TLS = &dynamic.TLSConfiguration{
			Certificates: certs,
		}
	}

	return conf
}

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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

注意

1.ingress模式traefik作为ingress controller，会更新ingress对象暴露地址

2.provider根据ingress及关联规则渲染router和service

3.支持ingress annotation声明middleware和tls，也支持关联的service annotation声明连接参数及nativeLB

# 2.3.file实现

file provider实现相对简单，主要监听配置文件变更及通知给watcher，watcher向外推送配置以传递给listener更新规则。

// Provide allows the file provider to provide configurations to traefik
// using the given configuration channel.
func (p *Provider) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
	...

	if p.Watch {
		...
        // 指定位置下所有文件
        watchItems = append(watchItems, ...)
		...
        // 监听
		p.addWatcher(pool, watchItems, configurationChan, p.applyConfiguration)
        ...
	}

	pool.GoCtx(func(ctx context.Context) {
		...
		signal.Notify(signals, syscall.SIGHUP)

		for {
			select {
			case <-ctx.Done():
				return
			// 退出信号
			case <-signals:
                // 加载文件配置及推送
				p.applyConfiguration(configurationChan)
                ...
			}
		}
	})

    // 加载文件配置及推送
	p.applyConfiguration(configurationChan)
    ...

	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

addWatcher()负责监听配置文件变化，发现变化后读取配置内容，将最新的配置内容发送到外部管道，以供listener消费及更新规则。

func (p *Provider) addWatcher(pool *safe.Pool, items []string, configurationChan chan<- dynamic.Message, callback func(chan<- dynamic.Message) error) error {
	watcher, err := fsnotify.NewWatcher()
	...

    // 注册watcher
	for _, item := range items {
		watcher.Add(item)
        ...
	}

	// 异步监听
	pool.GoCtx(func(ctx context.Context) {

		defer watcher.Close()
		for {
			select {
			case <-ctx.Done():
				return
            // 监听到配置变化
			case evt := <-watcher.Events:
				if p.Directory == "" {
					_, evtFileName := filepath.Split(evt.Name)
					_, confFileName := filepath.Split(p.Filename)
					if evtFileName == confFileName {
                        // 回调读取配置内容
						callback(configurationChan)
						...
					}
				} else {
                    // 回调读取配置内容
					callback(configurationChan)
					...
				}
			...
			}
		}
	})
	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

注意

核心还是inotity文件变化，将变化作为信号触发配置重新加载及推送

# 3.配置更新

# 3.1.dialListener

dialListener会消费watcher推送配置，根据最新推送配置动态更新http和tcp连接池配置，后续的流量处理均基于新的连接进行。

// Update updates http roundtrippers configurations.
func (r *RoundTripperManager) Update(newConfigs map[string]*dynamic.ServersTransport) {
	r.rtLock.Lock()
	defer r.rtLock.Unlock()

    // 已有配置
	for configName, config := range r.configs {
        newConfig, ok := newConfigs[configName]
        ...
        // 构造及更新连接(http1.1/http2)
		r.roundTrippers[configName], err = r.createRoundTripper(newConfig)
		...
	}

    // 新配置
	for newConfigName, newConfig := range newConfigs {
		...
        // 构造及更新连接(http1.1/http2)
		r.roundTrippers[newConfigName], err = r.createRoundTripper(newConfig)
		...
	}

	r.configs = newConfigs
}

// Update updates tcp dialers configurations.
func (d *DialerManager) Update(configs map[string]*dynamic.TCPServersTransport) {
	d.rtLock.Lock()
	defer d.rtLock.Unlock()
	...
	for configName, config := range configs {
        // 更新连接配置(相对固定的几个)
		d.createDialers(configName, config)
        ...
	}
}

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

注意

dialListener基于server transport更新，本质是清理旧的连接池及根据最新连接配置新建连接池

# 3.2.routeListener

routeListener会消费watcher推送配置，根据配置生成最新的handler及切换http.Handler、tcp.handler及udp.Handler。

func switchRouter(routerFactory *server.RouterFactory, serverEntryPointsTCP server.TCPEntryPoints, serverEntryPointsUDP server.UDPEntryPoints) func(conf dynamic.Configuration) {
	return func(conf dynamic.Configuration) {
        // 配置转换
		rtConf := runtime.NewConfig(conf)

        // 构造routers处理器
		routers, udpRouters := routerFactory.CreateRouters(rtConf)

        // http/tcp handler切换
		serverEntryPointsTCP.Switch(routers)
        // udp handler切换
		serverEntryPointsUDP.Switch(udpRouters)
	}
}

// CreateRouters creates new TCPRouters and UDPRouters.
func (f *RouterFactory) CreateRouters(rtConf *runtime.Configuration) (map[string]*tcprouter.Router, map[string]udp.Handler) {
    // 终止旧路由及健康检查
	if f.cancelPrevState != nil {
		f.cancelPrevState()
	}

	...
    // 新的上下文及取消回调
	ctx, f.cancelPrevState = context.WithCancel(context.Background())
	...

    // HTTP
	routerManager := router.NewManager(rtConf, serviceManager, middlewaresBuilder, f.observabilityMgr, f.tlsManager)
    // http路由器
	handlersNonTLS := routerManager.BuildHandlers(ctx, f.entryPointsTCP, false)
    // https路由器
	handlersTLS := routerManager.BuildHandlers(ctx, f.entryPointsTCP, true)
	// 健康检查
	serviceManager.LaunchHealthCheck(ctx)
	...
    
	// TCP
    // http handler会挂载到tcp router(三次握手四次挥手)
	rtTCPManager := tcprouter.NewManager(rtConf, svcTCPManager, middlewaresTCPBuilder, handlersNonTLS, handlersTLS, f.tlsManager)
    // tcp路由器
	routersTCP := rtTCPManager.BuildHandlers(ctx, f.entryPointsTCP)
	...
    
	// UDP
	rtUDPManager := udprouter.NewManager(rtConf, svcUDPManager)
    // udp路由器
	routersUDP := rtUDPManager.BuildHandlers(ctx, f.entryPointsUDP)

    // 标记配置引用关系
	rtConf.PopulateUsedBy()

	return routersTCP, routersUDP
}

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

// Switch swaps out all the given handlers in their associated entrypoints.
func (eps UDPEntryPoints) Switch(handlers map[string]udp.Handler) {
	for epName, handler := range handlers {
		if ep, ok := eps[epName]; ok {
			ep.Switch(handler)
			continue
		}
	}
}

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

注意

http server挂载到tcp router，http handler会基于http server switcher动态切换，http forwarder作为tcp router转发流量到http server的入口

# 3.3.switcher

switcher用于线程安全获取和更新handler，本质上都是加锁获取或更新switcher.safe的handler，traefik各组件有switcher实现。

// SwitchRouter switches the TCP router handler.
func (e *TCPEntryPoint) SwitchRouter(rt *tcprouter.Router) {
    // 同步http forwarder
	rt.SetHTTPForwarder(e.httpServer.Forwarder)

    // 基于switcher切换http handler
	httpHandler := rt.GetHTTPHandler()
	...
	e.httpServer.Switcher.UpdateHandler(httpHandler)

    // 同步https forwarder及注册https路由
	rt.SetHTTPSForwarder(e.httpsServer.Forwarder)

    // 基于switcher切换https handler
	httpsHandler := rt.GetHTTPSHandler()
	...
	e.httpsServer.Switcher.UpdateHandler(httpsHandler)

    // 切换tcp router
	e.switcher.Switch(rt)

    // 切换http3的handler
	if e.http3Server != nil {
		e.http3Server.Switch(rt)
	}
}

// Switch switches the UDP handler.
func (ep *UDPEntryPoint) Switch(handler udp.Handler) {
    // 切换udp handler值
	ep.switcher.Switch(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

注意

switcher.switch()其实就是切换switcher handler的值，包装一层的目的是动态切换handler及不影响旧的流量