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

# 1.简介

# 1.1.优势

typha是datastore-->felix的中间缓存，用于提高数据存储的性能和可扩展性，间接减少大规模集群felix实例对数据存储的负载。

--- 优点
	1.可扩展性,减少felix直连datastore数量,typha实例可以为数百个felix实例提供服务
	2.降低负载,felix未直接连接datastore,最小化数据存储的watch和read请求
	3.弹性,typha基于内存缓存向连接的felix实例提供一定程序的弹性,应对临时的数据存储问题
	4.数据验证,typha将数据分发给felix实例之前进行必要的数据验证
	5.高效更新,typha维护数据存储的快照,向felix发送增量更新,减少传输的数据量

1
2
3
4
5
6

注意

typha监听datastore资源变更，向注册连接的felix实例推送网络策略、端点及其它配置更新

# 1.2.架构

daemon是typha进程的入口，用于初始化及管理其它typha组件的生命周期，基于加载的配置设置连接、缓存、服务发现及syncer相关参数。

# 2.入口

# 2.1.entry

daemon.main()是typha启动入口，会实例化typha server对象，基于typha.InitializeAndServeForever()初始化及启动各模块。

// main is the entry point to the calico-typha binary.
func main() {
    // 实例化typha server
	typha := daemon.New()
    // 初始化及激活typha server
	typha.InitializeAndServeForever(context.Background())
}

