Kubernetes CNI 与 Flannel VXLAN 模式

$ tree -L 2 ./plugins
plugins
├── ipam
│   ├── dhcp
│   ├── host-local
│   └── static
├── linux_only.txt
├── main
│   ├── bridge
│   ├── host-device
│   ├── ipvlan
│   ├── loopback
│   ├── macvlan
│   ├── ptp
│   ├── vlan
│   └── windows
├── meta
│   ├── bandwidth
│   ├── firewall
│   ├── portmap
│   ├── sbr
│   ├── tuning
│   └── vrf
├── sample
│   ├── main.go
│   ├── README.md
│   ├── sample_linux_test.go
│   └── sample_suite_test.go
└── windows_only.txt

$ ip addr show cni0
5: cni0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1450 qdisc noqueue state UP group default qlen 1000
    link/ether 16:99:c2:e5:fc:85 brd ff:ff:ff:ff:ff:ff
    inet 10.244.0.1/24 brd 10.244.0.255 scope global cni0
       valid_lft forever preferred_lft forever
    inet6 fe80::1499:c2ff:fee5:fc85/64 scope link
       valid_lft forever preferred_lft forever

{
  "name": "cbr0",
  "cniVersion": "0.3.1",
  "plugins": [
    {
      "type": "flannel",
      "delegate": {
        "hairpinMode": true,
        "isDefaultGateway": true
      }
    },
    {
      "type": "portmap",
      "capabilities": {
        "portMappings": true
      }
    }
  ]
}

const defaultBrName = "cni0"

func ensureBridge(brName string, mtu int, promiscMode, vlanFiltering bool) (*netlink.Bridge, error) {
    br := &netlink.Bridge{
        LinkAttrs: netlink.LinkAttrs{
            Name: brName,
            MTU:  mtu,
            // Let kernel use default txqueuelen; leaving it unset
            // means 0, and a zero-length TX queue messes up FIFO
            // traffic shapers which use TX queue length as the
            // default packet limit
            TxQLen: -1,
        },
    }
    if vlanFiltering {
        br.VlanFiltering = &vlanFiltering
    }

    err := netlink.LinkAdd(br)
    if err != nil && err != syscall.EEXIST {
        return nil, fmt.Errorf("could not add %q: %v", brName, err)
    }

    if promiscMode {
        if err := netlink.SetPromiscOn(br); err != nil {
            return nil, fmt.Errorf("could not set promiscuous mode on %q: %v", brName, err)
        }
    }

    // Re-fetch link to read all attributes and if it already existed,
    // ensure it's really a bridge with similar configuration
    br, err = bridgeByName(brName)
    if err != nil {
        return nil, err
    }

    // we want to own the routes for this interface
    _, _ = sysctl.Sysctl(fmt.Sprintf("net/ipv6/conf/%s/accept_ra", brName), "0")

    if err := netlink.LinkSetUp(br); err != nil {
        return nil, err
    }

    return br, nil
}

func cmdAdd(args *skel.CmdArgs) error {
    var success bool = false

    n, cniVersion, err := loadNetConf(args.StdinData)
    if err != nil {
        return err
    }

    // a lot of code here
}

func (c *CNIConfig) addNetwork(ctx context.Context, name, cniVersion string, net *NetworkConfig, prevResult types.Result, rt *RuntimeConf) (types.Result, error) {
    c.ensureExec()
    pluginPath, err := c.exec.FindInPath(net.Network.Type, c.Path)
    if err != nil {
        return nil, err
    }

    newConf, err := buildOneConfig(name, cniVersion, net, prevResult, rt)
    if err != nil {
        return nil, err
    }

    return invoke.ExecPluginWithResult(ctx, pluginPath, newConf.Bytes, c.args("ADD", rt), c.exec)
}

    netns, err := ns.GetNS(args.Netns)
    if err != nil {
        return fmt.Errorf("failed to open netns %q: %v", args.Netns, err)
    }
    defer netns.Close()

    hostInterface, containerInterface, err := setupVeth(netns, br, args.IfName, n.MTU, n.HairpinMode, n.Vlan)
    if err != nil {
        return err
    }

    // a lot of code here

func setupVeth(netns ns.NetNS, br *netlink.Bridge, ifName string, mtu int, hairpinMode bool, vlanID int) (*current.Interface, *current.Interface, error) {
    contIface := &current.Interface{}
    hostIface := &current.Interface{}

    err := netns.Do(func(hostNS ns.NetNS) error {
        // create the veth pair in the container and move host end into host netns
        hostVeth, containerVeth, err := ip.SetupVeth(ifName, mtu, hostNS)
        if err != nil {
            return err
        }
        contIface.Name = containerVeth.Name
        contIface.Mac = containerVeth.HardwareAddr.String()
        contIface.Sandbox = netns.Path()
        hostIface.Name = hostVeth.Name
        return nil
    })
    if err != nil {
        return nil, nil, err
    }

    // a lot of code here
}

    // connect host veth end to the bridge
    if err := netlink.LinkSetMaster(hostVeth, br); err != nil {
        return nil, nil, fmt.Errorf("failed to connect %q to bridge %v: %v", hostVeth.Attrs().Name, br.Attrs().Name, err)
    }

    // set hairpin mode
    if err = netlink.LinkSetHairpin(hostVeth, hairpinMode); err != nil {
        return nil, nil, fmt.Errorf("failed to setup hairpin mode for %v: %v", hostVeth.Attrs().Name, err)
    }

        // run the IPAM plugin and get back the config to apply
        r, err := ipam.ExecAdd(n.IPAM.Type, args.StdinData)
        if err != nil {
            return err
        }

        // release IP in case of failure
        defer func() {
            if !success {
                ipam.ExecDel(n.IPAM.Type, args.StdinData)
            }
        }()

        // Convert whatever the IPAM result was into the current Result type
        ipamResult, err := current.NewResultFromResult(r)
        if err != nil {
            return err
        }

        result.IPs = ipamResult.IPs
        result.Routes = ipamResult.Routes

        // Configure the container hardware address and IP address(es)
        if err := netns.Do(func(_ ns.NetNS) error {
            // a lot of code here

            // Add the IP to the interface
            if err := ipam.ConfigureIface(args.IfName, result); err != nil {
                return err
            }
            return nil
        }); err != nil {
            return err
        }

// ConfigureIface takes the result of IPAM plugin and
// applies to the ifName interface
func ConfigureIface(ifName string, res *current.Result) error {
    if len(res.Interfaces) == 0 {
        return fmt.Errorf("no interfaces to configure")
    }

    link, err := netlink.LinkByName(ifName)
    if err != nil {
        return fmt.Errorf("failed to lookup %q: %v", ifName, err)
    }

    if err := netlink.LinkSetUp(link); err != nil {
        return fmt.Errorf("failed to set %q UP: %v", ifName, err)
    }

    var v4gw, v6gw net.IP
    var has_enabled_ipv6 bool = false
    for _, ipc := range res.IPs {
        // a lot of code here
        addr := &netlink.Addr{IPNet: &ipc.Address, Label: ""}
        if err = netlink.AddrAdd(link, addr); err != nil {
            return fmt.Errorf("failed to add IP addr %v to %q: %v", ipc, ifName, err)
        }

        gwIsV4 := ipc.Gateway.To4() != nil
        if gwIsV4 && v4gw == nil {
            v4gw = ipc.Gateway
        } else if !gwIsV4 && v6gw == nil {
            v6gw = ipc.Gateway
        }
    }

    for _, r := range res.Routes {
        // a lot of code here
        if err = ip.AddRoute(&r.Dst, gw, link); err != nil {
            // we skip over duplicate routes as we assume the first one wins
            if !os.IsExist(err) {
                return fmt.Errorf("failed to add route '%v via %v dev %v': %v", r.Dst, gw, ifName, err)
            }
        }
    }

    return nil

$ ip route
10.211.55.0/24 dev eth0 proto kernel scope link src 10.211.55.114 metric 100
10.244.0.0/24 dev cni0 proto kernel scope link src 10.244.0.1
10.244.1.0/24 via 10.244.1.0 dev flannel.1 onlink

$ ps -ef | grep flanneld
root      5476  5371  0 05:21 ?        00:00:03 /opt/bin/flanneld --ip-masq --kube-subnet-mgr

func newSubnetManager(ctx context.Context) (subnet.Manager, error) {
    if opts.kubeSubnetMgr {
        return kube.NewSubnetManager(ctx, opts.kubeApiUrl, opts.kubeConfigFile, opts.kubeAnnotationPrefix, opts.netConfPath, opts.setNodeNetworkUnavailable)
    }

    // a lot of code here
}

    indexer, controller := cache.NewIndexerInformer(
        &cache.ListWatch{
            ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                return ksm.client.CoreV1().Nodes().List(ctx, options)
            },
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                return ksm.client.CoreV1().Nodes().Watch(ctx, options)
            },
        },
        // a lot of code here

