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

https://github.com/projectcalico/cni-plugin本文基于v1.11.0，v3 主要差别是etcdv3以及设置的key路径calico解决不同物理机上容器之间的通信，而calico-plugin是在k8s创建Pod时为Pod设置虚拟网卡(容器中的eth0和lo网卡)，calico-plugin是由两个静态的二进制文件组成，由kubelet以命令行的形式调...

文章共2,324字 · 阅读需要大约8分钟

一键AI生成摘要，助你高效阅读

问答

张忠琳

5509人浏览 · 2018-09-27 15:30:35

张忠琳 · 2018-09-27 15:30:35 发布

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

路径 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 与宿主机性能对比，性能比较接近宿主机，可能由于分别测试有误差