【kubernetes/k8s源码分析】CNI calico ipam源码分析

左手的ㄟ右手 2022-05-10 08:48 208阅读 0赞

[https://github.com/projectcalico/cni-plugin][https_github.com_projectcalico_cni-plugin]

本文基于v1.11.0，v3 主要差别是etcdv3以及设置的key路径

calico解决不同物理机上容器之间的通信，而calico-plugin是在k8s创建Pod时为Pod设置虚拟网卡(容器中的`eth0`和`lo`网卡)，calico-plugin是由两个静态的二进制文件组成，由kubelet以命令行的形式调用，这两个二进制的作用如下：

*  calico-ipam：分配维护IP，依赖etcd
 *  calico：系统调用API来修改namespace中的网卡信息

## 配置文件 ##

{
        "name": "calico-k8s-network",
        "cniVersion": "0.1.0",
        "type": "calico",
        "etcd_endpoints": "https://10.12.51.233:2379,https://10.12.51.234:2379,https://10.12.51.235:2379",
        "etcd_key_file": "/etc/calico/ssl/calico-key.pem",
        "etcd_cert_file": "/etc/calico/ssl/calico.pem",
        "etcd_ca_cert_file": "/etc/calico/ssl/ca.pem",
        "log_level": "info",
        "mtu": 1500,
        "ipam": {
            "type": "calico-ipam"
        },
        "policy": {
            "type": "k8s"
        },
        "kubernetes": {
            "kubeconfig": "/root/.kube/config"
        }
    }

## 结构体NetConf ##

NetConf结构体字段很清晰

// NetConf stores the common network config for Calico CNI plugin
    type NetConf struct {
    	CNIVersion string `json:"cniVersion,omitempty"`
    	Name       string `json:"name"`
    	Type       string `json:"type"`
    	IPAM       struct {
    		Name       string
    		Type       string   `json:"type"`
    		Subnet     string   `json:"subnet"`
    		AssignIpv4 *string  `json:"assign_ipv4"`
    		AssignIpv6 *string  `json:"assign_ipv6"`
    		IPv4Pools  []string `json:"ipv4_pools,omitempty"`
    		IPv6Pools  []string `json:"ipv6_pools,omitempty"`
    	} `json:"ipam,omitempty"`
    	MTU            int        `json:"mtu"`
    	Hostname       string     `json:"hostname"`
    	Nodename       string     `json:"nodename"`
    	DatastoreType  string     `json:"datastore_type"`
    	EtcdAuthority  string     `json:"etcd_authority"`
    	EtcdEndpoints  string     `json:"etcd_endpoints"`
    	LogLevel       string     `json:"log_level"`
    	Policy         Policy     `json:"policy"`
    	Kubernetes     Kubernetes `json:"kubernetes"`
    	Args           Args       `json:"args"`
    	EtcdScheme     string     `json:"etcd_scheme"`
    	EtcdKeyFile    string     `json:"etcd_key_file"`
    	EtcdCertFile   string     `json:"etcd_cert_file"`
    	EtcdCaCertFile string     `json:"etcd_ca_cert_file"`
    }
    	IncludeDefaultRoutes bool              `json:"include_default_routes,omitempty"`
    
    	// Options below here are deprecated.
    	EtcdAuthority string `json:"etcd_authority"`
    	Hostname      string `json:"hostname"`
    }

## 结构体K8sArgs ##

// K8sArgs is the valid CNI_ARGS used for Kubernetes
    type K8sArgs struct {
    	types.CommonArgs
    	IP                         net.IP
    	K8S_POD_NAME               types.UnmarshallableString
    	K8S_POD_NAMESPACE          types.UnmarshallableString
    	K8S_POD_INFRA_CONTAINER_ID types.UnmarshallableString
    }

# 结构体Result #

// Result is what gets returned from the plugin (via stdout) to the caller
    type Result struct {
    	CNIVersion string         `json:"cniVersion,omitempty"`
    	Interfaces []*Interface   `json:"interfaces,omitempty"`
    	IPs        []*IPConfig    `json:"ips,omitempty"`
    	Routes     []*types.Route `json:"routes,omitempty"`
    	DNS        types.DNS      `json:"dns,omitempty"`
    }

