You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
// TODO: Remove this once the kernel bug (#20096) is fixed.
/*Copyright 2014 The Kubernetes Authors.Licensed under the Apache License, Version 2.0 (the "License");you may not use this file except in compliance with the License.You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0Unless required by applicable law or agreed to in writing, softwaredistributed under the License is distributed on an "AS IS" BASIS,WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.See the License for the specific language governing permissions andlimitations under the License.*/package kubenet
import (
"context""fmt""io/ioutil""net""strings""sync""time""github.com/containernetworking/cni/libcni"
cnitypes "github.com/containernetworking/cni/pkg/types"
cnitypes020 "github.com/containernetworking/cni/pkg/types/020""github.com/vishvananda/netlink""golang.org/x/sys/unix"
utilerrors "k8s.io/apimachinery/pkg/util/errors"
utilnet "k8s.io/apimachinery/pkg/util/net"
utilsets "k8s.io/apimachinery/pkg/util/sets""k8s.io/klog/v2"
kubeletconfig "k8s.io/kubernetes/pkg/kubelet/apis/config"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container""k8s.io/kubernetes/pkg/kubelet/dockershim/network""k8s.io/kubernetes/pkg/kubelet/dockershim/network/hostport""k8s.io/kubernetes/pkg/util/bandwidth"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
utilsysctl "k8s.io/kubernetes/pkg/util/sysctl"
utilexec "k8s.io/utils/exec"
utilebtables "k8s.io/utils/net/ebtables"
utilfeature "k8s.io/apiserver/pkg/util/feature"
kubefeatures "k8s.io/kubernetes/pkg/features"
netutils "k8s.io/utils/net"
)
const (
BridgeName="cbr0"DefaultCNIDir="/opt/cni/bin"sysctlBridgeCallIPTables="net/bridge/bridge-nf-call-iptables"// fallbackMTU is used if an MTU is not specified, and we cannot determine the MTUfallbackMTU=1460// ebtables Chain to store dedup rulesdedupChain=utilebtables.Chain("KUBE-DEDUP")
zeroCIDRv6="::/0"zeroCIDRv4="0.0.0.0/0"NET_CONFIG_TEMPLATE=`{ "cniVersion": "0.1.0", "name": "kubenet", "type": "bridge", "bridge": "%s", "mtu": %d, "addIf": "%s", "isGateway": true, "ipMasq": false, "hairpinMode": %t, "ipam": { "type": "host-local", "ranges": [%s], "routes": [%s] }}`
)
// CNI plugins required by kubenet in /opt/cni/bin or user-specified directoryvarrequiredCNIPlugins= [...]string{"bridge", "host-local", "loopback"}
typekubenetNetworkPluginstruct {
network.NoopNetworkPluginhost network.HostnetConfig*libcni.NetworkConfigloConfig*libcni.NetworkConfigcniConfig libcni.CNIbandwidthShaper bandwidth.Shapermu sync.Mutex//Mutex for protecting podIPs map, netConfig, and shaper initializationpodIPsmap[kubecontainer.ContainerID]utilsets.Stringmtuintexecer utilexec.InterfacensenterPathstringhairpinMode kubeletconfig.HairpinModehostportManager hostport.HostPortManagerhostportManagerv6 hostport.HostPortManageriptables utiliptables.Interfaceiptablesv6 utiliptables.Interfacesysctl utilsysctl.Interfaceebtables utilebtables.Interface// binDirs is passed by kubelet cni-bin-dir parameter.// kubenet will search for CNI binaries in DefaultCNIDir first, then continue to binDirs.binDirs []stringnonMasqueradeCIDRstringcacheDirstringpodCIDRs []*net.IPNet
}
funcNewPlugin(networkPluginDirs []string, cacheDirstring) network.NetworkPlugin {
execer:=utilexec.New()
iptInterface:=utiliptables.New(execer, utiliptables.ProtocolIPv4)
iptInterfacev6:=utiliptables.New(execer, utiliptables.ProtocolIPv6)
return&kubenetNetworkPlugin{
podIPs: make(map[kubecontainer.ContainerID]utilsets.String),
execer: utilexec.New(),
iptables: iptInterface,
iptablesv6: iptInterfacev6,
sysctl: utilsysctl.New(),
binDirs: append([]string{DefaultCNIDir}, networkPluginDirs...),
hostportManager: hostport.NewHostportManager(iptInterface),
hostportManagerv6: hostport.NewHostportManager(iptInterfacev6),
nonMasqueradeCIDR: "10.0.0.0/8",
cacheDir: cacheDir,
podCIDRs: make([]*net.IPNet, 0),
}
}
func (plugin*kubenetNetworkPlugin) Init(host network.Host, hairpinMode kubeletconfig.HairpinMode, nonMasqueradeCIDRstring, mtuint) error {
plugin.host=hostplugin.hairpinMode=hairpinModeplugin.nonMasqueradeCIDR=nonMasqueradeCIDRplugin.cniConfig=&libcni.CNIConfig{Path: plugin.binDirs}
ifmtu==network.UseDefaultMTU {
iflink, err:=findMinMTU(); err==nil {
plugin.mtu=link.MTUklog.V(5).Infof("Using interface %s MTU %d as bridge MTU", link.Name, link.MTU)
} else {
plugin.mtu=fallbackMTUklog.Warningf("Failed to find default bridge MTU, using %d: %v", fallbackMTU, err)
}
} else {
plugin.mtu=mtu
}
// Since this plugin uses a Linux bridge, set bridge-nf-call-iptables=1// is necessary to ensure kube-proxy functions correctly.//// This will return an error on older kernel version (< 3.18) as the module// was built-in, we simply ignore the error here. A better thing to do is// to check the kernel version in the future.plugin.execer.Command("modprobe", "br-netfilter").CombinedOutput()
err:=plugin.sysctl.SetSysctl(sysctlBridgeCallIPTables, 1)
iferr!=nil {
klog.Warningf("can't set sysctl %s: %v", sysctlBridgeCallIPTables, err)
}
plugin.loConfig, err=libcni.ConfFromBytes([]byte(`{ "cniVersion": "0.1.0", "name": "kubenet-loopback", "type": "loopback"}`))
iferr!=nil {
returnfmt.Errorf("failed to generate loopback config: %v", err)
}
plugin.nsenterPath, err=plugin.execer.LookPath("nsenter")
iferr!=nil {
returnfmt.Errorf("failed to find nsenter binary: %v", err)
}
// Need to SNAT outbound traffic from clusteriferr=plugin.ensureMasqRule(); err!=nil {
returnerr
}
returnnil
}
// TODO: move thic logic into cni bridge plugin and remove this from kubenetfunc (plugin*kubenetNetworkPlugin) ensureMasqRule() error {
ifplugin.nonMasqueradeCIDR!=zeroCIDRv4&&plugin.nonMasqueradeCIDR!=zeroCIDRv6 {
// switch according to target nonMasqueradeCidr ip familyipt:=plugin.iptablesifnetutils.IsIPv6CIDRString(plugin.nonMasqueradeCIDR) {
ipt=plugin.iptablesv6
}
if_, err:=ipt.EnsureRule(utiliptables.Append, utiliptables.TableNAT, utiliptables.ChainPostrouting,
"-m", "comment", "--comment", "kubenet: SNAT for outbound traffic from cluster",
"-m", "addrtype", "!", "--dst-type", "LOCAL",
"!", "-d", plugin.nonMasqueradeCIDR,
"-j", "MASQUERADE"); err!=nil {
returnfmt.Errorf("failed to ensure that %s chain %s jumps to MASQUERADE: %v", utiliptables.TableNAT, utiliptables.ChainPostrouting, err)
}
}
returnnil
}
funcfindMinMTU() (*net.Interface, error) {
intfs, err:=net.Interfaces()
iferr!=nil {
returnnil, err
}
mtu:=999999defIntfIndex:=-1fori, intf:=rangeintfs {
if ((intf.Flags&net.FlagUp) !=0) && (intf.Flags&(net.FlagLoopback|net.FlagPointToPoint) ==0) {
ifintf.MTU<mtu {
mtu=intf.MTUdefIntfIndex=i
}
}
}
ifmtu>=999999||mtu<576||defIntfIndex<0 {
returnnil, fmt.Errorf("no suitable interface: %v", BridgeName)
}
return&intfs[defIntfIndex], nil
}
func (plugin*kubenetNetworkPlugin) Event(namestring, detailsmap[string]interface{}) {
varerrerrorifname!=network.NET_PLUGIN_EVENT_POD_CIDR_CHANGE {
return
}
plugin.mu.Lock()
deferplugin.mu.Unlock()
podCIDR, ok:=details[network.NET_PLUGIN_EVENT_POD_CIDR_CHANGE_DETAIL_CIDR].(string)
if!ok {
klog.Warningf("%s event didn't contain pod CIDR", network.NET_PLUGIN_EVENT_POD_CIDR_CHANGE)
return
}
ifplugin.netConfig!=nil {
klog.Warningf("Ignoring subsequent pod CIDR update to %s", podCIDR)
return
}
klog.V(4).Infof("kubenet: PodCIDR is set to %q", podCIDR)
podCIDRs:=strings.Split(podCIDR, ",")
// reset to one cidr if dual stack is not enabledif!utilfeature.DefaultFeatureGate.Enabled(kubefeatures.IPv6DualStack) &&len(podCIDRs) >1 {
klog.V(2).Infof("This node has multiple pod cidrs assigned and dual stack is not enabled. ignoring all except first cidr")
podCIDRs=podCIDRs[0:1]
}
foridx, currentPodCIDR:=rangepodCIDRs {
_, cidr, err:=net.ParseCIDR(currentPodCIDR)
ifnil!=err {
klog.Warningf("Failed to generate CNI network config with cidr %s at index:%v: %v", currentPodCIDR, idx, err)
return
}
// create list of ipsplugin.podCIDRs=append(plugin.podCIDRs, cidr)
}
//setup hairpinModesetHairpin:=plugin.hairpinMode==kubeletconfig.HairpinVethjson:=fmt.Sprintf(NET_CONFIG_TEMPLATE, BridgeName, plugin.mtu, network.DefaultInterfaceName, setHairpin, plugin.getRangesConfig(), plugin.getRoutesConfig())
klog.V(4).Infof("CNI network config set to %v", json)
plugin.netConfig, err=libcni.ConfFromBytes([]byte(json))
iferr!=nil {
klog.Warningf("** failed to set up CNI with %v err:%v", json, err)
// just incase it was set by mistakeplugin.netConfig=nil// we bail out by clearing the *entire* list// of addresses assigned to cbr0plugin.clearUnusedBridgeAddresses()
}
}
// clear all address on bridge except those operated on by kubenetfunc (plugin*kubenetNetworkPlugin) clearUnusedBridgeAddresses() {
cidrIncluded:=func(list []*net.IPNet, check*net.IPNet) bool {
for_, thisNet:=rangelist {
ifutilnet.IPNetEqual(thisNet, check) {
returntrue
}
}
returnfalse
}
bridge, err:=netlink.LinkByName(BridgeName)
iferr!=nil {
return
}
addrs, err:=netlink.AddrList(bridge, unix.AF_INET)
iferr!=nil {
klog.V(2).Infof("attempting to get address for interface: %s failed with err:%v", BridgeName, err)
return
}
for_, addr:=rangeaddrs {
if!cidrIncluded(plugin.podCIDRs, addr.IPNet) {
klog.V(2).Infof("Removing old address %s from %s", addr.IPNet.String(), BridgeName)
netlink.AddrDel(bridge, &addr)
}
}
}
func (plugin*kubenetNetworkPlugin) Name() string {
returnKubenetPluginName
}
func (plugin*kubenetNetworkPlugin) Capabilities() utilsets.Int {
returnutilsets.NewInt()
}
// setup sets up networking through CNI using the given ns/name and sandbox ID.func (plugin*kubenetNetworkPlugin) setup(namespacestring, namestring, id kubecontainer.ContainerID, annotationsmap[string]string) error {
varipv4, ipv6 net.IPvarpodGateways []net.IPvarpodCIDRs []net.IPNet// Disable DAD so we skip the kernel delay on bringing up new interfaces.iferr:=plugin.disableContainerDAD(id); err!=nil {
klog.V(3).Infof("Failed to disable DAD in container: %v", err)
}
// Bring up container loopback interfaceif_, err:=plugin.addContainerToNetwork(plugin.loConfig, "lo", namespace, name, id); err!=nil {
returnerr
}
// Hook container up with our bridgeresT, err:=plugin.addContainerToNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id)
iferr!=nil {
returnerr
}
// Coerce the CNI result versionres, err:=cnitypes020.GetResult(resT)
iferr!=nil {
returnfmt.Errorf("unable to understand network config: %v", err)
}
//TODO: v1.16 (khenidak) update NET_CONFIG_TEMPLATE to CNI version 0.3.0 or later so// that we get multiple IP addresses in the returned Result structureifres.IP4!=nil {
ipv4=res.IP4.IP.IP.To4()
podGateways=append(podGateways, res.IP4.Gateway)
podCIDRs=append(podCIDRs, net.IPNet{IP: ipv4.Mask(res.IP4.IP.Mask), Mask: res.IP4.IP.Mask})
}
ifres.IP6!=nil {
ipv6=res.IP6.IP.IPpodGateways=append(podGateways, res.IP6.Gateway)
podCIDRs=append(podCIDRs, net.IPNet{IP: ipv6.Mask(res.IP6.IP.Mask), Mask: res.IP6.IP.Mask})
}
ifipv4==nil&&ipv6==nil {
returnfmt.Errorf("cni didn't report ipv4 ipv6")
}
// Put the container bridge into promiscuous mode to force it to accept hairpin packets.// TODO: Remove this once the kernel bug (#20096) is fixed.ifplugin.hairpinMode==kubeletconfig.PromiscuousBridge {
link, err:=netlink.LinkByName(BridgeName)
iferr!=nil {
returnfmt.Errorf("failed to lookup %q: %v", BridgeName, err)
}
iflink.Attrs().Promisc!=1 {
// promiscuous mode is not on, then turn it on.err:=netlink.SetPromiscOn(link)
iferr!=nil {
returnfmt.Errorf("error setting promiscuous mode on %s: %v", BridgeName, err)
}
}
// configure the ebtables rules to eliminate duplicate packets by best effortplugin.syncEbtablesDedupRules(link.Attrs().HardwareAddr, podCIDRs, podGateways)
}
// add the ip to tracked ipsifipv4!=nil {
plugin.addPodIP(id, ipv4.String())
}
ifipv6!=nil {
plugin.addPodIP(id, ipv6.String())
}
iferr:=plugin.addTrafficShaping(id, annotations); err!=nil {
returnerr
}
returnplugin.addPortMapping(id, name, namespace)
}
// The first SetUpPod call creates the bridge; get a shaper for the sake of initialization// TODO: replace with CNI traffic shaper pluginfunc (plugin*kubenetNetworkPlugin) addTrafficShaping(id kubecontainer.ContainerID, annotationsmap[string]string) error {
shaper:=plugin.shaper()
ingress, egress, err:=bandwidth.ExtractPodBandwidthResources(annotations)
iferr!=nil {
returnfmt.Errorf("error reading pod bandwidth annotations: %v", err)
}
iplist, exists:=plugin.getCachedPodIPs(id)
if!exists {
returnfmt.Errorf("pod %s does not have recorded ips", id)
}
ifegress!=nil||ingress!=nil {
for_, ip:=rangeiplist {
mask:=32ifnetutils.IsIPv6String(ip) {
mask=128
}
iferr!=nil {
returnfmt.Errorf("failed to setup traffic shaping for pod ip%s", ip)
}
iferr:=shaper.ReconcileCIDR(fmt.Sprintf("%v/%v", ip, mask), egress, ingress); err!=nil {
returnfmt.Errorf("failed to add pod to shaper: %v", err)
}
}
}
returnnil
}
// TODO: replace with CNI port-forwarding pluginfunc (plugin*kubenetNetworkPlugin) addPortMapping(id kubecontainer.ContainerID, name, namespacestring) error {
portMappings, err:=plugin.host.GetPodPortMappings(id.ID)
iferr!=nil {
returnerr
}
iflen(portMappings) ==0 {
returnnil
}
iplist, exists:=plugin.getCachedPodIPs(id)
if!exists {
returnfmt.Errorf("pod %s does not have recorded ips", id)
}
for_, ip:=rangeiplist {
pm:=&hostport.PodPortMapping{
Namespace: namespace,
Name: name,
PortMappings: portMappings,
IP: net.ParseIP(ip),
HostNetwork: false,
}
ifnetutils.IsIPv6(pm.IP) {
iferr:=plugin.hostportManagerv6.Add(id.ID, pm, BridgeName); err!=nil {
returnerr
}
} else {
iferr:=plugin.hostportManager.Add(id.ID, pm, BridgeName); err!=nil {
returnerr
}
}
}
returnnil
}
func (plugin*kubenetNetworkPlugin) SetUpPod(namespacestring, namestring, id kubecontainer.ContainerID, annotations, optionsmap[string]string) error {
start:=time.Now()
iferr:=plugin.Status(); err!=nil {
returnfmt.Errorf("kubenet cannot SetUpPod: %v", err)
}
deferfunc() {
klog.V(4).Infof("SetUpPod took %v for %s/%s", time.Since(start), namespace, name)
}()
iferr:=plugin.setup(namespace, name, id, annotations); err!=nil {
iferr:=plugin.teardown(namespace, name, id); err!=nil {
// Not a hard error or warningklog.V(4).Infof("Failed to clean up %s/%s after SetUpPod failure: %v", namespace, name, err)
}
returnerr
}
// Need to SNAT outbound traffic from clusteriferr:=plugin.ensureMasqRule(); err!=nil {
klog.Errorf("Failed to ensure MASQ rule: %v", err)
}
returnnil
}
// Tears down as much of a pod's network as it can even if errors occur. Returns// an aggregate error composed of all errors encountered during the teardown.func (plugin*kubenetNetworkPlugin) teardown(namespacestring, namestring, id kubecontainer.ContainerID) error {
errList:= []error{}
// Loopback network deletion failure should not be fatal on teardowniferr:=plugin.delContainerFromNetwork(plugin.loConfig, "lo", namespace, name, id); err!=nil {
klog.Warningf("Failed to delete loopback network: %v", err)
errList=append(errList, err)
}
// no ip dependent actionsiferr:=plugin.delContainerFromNetwork(plugin.netConfig, network.DefaultInterfaceName, namespace, name, id); err!=nil {
klog.Warningf("Failed to delete %q network: %v", network.DefaultInterfaceName, err)
errList=append(errList, err)
}
// If there are no IPs registered we can't teardown pod's IP dependenciesiplist, exists:=plugin.getCachedPodIPs(id)
if!exists||len(iplist) ==0 {
klog.V(5).Infof("container %s (%s/%s) does not have recorded. ignoring teardown call", id, name, namespace)
returnnil
}
// get the list of port mappingsportMappings, err:=plugin.host.GetPodPortMappings(id.ID)
iferr!=nil {
errList=append(errList, err)
}
// process each pod IPfor_, ip:=rangeiplist {
isV6:=netutils.IsIPv6String(ip)
klog.V(5).Infof("Removing pod port mappings from IP %s", ip)
ifportMappings!=nil&&len(portMappings) >0 {
ifisV6 {
iferr=plugin.hostportManagerv6.Remove(id.ID, &hostport.PodPortMapping{
Namespace: namespace,
Name: name,
PortMappings: portMappings,
HostNetwork: false,
}); err!=nil {
errList=append(errList, err)
}
} else {
iferr=plugin.hostportManager.Remove(id.ID, &hostport.PodPortMapping{
Namespace: namespace,
Name: name,
PortMappings: portMappings,
HostNetwork: false,
}); err!=nil {
errList=append(errList, err)
}
}
}
klog.V(5).Infof("Removing pod IP %s from shaper for (%s/%s)", ip, name, namespace)
// shaper uses a cidr, but we are using a single IP.mask:="32"ifisV6 {
mask="128"
}
iferr:=plugin.shaper().Reset(fmt.Sprintf("%s/%s", ip, mask)); err!=nil {
// Possible bandwidth shaping wasn't enabled for this pod anywaysklog.V(4).Infof("Failed to remove pod IP %s from shaper: %v", ip, err)
}
plugin.removePodIP(id, ip)
}
returnutilerrors.NewAggregate(errList)
}
func (plugin*kubenetNetworkPlugin) TearDownPod(namespacestring, namestring, id kubecontainer.ContainerID) error {
start:=time.Now()
deferfunc() {
klog.V(4).Infof("TearDownPod took %v for %s/%s", time.Since(start), namespace, name)
}()
ifplugin.netConfig==nil {
returnfmt.Errorf("kubenet needs a PodCIDR to tear down pods")
}
iferr:=plugin.teardown(namespace, name, id); err!=nil {
returnerr
}
// Need to SNAT outbound traffic from clusteriferr:=plugin.ensureMasqRule(); err!=nil {
klog.Errorf("Failed to ensure MASQ rule: %v", err)
}
returnnil
}
// TODO: Use the addToNetwork function to obtain the IP of the Pod. That will assume idempotent ADD call to the plugin.// Also fix the runtime's call to Status function to be done only in the case that the IP is lost, no need to do periodic callsfunc (plugin*kubenetNetworkPlugin) GetPodNetworkStatus(namespacestring, namestring, id kubecontainer.ContainerID) (*network.PodNetworkStatus, error) {
// try cached versionnetworkStatus:=plugin.getNetworkStatus(id)
ifnetworkStatus!=nil {
returnnetworkStatus, nil
}
// not a cached version, get via network nsnetnsPath, err:=plugin.host.GetNetNS(id.ID)
iferr!=nil {
returnnil, fmt.Errorf("kubenet failed to retrieve network namespace path: %v", err)
}
ifnetnsPath=="" {
returnnil, fmt.Errorf("cannot find the network namespace, skipping pod network status for container %q", id)
}
ips, err:=network.GetPodIPs(plugin.execer, plugin.nsenterPath, netnsPath, network.DefaultInterfaceName)
iferr!=nil {
returnnil, err
}
// cache the ipsfor_, ip:=rangeips {
plugin.addPodIP(id, ip.String())
}
// return from cachedreturnplugin.getNetworkStatus(id), nil
}
// returns networkstatusfunc (plugin*kubenetNetworkPlugin) getNetworkStatus(id kubecontainer.ContainerID) *network.PodNetworkStatus {
// Assuming the ip of pod does not change. Try to retrieve ip from kubenet map first.iplist, ok:=plugin.getCachedPodIPs(id)
if!ok {
returnnil
}
iflen(iplist) ==0 {
returnnil
}
ips:=make([]net.IP, 0, len(iplist))
for_, ip:=rangeiplist {
ips=append(ips, net.ParseIP(ip))
}
return&network.PodNetworkStatus{
IP: ips[0],
IPs: ips,
}
}
func (plugin*kubenetNetworkPlugin) Status() error {
// Can't set up pods if we don't have a PodCIDR yetifplugin.netConfig==nil {
returnfmt.Errorf("kubenet does not have netConfig. This is most likely due to lack of PodCIDR")
}
if!plugin.checkRequiredCNIPlugins() {
returnfmt.Errorf("could not locate kubenet required CNI plugins %v at %q", requiredCNIPlugins, plugin.binDirs)
}
returnnil
}
// checkRequiredCNIPlugins returns if all kubenet required cni plugins can be found at /opt/cni/bin or user specified NetworkPluginDir.func (plugin*kubenetNetworkPlugin) checkRequiredCNIPlugins() bool {
for_, dir:=rangeplugin.binDirs {
ifplugin.checkRequiredCNIPluginsInOneDir(dir) {
returntrue
}
}
returnfalse
}
// checkRequiredCNIPluginsInOneDir returns true if all required cni plugins are placed in dirfunc (plugin*kubenetNetworkPlugin) checkRequiredCNIPluginsInOneDir(dirstring) bool {
files, err:=ioutil.ReadDir(dir)
iferr!=nil {
returnfalse
}
for_, cniPlugin:=rangerequiredCNIPlugins {
found:=falsefor_, file:=rangefiles {
ifstrings.TrimSpace(file.Name()) ==cniPlugin {
found=truebreak
}
}
if!found {
returnfalse
}
}
returntrue
}
func (plugin*kubenetNetworkPlugin) buildCNIRuntimeConf(ifNamestring, id kubecontainer.ContainerID, needNetNsbool) (*libcni.RuntimeConf, error) {
netnsPath, err:=plugin.host.GetNetNS(id.ID)
ifneedNetNs&&err!=nil {
klog.Errorf("Kubenet failed to retrieve network namespace path: %v", err)
}
return&libcni.RuntimeConf{
ContainerID: id.ID,
NetNS: netnsPath,
IfName: ifName,
CacheDir: plugin.cacheDir,
}, nil
}
func (plugin*kubenetNetworkPlugin) addContainerToNetwork(config*libcni.NetworkConfig, ifName, namespace, namestring, id kubecontainer.ContainerID) (cnitypes.Result, error) {
rt, err:=plugin.buildCNIRuntimeConf(ifName, id, true)
iferr!=nil {
returnnil, fmt.Errorf("error building CNI config: %v", err)
}
klog.V(3).Infof("Adding %s/%s to '%s' with CNI '%s' plugin and runtime: %+v", namespace, name, config.Network.Name, config.Network.Type, rt)
// Because the default remote runtime request timeout is 4 min,so set slightly less than 240 seconds// Todo get the timeout from parent ctxcniTimeoutCtx, cancelFunc:=context.WithTimeout(context.Background(), network.CNITimeoutSec*time.Second)
defercancelFunc()
res, err:=plugin.cniConfig.AddNetwork(cniTimeoutCtx, config, rt)
iferr!=nil {
returnnil, fmt.Errorf("error adding container to network: %v", err)
}
returnres, nil
}
func (plugin*kubenetNetworkPlugin) delContainerFromNetwork(config*libcni.NetworkConfig, ifName, namespace, namestring, id kubecontainer.ContainerID) error {
rt, err:=plugin.buildCNIRuntimeConf(ifName, id, false)
iferr!=nil {
returnfmt.Errorf("error building CNI config: %v", err)
}
klog.V(3).Infof("Removing %s/%s from '%s' with CNI '%s' plugin and runtime: %+v", namespace, name, config.Network.Name, config.Network.Type, rt)
// Because the default remote runtime request timeout is 4 min,so set slightly less than 240 seconds// Todo get the timeout from parent ctxcniTimeoutCtx, cancelFunc:=context.WithTimeout(context.Background(), network.CNITimeoutSec*time.Second)
defercancelFunc()
err=plugin.cniConfig.DelNetwork(cniTimeoutCtx, config, rt)
// The pod may not get deleted successfully at the first time.// Ignore "no such file or directory" error in case the network has already been deleted in previous attempts.iferr!=nil&&!strings.Contains(err.Error(), "no such file or directory") {
returnfmt.Errorf("error removing container from network: %v", err)
}
returnnil
}
// shaper retrieves the bandwidth shaper and, if it hasn't been fetched before,// initializes it and ensures the bridge is appropriately configuredfunc (plugin*kubenetNetworkPlugin) shaper() bandwidth.Shaper {
plugin.mu.Lock()
deferplugin.mu.Unlock()
ifplugin.bandwidthShaper==nil {
plugin.bandwidthShaper=bandwidth.NewTCShaper(BridgeName)
plugin.bandwidthShaper.ReconcileInterface()
}
returnplugin.bandwidthShaper
}
//TODO: make this into a goroutine and rectify the dedup rules periodicallyfunc (plugin*kubenetNetworkPlugin) syncEbtablesDedupRules(macAddr net.HardwareAddr, podCIDRs []net.IPNet, podGateways []net.IP) {
ifplugin.ebtables==nil {
plugin.ebtables=utilebtables.New(plugin.execer)
klog.V(3).Infof("Flushing dedup chain")
iferr:=plugin.ebtables.FlushChain(utilebtables.TableFilter, dedupChain); err!=nil {
klog.Errorf("Failed to flush dedup chain: %v", err)
}
}
_, err:=plugin.ebtables.GetVersion()
iferr!=nil {
klog.Warningf("Failed to get ebtables version. Skip syncing ebtables dedup rules: %v", err)
return
}
// ensure custom chain exists_, err=plugin.ebtables.EnsureChain(utilebtables.TableFilter, dedupChain)
iferr!=nil {
klog.Errorf("Failed to ensure %v chain %v", utilebtables.TableFilter, dedupChain)
return
}
// jump to custom chain to the chain from core tables_, err=plugin.ebtables.EnsureRule(utilebtables.Append, utilebtables.TableFilter, utilebtables.ChainOutput, "-j", string(dedupChain))
iferr!=nil {
klog.Errorf("Failed to ensure %v chain %v jump to %v chain: %v", utilebtables.TableFilter, utilebtables.ChainOutput, dedupChain, err)
return
}
// per gateway ruleforidx, gw:=rangepodGateways {
klog.V(3).Infof("Filtering packets with ebtables on mac address: %v, gateway: %v, pod CIDR: %v", macAddr.String(), gw.String(), podCIDRs[idx].String())
bIsV6:=netutils.IsIPv6(gw)
IPFamily:="IPv4"ipSrc:="--ip-src"ifbIsV6 {
IPFamily="IPv6"ipSrc="--ip6-src"
}
commonArgs:= []string{"-p", IPFamily, "-s", macAddr.String(), "-o", "veth+"}
_, err=plugin.ebtables.EnsureRule(utilebtables.Prepend, utilebtables.TableFilter, dedupChain, append(commonArgs, ipSrc, gw.String(), "-j", "ACCEPT")...)
iferr!=nil {
klog.Errorf("Failed to ensure packets from cbr0 gateway:%v to be accepted with error:%v", gw.String(), err)
return
}
_, err=plugin.ebtables.EnsureRule(utilebtables.Append, utilebtables.TableFilter, dedupChain, append(commonArgs, ipSrc, podCIDRs[idx].String(), "-j", "DROP")...)
iferr!=nil {
klog.Errorf("Failed to ensure packets from podCidr[%v] but has mac address of cbr0 to get dropped. err:%v", podCIDRs[idx].String(), err)
return
}
}
}
// disableContainerDAD disables duplicate address detection in the container.// DAD has a negative affect on pod creation latency, since we have to wait// a second or more for the addresses to leave the "tentative" state. Since// we're sure there won't be an address conflict (since we manage them manually),// this is safe. See issue 54651.//// This sets net.ipv6.conf.default.dad_transmits to 0. It must be run *before*// the CNI plugins are run.func (plugin*kubenetNetworkPlugin) disableContainerDAD(id kubecontainer.ContainerID) error {
key:="net/ipv6/conf/default/dad_transmits"sysctlBin, err:=plugin.execer.LookPath("sysctl")
iferr!=nil {
returnfmt.Errorf("could not find sysctl binary: %s", err)
}
netnsPath, err:=plugin.host.GetNetNS(id.ID)
iferr!=nil {
returnfmt.Errorf("failed to get netns: %v", err)
}
ifnetnsPath=="" {
returnfmt.Errorf("pod has no network namespace")
}
// If the sysctl doesn't exist, it means ipv6 is disabled; log and move onif_, err:=plugin.sysctl.GetSysctl(key); err!=nil {
returnfmt.Errorf("ipv6 not enabled: %v", err)
}
output, err:=plugin.execer.Command(plugin.nsenterPath,
fmt.Sprintf("--net=%s", netnsPath), "-F", "--",
sysctlBin, "-w", fmt.Sprintf("%s=%s", key, "0"),
).CombinedOutput()
iferr!=nil {
returnfmt.Errorf("failed to write sysctl: output: %s error: %s",
output, err)
}
returnnil
}
// given a n cidrs assigned to nodes,// create bridge configuration that conforms to themfunc (plugin*kubenetNetworkPlugin) getRangesConfig() string {
createRange:=func(thisNet*net.IPNet) string {
template:=`[{"subnet": "%s"}]`returnfmt.Sprintf(template, thisNet.String())
}
ranges:=make([]string, len(plugin.podCIDRs))
foridx, thisCIDR:=rangeplugin.podCIDRs {
ranges[idx] =createRange(thisCIDR)
}
//[{range}], [{range}]// each range contains a subnet. gateway will be fetched from cni resultreturnstrings.Join(ranges[:], ",")
}
// given a n cidrs assigned to nodes,// create bridge routes configuration that conforms to themfunc (plugin*kubenetNetworkPlugin) getRoutesConfig() string {
var (
routes []stringhasV4, hasV6bool
)
for_, thisCIDR:=rangeplugin.podCIDRs {
ifthisCIDR.IP.To4() !=nil {
hasV4=true
} else {
hasV6=true
}
}
ifhasV4 {
routes=append(routes, fmt.Sprintf(`{"dst": "%s"}`, zeroCIDRv4))
}
ifhasV6 {
routes=append(routes, fmt.Sprintf(`{"dst": "%s"}`, zeroCIDRv6))
}
returnstrings.Join(routes, ",")
}
func (plugin*kubenetNetworkPlugin) addPodIP(id kubecontainer.ContainerID, ipstring) {
plugin.mu.Lock()
deferplugin.mu.Unlock()
_, exist:=plugin.podIPs[id]
if!exist {
plugin.podIPs[id] =utilsets.NewString()
}
if!plugin.podIPs[id].Has(ip) {
plugin.podIPs[id].Insert(ip)
}
}
func (plugin*kubenetNetworkPlugin) removePodIP(id kubecontainer.ContainerID, ipstring) {
plugin.mu.Lock()
deferplugin.mu.Unlock()
_, exist:=plugin.podIPs[id]
if!exist {
return// did we restart kubelet?
}
ifplugin.podIPs[id].Has(ip) {
plugin.podIPs[id].Delete(ip)
}
// if there is no more ips here. let us deleteifplugin.podIPs[id].Len() ==0 {
delete(plugin.podIPs, id)
}
}
// returns a copy of pod ips// false is returned if id does not existfunc (plugin*kubenetNetworkPlugin) getCachedPodIPs(id kubecontainer.ContainerID) ([]string, bool) {
plugin.mu.Lock()
deferplugin.mu.Unlock()
iplist, exists:=plugin.podIPs[id]
if!exists {
returnnil, false
}
returniplist.UnsortedList(), true
}
ewfilemode100644ndex0000000000000..460e78960c9d7++b/pkg/kubelet/dockershim/network/kubenet/kubenet_linux_test.go
5b2b58f81beee099ced19e2786aab0a839552343
The text was updated successfully, but these errors were encountered:
[Remove this once the kernel bug (kubernetes#20096] is fixed.
kubernetes/pkg/kubelet/dockershim/network/kubenet/kubenet_linux.go
Line 369 in dcda038
5b2b58f81beee099ced19e2786aab0a839552343
The text was updated successfully, but these errors were encountered: