Kubernetes | 读代码：Flannel Backends

2021 01 25, Mon

最近有点累，加上之前好奇CNI都是怎么实现的，所以这两天主要就是读读Flannel的实现。

这次探究的代码以0.13为例，Github上的Flannel仓库访问可能比较慢，可以用Gitee的镜像仓库

flannel是什么

flannel是一个K8S的网络实现，根据文档描述，flannel 主要是用来在节点上配置 K8S 节点的子网并打通节点上子网网段间的网络通信：

Flannel runs a small, single binary agent called flanneld on each host, and is responsible for allocating a subnet lease to each host out of a larger, preconfigured address space.

但是设计上并不希望太多的控制 pod 中网络的配置，只是有一个插件调用 bridge CNI 插件。

Flannel does not control how containers are networked to the host, only how the traffic is transported between hosts. However, flannel does provide a CNI plugin for Kubernetes and a guidance on integrating with Docker.

flannel 比较简单，所以我们用 flannel 作为起点来了解 Kubernetes 的网络架构。

flannel模块

项目直接放在了仓库根目录。

根据go.mod，可以得到下面的信息

使用go1.14。之后如果遇到了问题，以go 1.14.x的最高版本为准
看到了github.com/containernetworking/plugins v0.8.6
- 疑似编写CNI用的库。
- 大概浏览Git仓库，看起来是一个插件的仓库，可能是借用了其中的package。
- 搜索了一下，实际上只是用了github.com/containernetworking/plugins/pkg/utils/sysctl来配置vxlan网络中网络设备的accept_ra参数
看到了一些云厂商的库，肯定是用来操作云环境里面的资源的。分别是
- Google Cloud: cloud.google.com/go v0.9.1-0.20170629112852-7c205da84d8d
- AWS: github.com/aws/aws-sdk-go v1.12.54
- 阿里云: github.com/denverdino/aliyungo v0.0.0-20170629053852-f6cab0c35083
看到一些和系统有关的包，顺便看了一眼仓库里面的目录，似乎是网络协议的实现或者是操作系统资源的：
- 管理Windows下的容器: github.com/Microsoft/hcsshim v0.8.6
- StrongSwan（IPSec的一个实现）: github.com/bronze1man/goStrongswanVici v0.0.0-20171013065002-4d72634a2f11
- iptables: github.com/coreos/go-iptables v0.4.5
- systemd: github.com/coreos/go-systemd v0.0.0-20180511133405-39ca1b05acc7
- netlink API: github.com/vishvananda/netlink v0.0.0-20181108222139-023a6dafdcdf
- network namespace: github.com/vishvananda/netns v0.0.0-20180720170159-13995c7128cc
剩下的都无关紧要

项目结构

刨除没有用的文件，大致是这样的结构：

Documentation: 文档
backends: 网络实现
network: iptables类工具包
pkg:
- ip: 看起来主要是ip类的工具，还有就是利用github.com/vishvananda/netlink获取信息和操作网卡上的地址
- ns: 建立一个network namespace的包
- powershell: 看起来是用powershell跑命令的包
- routing: 定义了路由器接口和路由结构体的包
subnet: 在etcd中CRUD子网记录的包
version: 用来编译时注入版本信息的包

除了backends和subnet以外没有其他和逻辑有用的包了。subnet中就是常规的CRUD逻辑，没有必要。backends涉及到flannel的具体实现，后面具体说。

main.go

import的部分有10行都是import for side effects。根据名字，应该大部分都和backends有关系，之后看backends时一并分析。

init函数全部是读取我们直接从main函数看起。

一开始都是读取配置和检查NIC接口，一直到241行开始才是正式的逻辑：

新建Subnet Manager用来和etcd交互CRUD子网信息，设置信号Int/Term准备后续处理，准备一个WaitGroup用来跑多个goroutine：

    sm, err := newSubnetManager()
    // ...
    sigs := make(chan os.Signal, 1)
    signal.Notify(sigs, os.Interrupt, syscall.SIGTERM)

    ctx, cancel := context.WithCancel(context.Background())
    wg := sync.WaitGroup{}

启动一些goroutine，用来处理通信啊，准备退出啊什么的：


    wg.Add(1)
    go func() {
        shutdownHandler(ctx, sigs, cancel)
        wg.Done()
    }()

    // ...
    if opts.healthzPort > 0 {
        // It's not super easy to shutdown the HTTP server so don't attempt to stop it cleanly
        go mustRunHealthz()
    }