# **1、main函数** #

*  主要函数是skel.PluginMain，跟其他插件写法类似

func main() {
    	// Set up logging formatting.
    	log.SetFormatter(&logutils.Formatter{})
    
    	// Install a hook that adds file/line no information.
    	log.AddHook(&logutils.ContextHook{})
    
    	// Display the version on "-v", otherwise just delegate to the skel code.
    	// Use a new flag set so as not to conflict with existing libraries which use "flag"
    	flagSet := flag.NewFlagSet("calico-ipam", flag.ExitOnError)
    
    	version := flagSet.Bool("v", false, "Display version")
    	err := flagSet.Parse(os.Args[1:])
    
    	if err != nil {
    		fmt.Println(err)
    		os.Exit(1)
    	}
    
    	if *version {
    		fmt.Println(VERSION)
    		os.Exit(0)
    	}
    
    	skel.PluginMain(cmdAdd, cmdDel, cniSpecVersion.All)
    }

# **2、cmdAdd函数** #

###   2.1 将标准输入参数转为结构体NetConf ###

conf := types.NetConf{}
    	if err := json.Unmarshal(args.StdinData, &conf); err != nil {
    		return fmt.Errorf("failed to load netconf: %v", err)
    	}

###   2.2 CreateClient函数 ###

第三章讲解，设置datastore为etcdv2，并建立客户端连接

calicoClient, err := utils.CreateClient(conf)
    	if err != nil {
    		return err
    	}

###   2.3 GetIdentifiers函数 ###

将参数提取k8s相关的参数，比如container ID，pod name namespace 等

func GetIdentifiers(args *skel.CmdArgs) (workloadID string, orchestratorID string, err error) {
    	// Determine if running under k8s by checking the CNI args
    	k8sArgs := K8sArgs{}
    	if err = types.LoadArgs(args.Args, &k8sArgs); err != nil {
    		return workloadID, orchestratorID, err
    	}
    
    	if string(k8sArgs.K8S_POD_NAMESPACE) != "" && string(k8sArgs.K8S_POD_NAME) != "" {
    		workloadID = fmt.Sprintf("%s.%s", k8sArgs.K8S_POD_NAMESPACE, k8sArgs.K8S_POD_NAME)
    		orchestratorID = "k8s"
    	} else {
    		workloadID = args.ContainerID
    		orchestratorID = "cni"
    	}
    	return workloadID, orchestratorID, nil
    }

###   2.4 指定具体IP地址，这种情况使用少 ###

if ipamArgs.IP != nil {
    		fmt.Fprintf(os.Stderr, "Calico CNI IPAM request IP: %v\n", ipamArgs.IP)
    
    		// The hostname will be defaulted to the actual hostname if conf.Nodename is empty
    		assignArgs := ipam.AssignIPArgs{
    			IP:       cnet.IP{IP: ipamArgs.IP},
    			HandleID: &handleID,
    			Hostname: nodename,
    		}
    
    		logger.WithField("assignArgs", assignArgs).Info("Assigning provided IP")
    		err := calicoClient.IPAM().AssignIP(ctx, assignArgs)
    		if err != nil {
    			return err
    		}
    
    		var ipNetwork net.IPNet
    
    		if ipamArgs.IP.To4() == nil {
    			// It's an IPv6 address.
    			ipNetwork = net.IPNet{IP: ipamArgs.IP, Mask: net.CIDRMask(128, 128)}
    			r.IPs = append(r.IPs, &current.IPConfig{
    				Version: "6",
    				Address: ipNetwork,
    			})
    
    			logger.WithField("result.IPs", ipamArgs.IP).Info("Appending an IPv6 address to the result")
    		} else {
    			// It's an IPv4 address.
    			ipNetwork = net.IPNet{IP: ipamArgs.IP, Mask: net.CIDRMask(32, 32)}
    			r.IPs = append(r.IPs, &current.IPConfig{
    				Version: "4",
    				Address: ipNetwork,
    			})
    
    			logger.WithField("result.IPs", ipamArgs.IP).Info("Appending an IPv4 address to the result")
    		}
    	}