func New() *TyphaDaemon {
	return &TyphaDaemon{
        // etcd/kube client
		NewClientV3: func(config apiconfig.CalicoAPIConfig) (DatastoreClient, error) {
			client := clientv3.New(config)
			...
			return ClientV3Shim{client.(RealClientV3), config}, nil
		},
		...
		CachesBySyncerType:    map[syncproto.SyncerType]syncserver.BreadcrumbProvider{}, // syncer type-->cache
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

注意

这里的typha本质是server端，前面confd/felix提到的typha是client端

# 2.2.initialize

t.InitializeAndServeForever()作为入口会加载及合并typha配置，初始化datastore连接，实例化typha相关对象及启动监听处理。

func (t *TyphaDaemon) InitializeAndServeForever(cxt context.Context) error {
	...
    // 加载配置
	t.LoadConfiguration(cxt)
	...
    
    // 实例化typha server
	t.CreateServer()
    // 启动server及syncer
	t.Start(cxt)
    
	t.WaitAndShutDown(cxt)
	return nil
}

// uses the command-line configuration and environment variables to load our configuration.
// It initializes the datastore connection.
func (t *TyphaDaemon) LoadConfiguration(ctx context.Context) error {
	...
    
configRetry:
	for {
		...
        
		// environment variables.
		configParams = config.New()
		envConfig := config.LoadConfigFromEnvironment(os.Environ())
        
		// config file.
		fileConfig := config.LoadConfigFile(t.ConfigFilePath)
		...
        
		// parse and merge the local config.
		configParams.UpdateFrom(envConfig, config.EnvironmentVariable)
		...
		configParams.UpdateFrom(fileConfig, config.ConfigFile)
		...

        // validate the config params(etcd地址合法/证书路径存在/datastoreType合法...)
		configParams.Validate()
		...

		// connect to the datastore.
		datastoreConfig = configParams.DatastoreConfig()
		t.DatastoreClient = t.NewClientV3(datastoreConfig)
		...
        
		break configRetry
	}

    // kube client
	if datastoreConfig.Spec.DatastoreType == apiconfig.Kubernetes {
		...
        
		for {
			...
            // 初始化v1client
			civ1 = clients.LoadKDDClientV1FromAPIConfigV3(&datastoreConfig)
			...
			break
		}

		// migration helper to determine if need to perform a migration, and if so perform the migration.
		mh := migrator.New(t.DatastoreClient, civ1, nil)
		for {
			...
            // 迁移检查
			migrate, err := mh.ShouldMigrate()
            ...
            
            // 迁移
            if migrate {
                // 旧版数据-->新版数据
				if _, err := mh.Migrate(); err != nil {
					time.Sleep(1 * time.Second)
					continue
				}
				break
			}
			break
		}
	}

	// Ensure that, as soon as we are able to connect to the datastore at all, it is initialized.
	// Note: we block further start-up while we do this, which means, if we're stuck here for long enough,
	// the liveness healthcheck will time out and start to fail.  That's fairly reasonable, being stuck here
	// likely means we have some persistent datastore connection issue and restarting Typha may solve that.
	for {
		...
        
		func() {
			...
			t.DatastoreClient.EnsureInitialized(ctx, "", "typha")
		}()
		...

		break
	}
    
	t.ConfigParams = configParams
	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
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

注意

migrate.Migrate()会获取旧资源，经格式转换为新资源后重新Patch到集群供后续基于v3client查询使用

# 2.3.information

c.EnsureInitialized()主要执行c.ensureClusterInformation()，用于初始化或更新clusterInformation对象，确保字段最新状态。

// EnsureInitialized is used to ensure the backend datastore is correctly initialized for use by Calico.
func (c client) EnsureInitialized(ctx context.Context, calicoVersion, clusterType string) error {
	// etcd/kube client实现为空
	c.backend.EnsureInitialized()
    ...

    // clusterInformation/default创建或更新
	c.ensureClusterInformation(ctx, calicoVersion, clusterType)
    ...

	return nil
}

// ensures the fields i.e. ClusterType,CalicoVersion and ClusterGUID are set. It creates/updates the as needed.
func (c client) ensureClusterInformation(ctx context.Context, calicoVersion, clusterType string) error {
	// kube datastore,clusterType补充kdd迁移标记
	if c.config.Spec.DatastoreType == apiconfig.Kubernetes {
		// If clusterType is already set then append ",kdd" at the end.
		if clusterType != "" {
			// Trim the trailing ",", if any.
			clusterType = strings.TrimSuffix(clusterType, ",")
			// Append "kdd" very last thing in the list.
			clusterType = fmt.Sprintf("%s,%s", clusterType, "kdd")
		} else {
			clusterType = "kdd"
		}
	}

	for {
		...

        // 查询clusterInformation
		clusterInfo, err := c.ClusterInformation().Get(ctx, globalClusterInfoName, options.GetOptions{})
		if err != nil {
			// create the default config if it doesn't already exist.
			if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
				newClusterInfo := v3.NewClusterInformation()
				newClusterInfo.Name = globalClusterInfoName
				newClusterInfo.Spec.CalicoVersion = calicoVersion
				newClusterInfo.Spec.ClusterType = clusterType
				u := uuid.New()
				newClusterInfo.Spec.ClusterGUID = hex.EncodeToString(u[:])
				datastoreReady := true
				newClusterInfo.Spec.DatastoreReady = &datastoreReady
				c.ClusterInformation().Create(ctx, newClusterInfo, options.SetOptions{})
				...
			}
            
			break
		}

		updateNeeded := false
        // calico version更新
		if calicoVersion != "" {
			// Only update the version if it's different from what we have.
			if clusterInfo.Spec.CalicoVersion != calicoVersion {
				clusterInfo.Spec.CalicoVersion = calicoVersion
				updateNeeded = true
			}
		}

        // clusterGUID更新
		if clusterInfo.Spec.ClusterGUID == "" {
			u := uuid.New()
			clusterInfo.Spec.ClusterGUID = hex.EncodeToString(u[:])
			updateNeeded = true
		}

        // datastoreReady更新
		if clusterInfo.Spec.DatastoreReady == nil {
			// If the ready flag is nil, default it to true (but if it's explicitly false, leave it as-is).
			datastoreReady := true
			clusterInfo.Spec.DatastoreReady = &datastoreReady
			updateNeeded = true
		}

        // clusterType更新
		if clusterType != "" {
			if clusterInfo.Spec.ClusterType == "" {
				clusterInfo.Spec.ClusterType = clusterType
				updateNeeded = true
			} else {
				allClusterTypes := strings.Split(clusterInfo.Spec.ClusterType, ",")
				existingClusterTypes := set.FromArray(allClusterTypes)
				localClusterTypes := strings.Split(clusterType, ",")

				clusterTypeUpdateNeeded := false
				for _, lct := range localClusterTypes {
					if existingClusterTypes.Contains(lct) {
						continue
					}
					clusterTypeUpdateNeeded = true
					allClusterTypes = append(allClusterTypes, lct)
				}

				if clusterTypeUpdateNeeded {
					clusterInfo.Spec.ClusterType = strings.Join(allClusterTypes, ",")
					updateNeeded = true
				}
			}
		}

        // 更新clusterInformation
		if updateNeeded {
			c.ClusterInformation().Update(ctx, clusterInfo, options.SetOptions{})
			...
		}
		break
	}

	return nil
}

注意

这里比较明确，基于一些传入或默认配置更新clusterInformation对象，clusterInformation是startup阶段初始化的

# 2.4.server

t.CreateServer()会初始化typha server，构建typha的数据同步链路，同步的数据会fan out到typha server注册的连接。

// CreateServer creates and configures (but does not start) the server components.
func (t *TyphaDaemon) CreateServer() {
	...

	// create the Syncer and caching layer(one pipeline for each syncer we support).
	t.addSyncerPipeline(syncproto.SyncerTypeFelix, t.DatastoreClient.FelixSyncerByIface)
	t.addSyncerPipeline(syncproto.SyncerTypeBGP, t.DatastoreClient.BGPSyncerByIface)
	t.addSyncerPipeline(syncproto.SyncerTypeTunnelIPAllocation, t.DatastoreClient.TunnelIPAllocationSyncerByIface)
	t.addSyncerPipeline(syncproto.SyncerTypeNodeStatus, t.DatastoreClient.NodeStatusSyncerByIface)

	// Create the server, which listens for connections from Felix.
	t.Server = syncserver.New(
		t.CachesBySyncerType,
		syncserver.Config{
			MaxMessageSize:                 t.ConfigParams.ServerMaxMessageSize,
			MinBatchingAgeThreshold:        t.ConfigParams.ServerMinBatchingAgeThresholdSecs,
			MaxFallBehind:                  t.ConfigParams.ServerMaxFallBehindSecs,
			NewClientFallBehindGracePeriod: t.ConfigParams.ServerNewClientFallBehindGracePeriod,
			PingInterval:                   t.ConfigParams.ServerPingIntervalSecs,
			PongTimeout:                    t.ConfigParams.ServerPongTimeoutSecs,
			HandshakeTimeout:               t.ConfigParams.ServerHandshakeTimeoutSecs,
			DropInterval:                   t.ConfigParams.ConnectionDropIntervalSecs,
			ShutdownTimeout:                t.ConfigParams.ShutdownTimeoutSecs,
			ShutdownMaxDropInterval:        t.ConfigParams.ShutdownConnectionDropIntervalMaxSecs,
			MaxConns:                       t.ConfigParams.MaxConnectionsUpperLimit,
			Port:                           t.ConfigParams.ServerPort,
			HealthAggregator:               t.healthAggregator,
			KeyFile:                        t.ConfigParams.ServerKeyFile,
			CertFile:                       t.ConfigParams.ServerCertFile,
			CAFile:                         t.ConfigParams.CAFile,
			ClientCN:                       t.ConfigParams.ClientCN,
			ClientURISAN:                   t.ConfigParams.ClientURISAN,
		},
	)
}

func (t *TyphaDaemon) addSyncerPipeline(syncerType SyncerType,newSyncer func(callbacks SyncerCallbacks) Syncer){
    // syncer初始化(CR监听)
	syncerToValidator := calc.NewSyncerCallbacksDecoupler()
	syncer := newSyncer(syncerToValidator)

    // validator-->cache中间缓存
	toCache := calc.NewSyncerCallbacksDecoupler()
	...
    
    // validator初始化
	if syncerType == syncproto.SyncerTypeFelix {
        // felix type with nodeCounter wrapper
		t.nodeCounter = calc.NewNodeCounter(toCache)
		validator = calc.NewValidationFilter(t.nodeCounter)
	} else {
		// Otherwise, just go from validator to cache directly.
		validator = calc.NewValidationFilter(toCache)
	}

	// snapshot cache, which stores point-in-time copies of the datastore contents.
	cache := snapcache.New(snapcache.Config{
		MaxBatchSize:     t.ConfigParams.SnapshotCacheMaxBatchSize,
		...
		Name:             string(syncerType),
	})

    // 实例化syncer pipeline
	pipeline := &syncerPipeline{
		Type:              syncerType,
		Syncer:            syncer,
		SyncerToValidator: syncerToValidator,
		Validator:         validator,
		ValidatorToCache:  toCache,
		Cache:             cache,
	}
    
    // 注册pipeline及cache
	t.SyncerPipelines = append(t.SyncerPipelines, pipeline)
	t.CachesBySyncerType[syncerType] = cache
}

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

注意

typha底层还是依赖syncer同步datastore变更，经syncer-->validator-->cache记录资源快照，利用server实现fan out

# 2.5.start

t.Start()负责启动相关的syncer pipeline及server，kubernetes环境会持续统计typha实例、kubeNode数量及syncer类型数量。

// Start starts all the server components in background goroutines.
func (t *TyphaDaemon) Start(cxt context.Context) {
	// connected everything up, start the background processing threads.
	for _, s := range t.SyncerPipelines {
		s.Start(cxt)
	}
    
    // start server
	t.Server.Start(cxt)
    
    // kubernetes环境
	if t.ConfigParams.ConnectionRebalancingMode == "kubernetes" {
		k8sAPI := k8s.NewK8sAPI(t.nodeCounter)
        // [3s,30s]定时器
		ticker := jitter.NewTicker(t.ConfigParams.K8sServicePollIntervalSecs, t.ConfigParams.K8sServicePollIntervalSecs/10)
        // 周期统计集群数据
		go k8s.PollK8sForConnectionLimit(cxt,t.ConfigParams,ticker.C, k8sAPI,t.Server,len(t.CachesBySyncerType))
	}
	...
}

func PollK8sForConnectionLimit(cxt context.Context, configParams *config.Config, tickerC <-chan time.Time,
	k8sAPI K8sAPI, server MaxConnsAPI, numSyncerTypes int) {
	activeTarget := configParams.MaxConnectionsUpperLimit
	for {
		select {
        // 触发定时器
		case <-tickerC:
            ...
            
			// Get the number of Typhas in the service.
			numTyphas := k8sAPI.GetNumTyphas(reqCtx, configParams.K8sNamespace, configParams.K8sServiceName, configParams.K8sPortName)
			...
            
			// Get the number of nodes.  We expect one syncer connection of each type per node.
			numNodes, nErr := k8sAPI.GetNumNodes()
			...

			target := configParams.MaxConnectionsUpperLimit
			reason := "error"
			if tErr == nil && nErr == nil {
                // typha承载连接数计算
				target, reason = CalculateMaxConnLimit(configParams, numTyphas, numNodes, numSyncerTypes)
			}

			if target != activeTarget {
				server.SetMaxConns(target)
				activeTarget = target
			}
		case <-cxt.Done():
			return
		}
	}
}

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

连接数
$Candidate_{[MaxConnLowerLimit,\;MaxConnUpperLimit]} = NumSyncerTypes \times \left( 1 + \frac{1.2 \times NumNodes}{NumTyphas - 1} \right)$

# 3.cache

# 3.1.start

p.Start()本质是事件循环处理模块，将validator msg批聚合到cache，基于breadcrumb版本链发布不可变快照供下游server发布。

func (p syncerPipeline) Start(cxt context.Context) {
    // confd-load模块介绍过,基于watchCache监听,基于watchSyncer批量分发
	p.Syncer.Start()

    // syncer-->validator
	go p.SyncerToValidator.SendTo(p.Validator)

    // validator-->cache
	go p.ValidatorToCache.SendTo(p.Cache)

    // handle msg to cache
	p.Cache.Start(cxt)
}

// Start starts the cache's main loop in a background goroutine.
func (c *Cache) Start(ctx context.Context) {
	go c.loop(ctx)
}

func (c *Cache) loop(ctx context.Context) {
	for {
		// 批量消费validator msg,缓存至pending区
		c.fillBatchFromInputQueue(ctx)
        ...
        
		// Then publish the updates in new Breadcrumb(s).
		c.publishBreadcrumbs()
	}
}

// fillBatchFromInputQueue waits for some input on the input channel, then opportunistically
// pulls as much as possible from the channel.  Input is stored in the pendingXXX fields for
// the next stage of processing.
func (c *Cache) fillBatchFromInputQueue(ctx context.Context) error {
	somethingToSend := false
	batchSize := 0
	storePendingUpdate := func(obj interface{}) {
		somethingToSend = true
		switch obj := obj.(type) {
        // nodeStatus event
		case api.SyncStatus:
			c.pendingStatus = obj
			...
       	// resource event
		case []api.Update:
            // 暂存pending区
			c.pendingUpdates = append(c.pendingUpdates, obj...)
            ...
		default:
			log.WithField("obj", obj).Panic("Unexpected object")
		}
	}

	for ctx.Err() == nil && !somethingToSend {
		select {
        // validator msg
		case obj := <-c.inputC:
            // 暂存pending区
			storePendingUpdate(obj)
		batchLoop:
            // 未超出上限,继续取N条
			for batchSize < c.config.MaxBatchSize {
				select {
                // 暂存pending区
				case obj = <-c.inputC:
					storePendingUpdate(obj)
				...
				default:
					break batchLoop
				}
			}
		...
        // [1s,10s]触发一次,避免G阻塞卡住正常退出
		case <-c.wakeUpTicker.C:
			// 非广播窗口
			if time.Since(c.lastBroadcast) < c.config.WakeUpInterval {
				// Already did a broadcast recently due to minting a real breadcrumb.
				continue
			}
            
            // 广播唤醒wait协程
			c.breadcrumbCond.Broadcast()
			c.lastBroadcast = time.Now()
		...
		}
	}
    
	return ctx.Err()
}

注意

wakeUpTicker周期执行c.breadcrumbCond.Broadcast()用于唤醒阻塞的server协程，避免阻塞服务正常退出

# 3.2.publish

c.publishBreadcrumbs()批量消费pending updates，将KV变更应用到KV树及breadcrumb版本链，实现高效的MVCC状态发布及通知。

// publishBreadcrumbs sends a series of Breadcrumbs, draining the pending updates list.
func (c *Cache) publishBreadcrumbs() {
	// Always force one call in case we need to send a breadcrumb just to update the sync status.
	c.publishBreadcrumb()

	// Then send anything else we have left.
	for len(c.pendingUpdates) > 0 {
		c.publishBreadcrumb()
	}
}

// updates the master tree and publishes a new Breadcrumb containing a read-only snapshot of the tree and the 
// deltas from this batch.
func (c *Cache) publishBreadcrumb() {
	...
    
    // 多批截取分发
	if len(c.pendingUpdates) > c.config.MaxBatchSize {
		updates = c.pendingUpdates[:c.config.MaxBatchSize]
		c.pendingUpdates = c.pendingUpdates[c.config.MaxBatchSize:]
	} else {
		updates = c.pendingUpdates
		c.pendingUpdates = c.pendingUpdates[:0]
		lastUpdate = true
	}

	// 生成新的版本
	oldCrumb := c.CurrentBreadcrumb()
	newCrumb := &Breadcrumb{
		SequenceNumber: oldCrumb.SequenceNumber + 1,
		Timestamp:      time.Now(),
		SyncStatus:     oldCrumb.SyncStatus,
		nextCond:       c.breadcrumbCond,
		Deltas:         make([]syncproto.SerializedUpdate, 0, len(updates)),
		counterBreadcrumbBlock:    c.counterBreadcrumbBlock,
		counterBreadcrumbNonBlock: c.counterBreadcrumbNonBlock,
	}
    
    // 状态更新延迟到lastBatch
	if lastUpdate && c.pendingStatus != newCrumb.SyncStatus {
		// Only update the status if this is the last message in the batch, otherwise
		// we might tell the client that we're in sync too soon.
		somethingChanged = true
		newCrumb.SyncStatus = c.pendingStatus
	}
    
	// Update the main trie and record the updates in the new crumb.
	for _, upd := range updates {
		...
        
		// Pre-serialise the KV so that we only serialise once per update instead of once for each client.
		newUpd, err := syncproto.SerializeUpdate(upd)
		...
        
		// Update the master KV map.
		oldUpd, exists := c.kvs.Get(newUpd)
        
        // del event
		if upd.Value == nil {
			c.kvs.Delete(newUpd)
        // add/update event
		} else {
            // 无变化
			if exists && newUpd.WouldBeNoOp(oldUpd) {
				c.counterUpdatesSkipped.Inc()
				continue
			}
			
            // 更新kvs
			updToStore := newUpd
			updToStore.UpdateType = api.UpdateTypeKVNew
			c.kvs.ReplaceOrInsert(updToStore)
		}

		// Record the update in the new Breadcrumb so that clients following the chain of
		// Breadcrumbs can apply it as a delta.
		newCrumb.Deltas = append(newCrumb.Deltas, newUpd)
		somethingChanged = true
	}

	if !somethingChanged {
		return
	}
	...
	
	// Add the new read-only snapshot to the new crumb.
	newCrumb.KVs = c.kvs.Clone()

	// Replace the Breadcrumb and link the old Breadcrumb to the new so that clients can follow the trail.
	c.breadcrumbCond.L.Lock()
    // newCrumb接到链表尾部
	atomic.StorePointer(&(oldCrumb.next), (unsafe.Pointer)(newCrumb))
    // 当前crumb指向最新,oldCrumb后期会GC掉
	atomic.StorePointer(&c.currentBreadcrumb, (unsafe.Pointer)(newCrumb))
	c.breadcrumbCond.L.Unlock()
	
    // 唤醒阻塞协程
	c.breadcrumbCond.Broadcast()
	c.lastBroadcast = time.Now()
	...
}

注意

newCrumb作为最新版本稳定数据会被typha server消费，广播给felix