【kubernetes/k8s源码分析】calico node源码分析

末蓝、 2021-11-16 13:24 261阅读 0赞

## calico node 工作内容 ##

> ### 镜像命令 start\_runit    \#!/bin/sh   \# From https://github.com/faisyl/alpine-runit   env > /etc/envvars ###
> 
> /etc/rc.local  
> retval=$?  
> if \[ $retval -ne 0 \];  
> then  
>     echo >&2 "Calico node failed to start"  
>     exit $retval  
> fi
> 
> \# Export the nodename set by the startup procedure.   
> export NODENAME=$(cat /var/lib/calico/nodename)
> 
> RUNSVDIR=$(/usr/bin/which runsvdir)  
> exec $\{RUNSVDIR\} -P /etc/service/enabled

## 命令行参数 ##

> Usage of Calico:  
>   -allocate-tunnel-addrs  
>         Configure tunnel addresses for this node  
>   -bird-ready  
>         Run BIRD readiness checks  
>   -bird6-ready  
>         Run BIRD6 readiness checks  
>   -confd  
>         Run confd  
>   -confd-keep-stage-file  
>         Keep stage file when running confd  
>   -confd-run-once  
>         Run confd in oneshot mode  
>   -felix  
>         Run Felix  
>   -felix-ready  
>         Run felix readiness checks  
>   -startup  
>         Initialize a new node  
>   -threshold-time duration  
>         Threshold time for bird readiness (default 30s)  
>   -v    Display version  
> flag: help requested

# 1. main 函数 #

###     1.1 -v -felix -startup -confd 这几个参数不能同时出现 ###

// Perform some validation on the parsed flags. Only one of the following may be
    // specified at a time.
    onlyOne := []*bool{version, runFelix, runStartup, runConfd}
    oneSelected := false
    for _, o := range onlyOne {
    	if oneSelected && *o {
    		fmt.Println("More than one incompatible argument provided")
    		os.Exit(1)
    	}
    
    	if *o {
    		oneSelected = true
    	}
    }

进程启动使用 runsv 方式，包括目录 /etc/service/enabled/ 下，runsv 参考后文 runsv 描述

![20190728154550108.png][]

包括的进程有  felix bird bird5 confd 这四个分别如下讲述

###   1.2 calico-node -felix ###

配置文件 /etc/calico/felix.cfg

> \[global\]  
> MetadataAddr = None  
> LogFilePath = None  
> LogSeverityFile = None  
> LogSeveritySys = None

后文将分析 felix，内容比较多