###   2.5 未指定IP ###

主要函数calicoClient.IPAM().AutoAssign函数第四章讲解

} else {
    
    		v4pools, err := utils.ResolvePools(ctx, calicoClient, conf.IPAM.IPv4Pools, true)
    		if err != nil {
    			return err
    		}
    
    		v6pools, err := utils.ResolvePools(ctx, calicoClient, conf.IPAM.IPv6Pools, false)
    		if err != nil {
    			return err
    		}
    
    		fmt.Fprintf(os.Stderr, "Calico CNI IPAM handle=%s\n", handleID)
    		assignArgs := ipam.AutoAssignArgs{
    			Num4:      num4,
    			Num6:      num6,
    			HandleID:  &handleID,
    			Hostname:  nodename,
    			IPv4Pools: v4pools,
    			IPv6Pools: v6pools,
    		}
    		logger.WithField("assignArgs", assignArgs).Info("Auto assigning IP")
    		assignedV4, assignedV6, err := calicoClient.IPAM().AutoAssign(ctx, assignArgs)
    		fmt.Fprintf(os.Stderr, "Calico CNI IPAM assigned addresses IPv4=%v IPv6=%v\n", assignedV4, assignedV6)
    
    		logger.WithFields(logrus.Fields{"result.IPs": r.IPs}).Info("IPAM Result")
    	}

# 3. CreateClient函数 #

*   LoadClientConfigFromBytes初始化calicoAPIConfig结构体，设置datastore为etcdv3
 *  建立客户端连接，设置环境变量

func CreateClient(conf types.NetConf) (client.Interface, error) {
    	if err := ValidateNetworkName(conf.Name); err != nil {
    		return nil, err
    	}
    
    	// Use the config file to override environment variables.
    	// These variables will be loaded into the client config.
    	if conf.EtcdAuthority != "" {
    		if err := os.Setenv("ETCD_AUTHORITY", conf.EtcdAuthority); err != nil {
    			return nil, err
    		}
    	}
    	。。。。。。。。。。。。。。。。。
    	// Load the client config from the current environment.
    	clientConfig, err := apiconfig.LoadClientConfig("")
    	if err != nil {
    		return nil, err
    	}
    
    	// Create a new client.
    	calicoClient, err := client.New(*clientConfig)
    	if err != nil {
    		return nil, err
    	}
    	return calicoClient, nil
    }

*  CNI\_COMMAND=ADD CNI\_CONTAINERID=24d84edb48519f2e21886157260482e99a9471e3a291436779d5d7c8acfb0d25 CNI\_NETNS=/proc/8683/ns/net
 *  CNI\_ARGS=IgnoreUnknown=1;IgnoreUnknown=1;
 *  K8S\_POD\_NAMESPACE=default;
 *  K8S\_POD\_NAME=httpserver-d-649f47c8f8-pgc98;
 *  K8S\_POD\_INFRA\_CONTAINER\_ID=24d84edb48519f2e21886157260482e99a9471e3a291436779d5d7c8acfb0d25
 *  CNI\_IFNAME=eth0
 *  CNI\_PATH=/opt/multus/bin:/usr/sbin
 *  LANG=en\_US.UTF-8
 *  PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin ETCD\_ENDPOINTS=https://10.12.51.233:2379,https://10.12.51.234:2379,https://10.12.51.235:2379 ETCD\_KEY\_FILE=/etc/calico/ssl/calico-key.pem
 *  ETCD\_CERT\_FILE=/etc/calico/ssl/calico.pem
 *  ETCD\_CA\_CERT\_FILE=/etc/calico/ssl/ca.pem
 *  KUBECONFIG=/root/.kube/config

###   3.1 LoadClientConfig函数 ###

*  初始化结构体CalicoAPIConfig，设置datastore为etcdv3

