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

# 1.简介

# 1.1.calico

calico节点级组件会打包到calico-node Pod，内部会跑三个核心进程：felix、bird和confd，集群级别支持typha组件作为API代理。
注意

calico-node启动会经历多个阶段，比较重要的是felix.run和confd.run

# 1.2.startup

startup是calico-node启动前的初始化引导模块，负责节点网络身份注册及基础配置初始化，确保节点具备calico网络能力再激活核心组件。
注意

startup模块不属于常驻服务，节点网络环境准备好就会退出，主要管理IPPool/ClusterInformation/FelixConfig三个对象

# 1.3.入口

node/calico-node目录的main函数是程序唯一入口，基于不同命令分发激活不同模块，常驻模块主要是felix和confd。

func main() {
	...

	// 1.felix模块
	if *runFelix {
		felix.Run("/etc/calico/felix.cfg", buildinfo.GitVersion, buildinfo.BuildDate, buildinfo.GitRevision)
	// 2.ebpf命令管理
    } else if *runBPF {
		// Command-line tools should log to stderr to avoid confusion with the output.
		bpf.RunBPFCmd()
    // 3.挂载BPF/CGroupV2
	} else if *runInit {
		nodeinit.Run(*bestEffort)
    // 4.环境检查
	} else if *runStartup {
		startup.Run()
    // 5.本地IP变动监控
	} else if *monitorAddrs {
		startup.ConfigureLogging()
		startup.MonitorIPAddressSubnets()
    // 6.生成BGP配置
	} else if *runConfd {
		cfg, err := confdConfig.InitConfig(true)
		...
        
		cfg.ConfDir = *confdConfDir
		cfg.KeepStageFile = *confdKeep
		cfg.Onetime = *confdRunOnce
		confd.Run(cfg)
    // 7.分配Vxlan/IP-IP隧道地址
	} else if *runAllocateTunnelAddrs {
		if *allocateTunnelAddrsRunOnce {
			allocateip.Run(nil)
		} else {
			allocateip.Run(make(chan struct{}))
		}
    // 8.初始化calico工作目录
	} else if *initHostpaths {
		hostpathinit.Run()
	}
    ...
}

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

注意

calico-node经历init-->hostpath-->startup-->confd-->felix几个阶段，其它分支主要为辅助模块

# 2.运行

# 2.1.start

startup.Run()负责节点运行前准备工作，会基于集群状态生成节点必须的网络资源及配置，相关网络配置写入calico-node及通知集群网络就绪。

// This file contains the main startup processing for the calico/node.
func Run() {
	// 1.加载nodeName
	nodeName := utils.DetermineNodeName()

	// 2.阻塞至datastore就绪
	if os.Getenv("WAIT_FOR_DATASTORE") == "true" {
		waitForConnection(ctx, cli)
	}
	...

	// 3.获取calico node对象
	node := getNode(ctx, cli, nodeName)
    ...

	// 4.集群节点名修正
	k8sNodeName := nodeName
	if nodeRef := os.Getenv("CALICO_K8S_NODE_REF"); nodeRef != "" {
		k8sNodeName = nodeRef
	}

	// 5.K8S环境
	if config, err := rest.InClusterConfig(); err == nil {
		...

		// 5.1.获取kabeadm-config
		kubeadmConfig = clientset.CoreV1().ConfigMaps(metav1.NamespaceSystem).Get(ctx, KubeadmConfigMap, ...)
		...

        // 5.2.获取集群节点
		k8sNode, err = clientset.CoreV1().Nodes().Get(ctx, k8sNodeName, metav1.GetOptions{})
		...
	}

    // 6.配置节点IP及子网(点名IP/规则过滤/veth-docker-lo-tun外的可达IP网卡)
	needsNodeUpdate := configureAndCheckIPAddressSubnets(ctx, cli, node, k8sNode)
	
    // 7.配置BGP As(ENV加载)
	needsNodeUpdate = configureASNumber(node) || needsNodeUpdate
	
    // 8.配置calico-node到k8sNode映射
	needsNodeUpdate = configureNodeRef(node) || needsNodeUpdate
    
    // 9.更新calico-node
	if needsNodeUpdate {
		CreateOrUpdate(ctx, cli, node)
        ...
	}

	// 10.检查calico工作目录
	ensureFilesystemAsExpected()

	// 11.配置IPPool
	configureIPPools(ctx, cli, kubeadmConfig)

	// 12.clusterInformation/felixConfig对象初始化
	ensureDefaultConfig(ctx, cfg, cli, node, getOSType(), kubeadmConfig, rancherState)
    ...
    
    // 13.nodeName初始化
	utils.WriteNodeConfig(nodeName)
    ...

	// 14.通知k8s网络可用
	if os.Getenv("CALICO_NETWORKING_BACKEND") != "none" {
		if clientset != nil {
			utils.SetNodeNetworkUnavailableCondition(*clientset, k8sNodeName, false, 30*time.Second)
			...
		}
	}

	// 15.清理shutdown标记
	utils.RemoveShutdownTimestampFile()
    ...
}

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

补充

大部分模块比较清晰，重点是IPPool池初始化，下面会进一步分析

# 2.2.pool

configureIPPools()用于初始化IPPool对象，确保calico网络分配阶段可以获取到可用的CIDR，优先以环境变量加载的Pool为主。