获取网络的配置，获取到网络的实际实现（be backend.Backend）

    config, err := getConfig(ctx, sm)
    if err == errCanceled {
        wg.Wait()
        os.Exit(0)
    }

    bm := backend.NewManager(ctx, sm, extIface)
    be, err := bm.GetBackend(config.BackendType)
    if err != nil {
        log.Errorf("Error fetching backend: %s", err)
        cancel()
        wg.Wait()
        os.Exit(1)
    }

这里be实际配置了网络，顺便配置了iptables，把网络配置写到了本地。

    bn, err := be.RegisterNetwork(ctx, &wg, config)
    if err != nil {
        log.Errorf("Error registering network: %s", err)
        cancel()
        wg.Wait()
        os.Exit(1)
    }

    // Set up ipMasq if needed
    if opts.ipMasq {
        if err = recycleIPTables(config.Network, bn.Lease()); err != nil {
            log.Errorf("Failed to recycle IPTables rules, %v", err)
            cancel()
            wg.Wait()
            os.Exit(1)
        }
        log.Infof("Setting up masking rules")
        go network.SetupAndEnsureIPTables(network.MasqRules(config.Network, bn.Lease()), opts.iptablesResyncSeconds)
    }

    // Always enables forwarding rules. This is needed for Docker versions >1.13 (https://docs.docker.com/engine/userguide/networking/default_network/container-communication/  #container-communication-between-hosts)
    // In Docker 1.12 and earlier, the default FORWARD chain policy was ACCEPT.
    // In Docker 1.13 and later, Docker sets the default policy of the FORWARD chain to DROP.
    if opts.iptablesForwardRules {
        log.Infof("Changing default FORWARD chain policy to ACCEPT")
        go network.SetupAndEnsureIPTables(network.ForwardRules(config.Network.String()), opts.iptablesResyncSeconds)
    }

    if err := WriteSubnetFile(opts.subnetFile, config.Network, opts.ipMasq, bn); err != nil {
        // Continue, even though it failed.
        log.Warningf("Failed to write subnet file: %s", err)
    } else {
        log.Infof("Wrote subnet file to %s", opts.subnetFile)
    }

    // Start "Running" the backend network. This will block until the context is done so run in another goroutine.
    log.Info("Running backend.")
    wg.Add(1)
    go func() {
        bn.Run(ctx)
        wg.Done()
    }()

剩下的逻辑主要就是等待退出。除此之外还设置了systemd的状态

    daemon.SdNotify(false, "READY=1")

    // Kube subnet mgr doesn't lease the subnet for this node - it just uses the podCidr that's already assigned.
    if !opts.kubeSubnetMgr {
        err = MonitorLease(ctx, sm, bn, &wg)
        if err == errInterrupted {
            // The lease was "revoked" - shut everything down
            cancel()
        }
    }

    log.Info("Waiting for all goroutines to exit")
    // Block waiting for all the goroutines to finish.
    wg.Wait()
    log.Info("Exiting cleanly...")
    os.Exit(0)

backends

上面看了半天实际有用的其实就是backends的部分。backends目录有这些

common.go: 主要是定义了Backend和Network接口，Backend需要做的主要就是实现RegisterNetwork，在接管网络的时候做配置。Network主要是返回网络的配置（MTU、Lease），以及处理网络的变化（Run）。
manager: 注册后端实现、管理Backend的实例
simple_network.go: 实现了一个只保存信息什么都不做的网络，满足了Network接口。
route_network.go: 实现了一个简单的路由网络，满足了Network接口。在子网或者网络设备变化时更新自身状态。
其他目录: 各种类型的Backend的实现

我们今天先读一些比较简单的吧。

alloc

alloc包中包含一个只分配子网、其他的什么都不做的Backend实现。

RegisterNetwork只是在etcd中划分出来了一个子网。

type AllocBackend struct {
    sm       subnet.Manager
    extIface *backend.ExternalInterface
}

func New(sm subnet.Manager, extIface *backend.ExternalInterface) (backend.Backend, error) {
    be := AllocBackend{
        sm:       sm,
        extIface: extIface,
    }
    return &be, nil
}

func (be *AllocBackend) RegisterNetwork(ctx context.Context, wg *sync.WaitGroup, config *subnet.Config) (backend.Network, error) {
    attrs := subnet.LeaseAttrs{
        PublicIP: ip.FromIP(be.extIface.ExtAddr),
    }

    l, err := be.sm.AcquireLease(ctx, &attrs)
    switch err {
    case nil:
        return &backend.SimpleNetwork{
            SubnetLease: l,
            ExtIface:    be.extIface,
        }, nil

    case context.Canceled, context.DeadlineExceeded:
        return nil, err

    default:
        return nil, fmt.Errorf("failed to acquire lease: %v", err)
    }
}