// LoadClientConfig loads the ClientConfig from the supplied bytes containing
    // YAML or JSON format data.
    func LoadClientConfigFromBytes(b []byte) (*CalicoAPIConfig, error) {
    	var c CalicoAPIConfig
    
    	// Default the backend type to be etcd v3.  This will be overridden if
    	// explicitly specified in the file.
    	log.Debug("Loading config from JSON or YAML data")
    	c = CalicoAPIConfig{
    		Spec: CalicoAPIConfigSpec{
    			DatastoreType: EtcdV3,
    		},
    	}
    
    	if err := yaml.UnmarshalStrict(b, &c); err != nil {
    		return nil, err
    	}
    
    	// Validate the version and kind.
    	if c.APIVersion != apiv3.GroupVersionCurrent {
    		return nil, errors.New("invalid config file: unknown APIVersion '" + c.APIVersion + "'")
    	}
    	if c.Kind != KindCalicoAPIConfig {
    		return nil, errors.New("invalid config file: expected kind '" + KindCalicoAPIConfig + "', got '" + c.Kind + "'")
    	}
    
    	log.Debug("Datastore type: ", c.Spec.DatastoreType)
    	return &c, nil
    }

# 4. libcalico-go项目 #

*   [https://github.com/projectcalico/libcalico-go][https_github.com_projectcalico_libcalico-go]

路径 libcalico-go/lib/ipam/ipam.go

###   4.1 AutoAssign函数 ###

分配ipv4，参数未有IPv4Pools，直接进入主要函数autoAssign函数

// AutoAssign automatically assigns one or more IP addresses as specified by the
    // provided AutoAssignArgs.  AutoAssign returns the list of the assigned IPv4 addresses,
    // and the list of the assigned IPv6 addresses.
    func (c ipamClient) AutoAssign(ctx context.Context, args AutoAssignArgs) ([]net.IP, []net.IP, error) {
    	// Determine the hostname to use - prefer the provided hostname if
    	// non-nil, otherwise use the hostname reported by os.
    	hostname, err := decideHostname(args.Hostname)
    	if err != nil {
    		return nil, nil, err
    	}
    	log.Infof("Auto-assign %d ipv4, %d ipv6 addrs for host '%s'", args.Num4, args.Num6, hostname)
    
    	var v4list, v6list []net.IP
    
    	if args.Num4 != 0 {
    		// Assign IPv4 addresses.
    		log.Debugf("Assigning IPv4 addresses")
    		for _, pool := range args.IPv4Pools {
    			if pool.IP.To4() == nil {
    				return nil, nil, fmt.Errorf("provided IPv4 IPPools list contains one or more IPv6 IPPools")
    			}
    		}
    		v4list, err = c.autoAssign(ctx, args.Num4, args.HandleID, args.Attrs, args.IPv4Pools, ipv4, hostname)
    		if err != nil {
    			log.Errorf("Error assigning IPV4 addresses: %v", err)
    			return nil, nil, err
    		}
    	}
    
    	if args.Num6 != 0 {
    		// If no err assigning V4, try to assign any V6.
    		log.Debugf("Assigning IPv6 addresses")
    		for _, pool := range args.IPv6Pools {
    			if pool.IP.To4() != nil {
    				return nil, nil, fmt.Errorf("provided IPv6 IPPools list contains one or more IPv4 IPPools")
    			}
    		}
    		v6list, err = c.autoAssign(ctx, args.Num6, args.HandleID, args.Attrs, args.IPv6Pools, ipv6, hostname)
    		if err != nil {
    			log.Errorf("Error assigning IPV6 addresses: %v", err)
    			return nil, nil, err
    		}
    	}
    
    	return v4list, v6list, nil
    }

###   4.2 autoAssign函数 ###

内容比较多，分开讲解

###   4.2.1 getAffineBlocks函数 ###

调用etcd客户端中的方法List，List Key: /calico/ipam/v2/host/bj-zw-k8s-computer-h-51-186.xcm.cn/ipv4/block

后端路径在libcalico-go/lib/backend/etcd/etcd.go中的List方法，比较容易理解

func (rw blockReaderWriter) getAffineBlocks(host string, ver ipVersion, pools []cnet.IPNet) ([]cnet.IPNet, error) {
    	// Lookup all blocks by providing an empty BlockListOptions
    	// to the List operation.
    	opts := model.BlockAffinityListOptions{Host: host, IPVersion: ver.Number}
    	datastoreObjs, err := rw.client.Backend.List(opts)
    	if err != nil {
    		if _, ok := err.(errors.ErrorResourceDoesNotExist); ok {
    			// The block path does not exist yet.  This is OK - it means
    			// there are no affine blocks.
    			return []cnet.IPNet{}, nil
    
    		} else {
    			log.Errorf("Error getting affine blocks: %s", err)
    			return nil, err
    		}
    	}
    
    	// Iterate through and extract the block CIDRs.
    	ids := []cnet.IPNet{}
    	for _, o := range datastoreObjs {
    		k := o.Key.(model.BlockAffinityKey)
    
    		// Add the block if no IP pools were specified, or if IP pools were specified
    		// and the block falls within the given IP pools.
    		if len(pools) == 0 {
    			ids = append(ids, k.CIDR)
    		} else {
    			for _, pool := range pools {
    				if pool.Contains(k.CIDR.IPNet.IP) {
    					ids = append(ids, k.CIDR)
    					break
    				}
    			}
    		}
    	}
    	return ids, nil
    }

###   4.2.2 assignFronExistingBlock函数 ###

第五章讲解

for len(ips) < num {
    		if len(affBlocks) == 0 {
    			log.Infof("Ran out of existing affine blocks for host '%s'", host)
    			break
    		}
    		cidr := affBlocks[0]
    		affBlocks = affBlocks[1:]
    		ips, _ = c.assignFromExistingBlock(cidr, num, handleID, attrs, host, true)
    		log.Debugf("Block '%s' provided addresses: %v", cidr.String(), ips)
    	}

###   4.2.3 成功将更新etcd记录 ###

// Update the block using CAS by passing back the original
    		// KVPair.
    		obj.Value = b.AllocationBlock
    		_, err = c.client.Backend.Update(obj)
    		if err != nil {
    			log.Infof("Failed to update block '%s' - try again", b.CIDR.String())
    			if handleID != nil {
    				c.decrementHandle(*handleID, blockCIDR, num)
    			}
    			continue
    		}

# 5. autoAssign函数 #

路径libcalico-go/lib/client/ipam\_block.go

获取IP，分为几个记录：

*  allocations
 *  unallocated
 *  attributes

func (b *allocationBlock) autoAssign(
    	num int, handleID *string, host string, attrs map[string]string, affinityCheck bool) ([]cnet.IP, error) {
    
    	// Determine if we need to check for affinity.
    	checkAffinity := b.StrictAffinity || affinityCheck
    	if checkAffinity && b.Affinity != nil && !hostAffinityMatches(host, b.AllocationBlock) {
    		// Affinity check is enabled but the host does not match - error.
    		s := fmt.Sprintf("Block affinity (%s) does not match provided (%s)", *b.Affinity, host)
    		return nil, errors.New(s)
    	}
    
    	// Walk the allocations until we find enough addresses.
    	ordinals := []int{}
    	for len(b.Unallocated) > 0 && len(ordinals) < num {
    		ordinals = append(ordinals, b.Unallocated[0])
    		b.Unallocated = b.Unallocated[1:]
    	}
    
    	// Create slice of IPs and perform the allocations.
    	ips := []cnet.IP{}
    	for _, o := range ordinals {
    		attrIndex := b.findOrAddAttribute(handleID, attrs)
    		b.Allocations[o] = &attrIndex
    
    		ips = append(ips, incrementIP(cnet.IP{b.CIDR.IP}, big.NewInt(int64(o))))
    	}
    
    	log.Debugf("Block %s returned ips: %v", b.CIDR.String(), ips)
    	return ips, nil
    }

calico 与 宿主机性能对比，性能比较接近宿主机，可能由于分别测试有误差

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][]

[https_github.com_projectcalico_cni-plugin]: https://github.com/projectcalico/cni-plugin
[https_github.com_projectcalico_libcalico-go]: https://github.com/projectcalico/libcalico-go
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: /images/20220510/7089d031725042e8b7bfb73b0fa45aa2.png