// configureIPPools ensures that default IP pools are created (unless explicitly requested otherwise).
func configureIPPools(ctx context.Context, client client.Interface, kubeadmConfig *v1.ConfigMap) {
	// 1.读取环境变量Pool
	ipv4Pool := os.Getenv("CALICO_IPV4POOL_CIDR")
	ipv6Pool := os.Getenv("CALICO_IPV6POOL_CIDR")

    // 2.禁用默认Pool
	if strings.ToLower(os.Getenv("NO_DEFAULT_POOLS")) == "true" {
		if len(ipv4Pool) > 0 || len(ipv6Pool) > 0 {
			utils.Terminate()
		}
		return
	}

	// 2.加载kubeadm记录的Pool
	if (len(ipv4Pool) == 0 || len(ipv6Pool) == 0) && kubeadmConfig != nil {
        v4, v6, err := extractKubeadmCIDRs(kubeadmConfig)
        ...
        if len(ipv4Pool) == 0 {
            ipv4Pool = v4
        }
        if len(ipv6Pool) == 0 {
            ipv6Pool = v6
        }
	}
	...
    
    // 3.加载BlockSize(26/122)
	ipv4BlockSizeEnvVar := os.Getenv("CALICO_IPV4POOL_BLOCK_SIZE")
	if ipv4BlockSizeEnvVar != "" {
		ipv4BlockSize = parseBlockSizeEnvironment(ipv4BlockSizeEnvVar)
	} else {
		ipv4BlockSize = DEFAULT_IPV4_POOL_BLOCK_SIZE
	}
	validateBlockSize(4, ipv4BlockSize)
    
	ipv6BlockSizeEnvVar := os.Getenv("CALICO_IPV6POOL_BLOCK_SIZE")
	if ipv6BlockSizeEnvVar != "" {
		ipv6BlockSize = parseBlockSizeEnvironment(ipv6BlockSizeEnvVar)
	} else {
		ipv6BlockSize = DEFAULT_IPV6_POOL_BLOCK_SIZE
	}
	validateBlockSize(6, ipv6BlockSize)
	...

	// 4.获取IPPool列表
	poolList, err := client.IPPools().List(ctx, options.ListOptions{})
    if checkIPV4PoolAndIPV6PoolPresent(poolList) {
        return
    }
	...

    // 5.未设置IPV4Pool,加载默认池(192.168.0.0/16)
	if ipv4Pool == "" {
		ipv4Pool = DEFAULT_IPV4_POOL_CIDR
		_, preferedNet, _ := net.ParseCIDR(DEFAULT_IPV4_POOL_CIDR)
        // 确认默认池可用,未冲突主机网络地址
		if selectedPool, err := ipv4.GetDefaultIPv4Pool(preferedNet); err == nil {
			ipv4Pool = selectedPool.String()
		}
	}
	_, ipv4Cidr, err := cnet.ParseCIDR(ipv4Pool)
	...

    // 6.未设置IPV6Pool,加载默认池(随机)
	if ipv6Pool == "" {
		ipv6Pool, err = GenerateIPv6ULAPrefix()
		...
	}
	_, ipv6Cidr, err := cnet.ParseCIDR(ipv6Pool)
	...

	// 7.没有IPV4Pool,进行创建
	if !ipv4Present {
		...
		createIPPool(ctx, client, ipv4Cidr, DEFAULT_IPV4_POOL_NAME, ipv4IpipModeEnvVar, ipv4VXLANModeEnvVar, outgoingNATEnabled, ipv4BlockSize, ipv4NodeSelector, bgpExportDisabled)
	}
    
    // 8.没有IPV6Pool,进行创建
	if !ipv6Present && ipv6Supported() {
		createIPPool(ctx, client, ipv6Cidr, DEFAULT_IPV6_POOL_NAME, string(api.IPIPModeNever), ipv6VXLANModeEnvVar, outgoingNATEnabled, ipv6BlockSize, ipv6NodeSelector, bgpExportDisabled)
	}
}

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

注意

对应协议IPPool已经存在不会创建默认池，因此一般无需关闭默认池初始化逻辑

# 2.3.create

createIPPool()基于加载的网段配置创建默认的IPPool对象，作为IP划分的节点网段来源，相关的网络封装模式会设置到IPPool对象。

// createIPPool creates an IP pool using the specified CIDR.  This
// method is a no-op if the pool already exists.
func createIPPool(ctx Context, client client, cidr *cnet.IPNet, poolName, ipipModeName, vxlanModeName string..){
	...

	// 1.IPIP模式加载
	switch strings.ToLower(ipipModeName) {
	case "", "off", "never":
		ipipMode = api.IPIPModeNever
	case "crosssubnet", "cross-subnet":
		ipipMode = api.IPIPModeCrossSubnet
	case "always":
		ipipMode = api.IPIPModeAlways
	...
	}

	// 2.VALan模式加载
	switch strings.ToLower(vxlanModeName) {
	case "", "off", "never":
		vxlanMode = api.VXLANModeNever
	case "crosssubnet", "cross-subnet":
		vxlanMode = api.VXLANModeCrossSubnet
	case "always":
		vxlanMode = api.VXLANModeAlways
	...
	}

	pool := &api.IPPool{
		ObjectMeta: metav1.ObjectMeta{
			Name: poolName,
		},
		Spec: api.IPPoolSpec{
			CIDR:             cidr.String(),
			NATOutgoing:      isNATOutgoingEnabled,
			IPIPMode:         ipipMode,
			VXLANMode:        vxlanMode,
			BlockSize:        blockSize,
			NodeSelector:     nodeSelector,
			DisableBGPExport: bgpExportDisabled,
		},
	}

	// 3.创建IPPool
	client.IPPools().Create(ctx, pool, options.SetOptions{})
    ...
}

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

注意

IPPool会作为后续calico-ipam的网段和IP划分的合法地址来源