$ kubectl describe node k8s-test27
Name:               k8s-test27
Roles:              master
Labels:             beta.kubernetes.io/arch=amd64
                    beta.kubernetes.io/os=linux
                    kubernetes.io/arch=amd64
                    kubernetes.io/hostname=k8s-test27
                    kubernetes.io/os=linux
                    node-role.kubernetes.io/master=
Annotations:        csi.volume.kubernetes.io/nodeid: {"rbd.csi.ceph.com":"k8s-test27"}
                    flannel.alpha.coreos.com/backend-data: {"VNI":1,"VtepMAC":"72:64:92:59:50:bc"}
                    flannel.alpha.coreos.com/backend-type: vxlan
                    flannel.alpha.coreos.com/kube-subnet-manager: true
                    flannel.alpha.coreos.com/public-ip: 10.211.55.114
                    kubeadm.alpha.kubernetes.io/cri-socket: /var/run/dockershim.sock
                    node.alpha.kubernetes.io/ttl: 0
                    volumes.kubernetes.io/controller-managed-attach-detach: true

func (n *RouteNetwork) handleSubnetEvents(batch []subnet.Event) {
    for _, evt := range batch {
        switch evt.Type {
        case subnet.EventAdded:
            if evt.Lease.Attrs.BackendType != n.BackendType {
                log.Warningf("Ignoring non-%v subnet: type=%v", n.BackendType, evt.Lease.Attrs.BackendType)
                continue
            }

            if evt.Lease.EnableIPv4 {
                log.Infof("Subnet added: %v via %v", evt.Lease.Subnet, evt.Lease.Attrs.PublicIP)

                route := n.GetRoute(&evt.Lease)
                routeAdd(route, netlink.FAMILY_V4, n.addToRouteList, n.removeFromV4RouteList)
            }

            if evt.Lease.EnableIPv6 {
                log.Infof("Subnet added: %v via %v", evt.Lease.IPv6Subnet, evt.Lease.Attrs.PublicIPv6)

                route := n.GetV6Route(&evt.Lease)
                routeAdd(route, netlink.FAMILY_V6, n.addToV6RouteList, n.removeFromV6RouteList)
            }

        case subnet.EventRemoved:
            if evt.Lease.Attrs.BackendType != n.BackendType {
                log.Warningf("Ignoring non-%v subnet: type=%v", n.BackendType, evt.Lease.Attrs.BackendType)
                continue
            }

            if evt.Lease.EnableIPv4 {
                log.Info("Subnet removed: ", evt.Lease.Subnet)

                route := n.GetRoute(&evt.Lease)
                // Always remove the route from the route list.
                n.removeFromV4RouteList(*route)

                if err := netlink.RouteDel(route); err != nil {
                    log.Errorf("Error deleting route to %v: %v", evt.Lease.Subnet, err)
                }
            }

            if evt.Lease.EnableIPv6 {
                log.Info("Subnet removed: ", evt.Lease.IPv6Subnet)

                route := n.GetV6Route(&evt.Lease)
                // Always remove the route from the route list.
                n.removeFromV6RouteList(*route)

                if err := netlink.RouteDel(route); err != nil {
                    log.Errorf("Error deleting route to %v: %v", evt.Lease.IPv6Subnet, err)
                }
            }

        default:
            log.Error("Internal error: unknown event type: ", int(evt.Type))
        }
    }
}