// Decide which action to take based on the given flags.
    if *version {
    	fmt.Println(startup.VERSION)
    	os.Exit(0)
    } else if *runFelix {
    	felix.Run("/etc/calico/felix.cfg")

###     1.3 runsv confd ###

参看文章 [https://blog.csdn.net/zhonglinzhang/article/details/97614844][https_blog.csdn.net_zhonglinzhang_article_details_97614844]

else if *runConfd {
    	cfg, err := confdConfig.InitConfig(true)
    	cfg.ConfDir = "/etc/calico/confd"
    	cfg.KeepStageFile = *confdKeep
    	cfg.Onetime = *confdRunOnce
    	if err != nil {
    		panic(err)
    	}
    	confd.Run(cfg)

###     1.4  calico-node -startup ###

else if *runStartup {
    	startup.Run()
    }

**    1.4.1 Run 函数**

configureIPsAndSubnets

--> autoDetectCIDR

**                   -->  **autoDetectCIDRFirstFound

路径 pkg/startup/startup.go 主要执行体

// This file contains the main startup processing for the calico/node.  This
    // includes:
    // -  Detecting IP address and Network to use for BGP
    // -  Configuring the node resource with IP/AS information provided in the
    //    environment, or autodetected.
    // -  Creating default IP Pools for quick-start use
    func Run() {
    	// Check $CALICO_STARTUP_LOGLEVEL to capture early log statements
    	configureLogging()
    
    	// Determine the name for this node.
    	nodeName := determineNodeName()
    
    	// Create the Calico API cli.
    	cfg, cli := calicoclient.CreateClient()
    
    	ctx := context.Background()

**    1.4.2 waitForConnnection 检查 client 是否连接成功**

// waitForConnection waits for the datastore to become accessible.
    func waitForConnection(ctx context.Context, c client.Interface) {
    	log.Info("Checking datastore connection")
    	for {
    		// Query some arbitrary configuration to see if the connection
    		// is working.  Getting a specific Node is a good option, even
    		// if the Node does not exist.
    		_, err := c.Nodes().Get(ctx, "foo", options.GetOptions{})
    
    		// We only care about a couple of error cases, all others would
    		// suggest the datastore is accessible.
    		if err != nil {
    			switch err.(type) {
    			case cerrors.ErrorConnectionUnauthorized:
    				log.Warn("Connection to the datastore is unauthorized")
    				terminate()
    			case cerrors.ErrorDatastoreError:
    				log.WithError(err).Info("Hit error connecting to datastore - retry")
    				time.Sleep(1000 * time.Millisecond)
    				continue
    			}
    		}
    
    		// We've connected to the datastore - break out of the loop.
    		break
    	}
    	log.Info("Datastore connection verified")
    }

**    1.4.3 configureIPsAndSubnets**

根据环境变量 IP=autodetect，未设置 IP\_AUTODETECTION\_METHOD，以及 IPV4 版本，autodetect 发现第一块网络接口

// Determine the autodetection type for IPv4 and IPv6.  Note that we
    	// only autodetect IPv4 when it has not been specified.  IPv6 must be
    	// explicitly requested using the "autodetect" value.
    	//
    	// If we aren't auto-detecting then we need to validate the configured
    	// value and possibly fix up missing subnet configuration.
    	ipv4Env := os.Getenv("IP")
    	if ipv4Env == "autodetect" || (ipv4Env == "" && node.Spec.BGP.IPv4Address == "") {
    		adm := os.Getenv("IP_AUTODETECTION_METHOD")
    		cidr := autoDetectCIDR(adm, 4)
    		if cidr != nil {
    			// We autodetected an IPv4 address so update the value in the node.
    			node.Spec.BGP.IPv4Address = cidr.String()
    		} else if node.Spec.BGP.IPv4Address == "" {
    			// No IPv4 address is configured, but we always require one, so exit.
    			log.Warn("Couldn't autodetect an IPv4 address. If auto-detecting, choose a different autodetection method. Otherwise provide an explicit address.")
    			return false, fmt.Errorf("Failed to autodetect an IPv4 address")
    		} else {
    			// No IPv4 autodetected, but a previous one was configured.
    			// Tell the user we are leaving the value unchanged.  We
    			// will validate that the IP matches one on the interface.
    			log.Warnf("Autodetection of IPv4 address failed, keeping existing value: %s", node.Spec.BGP.IPv4Address)
    			validateIP(node.Spec.BGP.IPv4Address)
    		}
    	} else if ipv4Env != "none" {
    		if ipv4Env != "" {
    			node.Spec.BGP.IPv4Address = parseIPEnvironment("IP", ipv4Env, 4)
    		}
    		validateIP(node.Spec.BGP.IPv4Address)
    	}

**    1.4.4 configureASNumner 函数**

未设置则使用 global

// configureASNumber configures the Node resource with the AS number specified
    // in the environment, or is a no-op if not specified.
    func configureASNumber(node *api.Node) {
    	// Extract the AS number from the environment
    	asStr := os.Getenv("AS")
    	if asStr != "" {
    		if asNum, err := numorstring.ASNumberFromString(asStr); err != nil {
    			log.WithError(err).Errorf("The AS number specified in the environment (AS=%s) is not valid", asStr)
    			terminate()
    		} else {
    			log.Infof("Using AS number specified in environment (AS=%s)", asNum)
    			node.Spec.BGP.ASNumber = &asNum
    		}
    	} else {
    		if node.Spec.BGP.ASNumber == nil {
    			log.Info("No AS number configured on node resource, using global value")
    		} else {
    			log.Infof("Using AS number %s configured in node resource", node.Spec.BGP.ASNumber)
    		}
    	}
    }

**    1.4.5 ensureFilesystemAsExpected**

**   存在目录 /var/run/calico /var/lib/calico，如果舍子日志 /var/log/calico**

// Checks that the filesystem is as expected and fix it if possible
    func ensureFilesystemAsExpected() {
    	// BIRD requires the /var/run/calico directory in order to provide status
    	// information over the control socket, but other backends do not
    	// need this check.
    	if strings.ToLower(os.Getenv("CALICO_NETWORKING_BACKEND")) == "bird" {
    		runDir := "/var/run/calico"
    		// Check if directory already exists
    		if _, err := os.Stat(runDir); err != nil {
    
    
    	// Make sure the /var/lib/calico directory exists.
    	libDir := "/var/lib/calico"
    	// Check if directory already exists
    	if _, err := os.Stat(libDir); err != nil {
    
    
    	// Ensure the log directory exists but only if logging to file is enabled.
    	if strings.ToLower(os.Getenv("CALICO_DISABLE_FILE_LOGGING")) != "true" {
    		logDir := "/var/log/calico"
    		// Check if directory already exists
    		if _, err := os.Stat(logDir); err != nil {
    
    	}
    }

**     1.4.6 configureIPPools**

**     配置 IP 地址池，资源 ippools，如果未指定则使用 default-ipv4-ippool**

**    获取环境变量 **CALICO\_IPV4POOL\_CIDR  CALICO\_IPV6POOL\_CIDR   CALICO\_IPV4POOL\_IPIP  CALICO\_IPV4POOL\_VXLAN

> **  apiVersion: crd.projectcalico.org/v1  
> kind: IPPool  
> metadata:  
>   annotations:  
>     projectcalico.org/metadata: '\{"uid":"f86d942a-b0f5-11e9-8106-080027603363","creationTimestamp":"2019-07-28T05:10:09Z"\}'  
>   creationTimestamp: "2019-07-28T05:10:09Z"  
>   generation: 1  
>   name: default-ipv4-ippool  
>   resourceVersion: "2444410"  
>   selfLink: /apis/crd.projectcalico.org/v1/ippools/default-ipv4-ippool  
>   uid: f870f4cf-b0f5-11e9-a59b-080027603363  
> spec:  
>   blockSize: 26  
>   cidr: 192.170.0.0/16  
>   ipipMode: Never  
>   natOutgoing: true  
>   nodeSelector: all()  
>   vxlanMode: Never**

// configureIPPools ensures that default IP pools are created (unless explicitly
    // requested otherwise).
    func configureIPPools(ctx context.Context, client client.Interface) {
    	// Read in environment variables for use here and later.
    	ipv4Pool := os.Getenv("CALICO_IPV4POOL_CIDR")
    	ipv6Pool := os.Getenv("CALICO_IPV6POOL_CIDR")
    
    	if strings.ToLower(os.Getenv("NO_DEFAULT_POOLS")) == "true" {
    		if len(ipv4Pool) > 0 || len(ipv6Pool) > 0 {
    			log.Error("Invalid configuration with NO_DEFAULT_POOLS defined and CALICO_IPV4POOL_CIDR or CALICO_IPV6POOL_CIDR defined.")
    			terminate()
    		}
    
    		log.Info("Skipping IP pool configuration")
    		return
    	}

**   **

# **2. createIPPool 函数** #

**      用指定的 CIDR 创建 IP Pool**

// createIPPool creates an IP pool using the specified CIDR.  This
    // method is a no-op if the pool already exists.
    func createIPPool(ctx context.Context, client client.Interface, cidr *cnet.IPNet, poolName, ipipModeName, vxlanModeName string, isNATOutgoingEnabled bool) {
    	version := cidr.Version()
    	var ipipMode api.IPIPMode
    	var vxlanMode api.VXLANMode

###     2.1 根据环境变量设置 IPIP VXLAN 参数 ###

// Parse the given IPIP mode.
    switch strings.ToLower(ipipModeName) {
    case "", "off", "never":
    	ipipMode = api.IPIPModeNever
    case "crosssubnet", "cross-subnet":
    	ipipMode = api.IPIPModeCrossSubnet
    case "always":
    	ipipMode = api.IPIPModeAlways
    default:
    	log.Errorf("Unrecognized IPIP mode specified in CALICO_IPV4POOL_IPIP '%s'", ipipModeName)
    	terminate()
    }
    
    // Parse the given VXLAN mode.
    switch strings.ToLower(vxlanModeName) {
    case "", "off", "never":
    	vxlanMode = api.VXLANModeNever
    case "always":
    	vxlanMode = api.VXLANModeAlways
    default:
    	log.Errorf("Unrecognized VXLAN mode specified in CALICO_IPV4POOL_VXLAN'%s'", vxlanModeName)
    	terminate()
    }

###     2.2  创建 ippools 资源 ###

>       spec:  
>   blockSize: 26  
>   cidr: 192.170.0.0/16  
>   ipipMode: Never  
>   natOutgoing: true  
>   nodeSelector: all()  
>   vxlanMode: Never

pool := &api.IPPool{
    	ObjectMeta: metav1.ObjectMeta{
    		Name: poolName,
    	},
    	Spec: api.IPPoolSpec{
    		CIDR:        cidr.String(),
    		NATOutgoing: isNATOutgoingEnabled,
    		IPIPMode:    ipipMode,
    		VXLANMode:   vxlanMode,
    	},
    }
    
    log.Infof("Ensure default IPv%d pool is created. IPIP mode: %s, VXLAN mode: %s", version, ipipMode, vxlanMode)
    
    // Create the pool.  There is a small chance that another node may
    // beat us to it, so handle the fact that the pool already exists.
    if _, err := client.IPPools().Create(ctx, pool, options.SetOptions{}); err != nil

# 3. ensureDefaultConfig 函数 #

确保所有默认设置都已经配置，请看下文逐个分析，环境变量 CLUSTER\_TYPE=k8s,bgp

// ensureDefaultConfig ensures all of the required default settings are
    // configured.
    func ensureDefaultConfig(ctx context.Context, cfg *apiconfig.CalicoAPIConfig, c client.Interface, node *api.Node) error {
    	// Ensure the ClusterInformation is populated.
    	// Get the ClusterType from ENV var. This is set from the manifest.
    	clusterType := os.Getenv("CLUSTER_TYPE")
    	if err := c.EnsureInitialized(ctx, VERSION, clusterType); err != nil {
    		return nil
    	}

###       3.1 未找到默认的 default FelixConfiguration，则创建 ###

>     // FelixConfigurationInterface has methods to work with FelixConfiguration resources.
>     type FelixConfigurationInterface interface {
>        Create(ctx context.Context, res *apiv3.FelixConfiguration, opts options.SetOptions) (*apiv3.FelixConfiguration, error)
>        Update(ctx context.Context, res *apiv3.FelixConfiguration, opts options.SetOptions) (*apiv3.FelixConfiguration, error)
>        Delete(ctx context.Context, name string, opts options.DeleteOptions) (*apiv3.FelixConfiguration, error)
>        Get(ctx context.Context, name string, opts options.GetOptions) (*apiv3.FelixConfiguration, error)
>        List(ctx context.Context, opts options.ListOptions) (*apiv3.FelixConfigurationList, error)
>        Watch(ctx context.Context, opts options.ListOptions) (watch.Interface, error)
>     }

> apiVersion: crd.projectcalico.org/v1  
> kind: FelixConfiguration  
> metadata:  
>   annotations:  
>     projectcalico.org/metadata: '\{"uid":"f8814ee8-b0f5-11e9-8106-080027603363","creationTimestamp":"2019-07-28T05:10:09Z"\}'  
>   creationTimestamp: "2019-07-28T05:10:09Z"  
>   generation: 1  
>   name: default  
>   resourceVersion: "2444412"  
>   selfLink: /apis/crd.projectcalico.org/v1/felixconfigurations/default  
>   uid: f881c3f0-b0f5-11e9-a59b-080027603363  
> spec:  
>   XDPRefreshInterval: null  
>   logSeverityScreen: Info  
>   reportingInterval: 0s

// By default we set the global reporting interval to 0 - this is
    // different from the defaults defined in Felix.
    //
    // Logging to file is disabled in the felix.cfg config file.  This
    // should always be disabled for calico/node.  By default we log to
    // screen - set the default logging value that we desire.
    felixConf, err := c.FelixConfigurations().Get(ctx, globalFelixConfigName, options.GetOptions{})
    if err != nil {
    	// Create the default config if it doesn't already exist.
    	if _, ok := err.(cerrors.ErrorResourceDoesNotExist); ok {
    		newFelixConf := api.NewFelixConfiguration()
    		newFelixConf.Name = globalFelixConfigName
    		newFelixConf.Spec.ReportingInterval = &metav1.Duration{Duration: 0}
    		newFelixConf.Spec.LogSeverityScreen = defaultLogSeverity
    		_, err = c.FelixConfigurations().Create(ctx, newFelixConf, options.SetOptions{})
    		if err != nil {
    			if conflict, ok := err.(cerrors.ErrorResourceAlreadyExists); ok {
    				log.Infof("Ignoring conflict when setting value %s", conflict.Identifier)
    			} else {
    				log.WithError(err).WithField("FelixConfig", newFelixConf).Errorf("Error creating Felix global config")
    				return err
    			}
    		}
    	} else {
    		log.WithError(err).WithField("FelixConfig", globalFelixConfigName).Errorf("Error getting Felix global config")
    		return err
    	}
    }

总结：

使用 runsv 监控启动 /etc/service/default 目录下的启动 run 文件

包括  felix bird bird5 confd

### runsv 描述 ###

>      由runit包提供的一个进程监控机制，对指定的进程（一般是后台服务）进行监控，发现其挂了就会自动重启。同时它也提供了对外操作接口，可以通过管道与其通讯，默认提供了一个sv命令行程序来控制（被监控）进程的启停
> 
>      在/etc/service/下面，有一堆子目录，这里每一个目录代表了一个要监控的服务，目录里面要放一些脚本文件如run/finish/down等等，分别会在各种时机被调用
> 
>     整个机制的总控进程是runsvdir，它会读取相应目录下的各个“服务配置”，为每一个“服务”分配启动一个runsv进程去监控执行

[20190728154550108.png]: /images/20211116/67c90719511d4b94a88ef9c4d0ab4269.png
[https_blog.csdn.net_zhonglinzhang_article_details_97614844]: https://blog.csdn.net/zhonglinzhang/article/details/97614844