4318

Committed 03 Mar 2021 05:19PM UTC coverage: 57.969% (-0.2%) from 58.164%

Build # 4318

Build Type

push

travis-ci-com

Committed by

tanyatukade

Commit Message

<a href="https://github.com/noironetworks/aci-containers/commit/<a class=hub.com/noironetworks/aci-containers/commit/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171">7c463bb0b">&lt;a href=&quot;https://github.com/noironetworks/aci-containers/commit/</a><a class="double-link" href="https://github.com/noironetworks/aci-containers/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/noironetworks/aci-containers/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/noironetworks/aci-containers/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/noironetworks/aci-containers/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171">7c463bb0b</a><a href="https://github.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171">&lt;a href=&quot;https://github.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/noironetworks/aci-containers/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171&amp;quot;&amp;gt;7c463bb0b&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/noironetworks/aci-containers/commit/7c463bb0bdeb5e58cede92fd0eb290c83dfd1171&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;quot;&amp;amp;amp;gt;Revert &amp;amp;amp;amp;quot;Add erspan sync with nod&amp;lt;/a&amp;gt;es&amp;a</a>mp;amp;quot;

This reverts com... (continued)

Run Details

10980 of 18941 relevant lines covered (57.97%)

0.65 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.54

/pkg/controller/nodes.go

// Copyright 2017 Cisco Systems, Inc.
//
// 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.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed 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 and
// limitations under the License.

// Handlers for node updates.

package controller

import (
        "crypto/rand"
        "encoding/json"
        "errors"
        "fmt"
        "net"
        "net/http"

        v1 "k8s.io/api/core/v1"
        kubeerr "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/tools/cache"

        "github.com/sirupsen/logrus"

        "github.com/noironetworks/aci-containers/pkg/apicapi"
        crdv1 "github.com/noironetworks/aci-containers/pkg/gbpcrd/apis/acipolicy/v1"
        "github.com/noironetworks/aci-containers/pkg/gbpserver/watchers"
        "github.com/noironetworks/aci-containers/pkg/ipam"
        "github.com/noironetworks/aci-containers/pkg/metadata"
        "github.com/noironetworks/aci-containers/pkg/util"
)

const (
        tunnelIDIncr = 2
)

type tunnelState struct {
        stateDriver  *watchers.K8sStateDriver
        nodeToTunnel map[string]int64
        nextID       int64
}

func (cont *AciController) initNodeInformerFromClient(
        kubeClient *kubernetes.Clientset) {

        cont.initNodeInformerBase(
                cache.NewListWatchFromClient(
                        kubeClient.CoreV1().RESTClient(), "nodes",
                        metav1.NamespaceAll, fields.Everything()))
}

func (cont *AciController) initNodeInformerBase(listWatch *cache.ListWatch) {
        cont.nodeIndexer, cont.nodeInformer = cache.NewIndexerInformer(
                listWatch, &v1.Node{}, 0,
                cache.ResourceEventHandlerFuncs{
                        AddFunc: func(obj interface{}) {
                                cont.syncPodNet(obj) // update cache
                                cont.nodeChanged(obj)
                        },
                        UpdateFunc: func(_ interface{}, obj interface{}) {
                                cont.nodeChanged(obj)
                        },
                        DeleteFunc: func(obj interface{}) {
                                cont.nodeDeleted(obj)
                        },
                },
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
}

func apicNodeNetPol(name string, tenantName string,
        nodeIps map[string]bool) apicapi.ApicObject {

        hpp := apicapi.NewHostprotPol(tenantName, name)
        hppDn := hpp.GetDn()
        nodeSubj := apicapi.NewHostprotSubj(hppDn, "local-node")
        if len(nodeIps) > 0 {
                nodeSubjDn := nodeSubj.GetDn()
                outbound := apicapi.NewHostprotRule(nodeSubjDn, "allow-all-egress")
                outbound.SetAttr("direction", "egress")
                outbound.SetAttr("ethertype", "ipv4")
                outbound.SetAttr("connTrack", "normal")

                inbound := apicapi.NewHostprotRule(nodeSubjDn, "allow-all-ingress")
                inbound.SetAttr("direction", "ingress")
                inbound.SetAttr("ethertype", "ipv4")
                inbound.SetAttr("connTrack", "normal")

                for ip := range nodeIps {
                        outbound.AddChild(apicapi.NewHostprotRemoteIp(outbound.GetDn(), ip))
                        inbound.AddChild(apicapi.NewHostprotRemoteIp(inbound.GetDn(), ip))
                }

                nodeSubj.AddChild(inbound)
                nodeSubj.AddChild(outbound)
        }
        hpp.AddChild(nodeSubj)
        return hpp
}

func (cont *AciController) createNetPolForNode(node *v1.Node) {
        nodeIps := make(map[string]bool)
        for _, a := range node.Status.Addresses {
                if a.Address != "" &&
                        (a.Type == "InternalIP" || a.Type == "ExternalIP") {
                        nodeIps[a.Address] = true
                }
        }

        sgName := cont.aciNameForKey("node", node.Name)
        cont.apicConn.WriteApicObjects(sgName,
                apicapi.ApicSlice{
                        apicNodeNetPol(sgName, cont.config.AciPolicyTenant, nodeIps),
                })
}

func (cont *AciController) createServiceEndpoint(existing *metadata.ServiceEndpoint, ep *metadata.ServiceEndpoint, deviceMac string, nodeName string) error {

        _, err := net.ParseMAC(deviceMac)
        if err == nil && deviceMac != "00:00:00:00:00:00" {
                ep.Mac = deviceMac
        } else {
                _, err := net.ParseMAC(existing.Mac)
                if err == nil {
                        ep.Mac = existing.Mac
                } else {
                        var mac net.HardwareAddr
                        mac = make([]byte, 6)
                        _, err := rand.Read(mac)
                        if err != nil {
                                return err
                        }

                        mac[0] = (mac[0] & 254) | 2
                        ep.Mac = mac.String()
                }
        }

        if ep.Ipv4 == nil && existing.Ipv4 != nil &&
                cont.nodeServiceIps.V4.RemoveIp(existing.Ipv4) {
                ep.Ipv4 = existing.Ipv4
        }

        if ep.Ipv4 == nil {
                ipv4, err := cont.nodeServiceIps.V4.GetIp()
                if err == nil {
                        ep.Ipv4 = ipv4
                } else {
                        ep.Ipv4 = nil
                }
        }

        if ep.Ipv6 == nil && existing.Ipv6 != nil &&
                cont.nodeServiceIps.V6.RemoveIp(existing.Ipv6) {
                ep.Ipv6 = existing.Ipv6
        }

        if ep.Ipv6 == nil {
                ipv6, err := cont.nodeServiceIps.V6.GetIp()
                if err == nil {
                        ep.Ipv6 = ipv6
                } else {
                        ep.Ipv6 = nil
                }
        }

        if ep.Ipv4 == nil && ep.Ipv6 == nil {
                return errors.New("No IP addresses available for service endpoint")
        }

        if (ep.HealthGroupDn == "") && (cont.config.AciServiceMonitorInterval > 0) {
                name := cont.aciNameForKey("svc", nodeName)
                healthGroupObj := apicapi.NewVnsRedirectHealthGroup(cont.config.AciVrfTenant, name)
                ep.HealthGroupDn = healthGroupObj.GetDn()
                cont.apicConn.WriteApicObjects(name, apicapi.ApicSlice{healthGroupObj})
        }

        return nil
}

func (cont *AciController) nodeFullSync() {
        cache.ListAll(cont.nodeIndexer, labels.Everything(),
                func(nodeobj interface{}) {
                        cont.syncPodNet(nodeobj) // update the cache
                        cont.nodeChanged(nodeobj)
                })
}

// syncPodNet syncs in the podnets from the node object's annotation
func (cont *AciController) syncPodNet(obj interface{}) {
        cont.indexMutex.Lock()
        defer cont.indexMutex.Unlock()

        node := obj.(*v1.Node)
        logger := cont.log.WithFields(logrus.Fields{
                "Node": node.ObjectMeta.Name,
        })

        if node.ObjectMeta.Annotations == nil {
                return // nothing to sync
        }

        netval, ok := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]
        if !ok {
                return // nothing to sync
        }

        nodePodNet := newNodePodNetMeta()
        cont.mergePodNet(nodePodNet, netval, logger)
        cont.nodePodNetCache[node.ObjectMeta.Name] = nodePodNet
}

func maxof(x1, x2 int64) int64 {
        if x1 > x2 {
                return x1
        }

        return x2
}

func (cont *AciController) getTunnelID(node *v1.Node) int64 {
        if cont.config.LBType == lbTypeAci {
                return 0
        }

        nodeIP := getNodeIP(node, v1.NodeInternalIP)
        if nodeIP == "" {
                cont.log.Errorf("Can't get IP for node %s", node.Name)
                return 0
        }

        if cont.tunnelGetter == nil {
                tunnelGetter := &tunnelState{}
                tunnelGetter.stateDriver = &watchers.K8sStateDriver{}
                err := tunnelGetter.stateDriver.Init(watchers.FieldTunnelID)
                if err != nil {
                        cont.log.Warnf("Can't get tunnelID for %s", node.Name)
                        return 0
                }

                gbps, err := tunnelGetter.stateDriver.Get()
                if err != nil {
                        cont.log.Warnf("Can't get tunnelID for %s", node.Name)
                        return 0
                }

                cont.tunnelGetter = tunnelGetter
                if gbps.Status.TunnelIDs == nil {
                        cont.log.Infof("Initializing tunnelIDs")
                        tunnelGetter.nodeToTunnel = make(map[string]int64)
                        cont.tunnelGetter.nextID = 0
                } else {
                        tunnelGetter.nodeToTunnel = gbps.Status.TunnelIDs
                        for _, val := range tunnelGetter.nodeToTunnel {
                                cont.tunnelGetter.nextID = maxof(cont.tunnelGetter.nextID, val)
                        }

                        cont.tunnelGetter.nextID = cont.tunnelGetter.nextID + tunnelIDIncr
                }
        }

        id, found := cont.tunnelGetter.nodeToTunnel[nodeIP]
        if found {
                return id + int64(cont.config.CSRTunnelIDBase)
        }

        id = cont.tunnelGetter.nextID
        if id >= int64(cont.config.MaxCSRTunnels*tunnelIDIncr) {
                cont.log.Infof("Max tunnels %d reached", cont.config.MaxCSRTunnels)
                return 0
        }

        // each allocated id packs two tunnels.
        cont.tunnelGetter.nodeToTunnel[nodeIP] = id
        cont.tunnelGetter.nextID = id + tunnelIDIncr
        state := &crdv1.GBPSState{}
        state.Status.TunnelIDs = cont.tunnelGetter.nodeToTunnel
        cont.tunnelGetter.stateDriver.Update(state)
        return id + int64(cont.config.CSRTunnelIDBase)
}

func (cont *AciController) writeApicNode(node *v1.Node) {
        tunnelID := cont.getTunnelID(node)
        //cont.log.Infof("=>  Node: %s, tunnelID: %v", node.Name, tunnelID)
        key := cont.aciNameForKey("node-vmm", node.Name)
        aobj := apicapi.NewVmmInjectedHost(cont.vmmDomainProvider(),
                cont.config.AciVmmDomain, cont.config.AciVmmController,
                node.Name)
        aobj.SetAttr("mgmtIp", getNodeIP(node, v1.NodeInternalIP))
        aobj.SetAttr("os", node.Status.NodeInfo.OSImage)
        aobj.SetAttr("kernelVer", node.Status.NodeInfo.KernelVersion)
        if apicapi.ApicVersion >= "5.0" {
                aobj.SetAttr("id", fmt.Sprintf("%v", tunnelID))
        }
        cont.apicConn.WriteApicObjects(key, apicapi.ApicSlice{aobj})
}

func getNodeIP(node *v1.Node, aType v1.NodeAddressType) string {
        for _, a := range node.Status.Addresses {
                if a.Type == aType {
                        return a.Address
                }

        }

        return ""
}

func (cont *AciController) nodeChangedByName(nodeName string) {
        node, exists, err := cont.nodeIndexer.GetByKey(nodeName)
        if err != nil {
                cont.log.Error("Could not lookup node: ", err)
                return
        }
        if exists && node != nil {
                cont.nodeChanged(node)
        }
}

func (cont *AciController) nodeChanged(obj interface{}) {
        cont.indexMutex.Lock()

        node := obj.(*v1.Node)
        logger := cont.log.WithFields(logrus.Fields{
                "Node": node.ObjectMeta.Name,
        })

        nodeUpdated := false
        if node.ObjectMeta.Annotations == nil {
                node.ObjectMeta.Annotations = make(map[string]string)
        }

        nodeMeta, metaok := cont.nodeServiceMetaCache[node.ObjectMeta.Name]
        epval, epok := node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation]
        deviceMac, hasDevice := cont.deviceMacForNode(node.ObjectMeta.Name)

        if cont.nodeSyncEnabled && hasDevice {
                if !metaok {
                        nodeMeta = &nodeServiceMeta{}
                        cont.nodeServiceMetaCache[node.ObjectMeta.Name] = nodeMeta
                }

                existing := &metadata.ServiceEndpoint{}
                if epok {
                        err := json.Unmarshal([]byte(epval), existing)
                        if err != nil {
                                logger.WithFields(logrus.Fields{
                                        "epval": epval,
                                }).Warn("Could not parse existing node ",
                                        "service endpoint annotation: ", err)
                        }
                }

                cont.createServiceEndpoint(existing, &nodeMeta.serviceEp, deviceMac, node.ObjectMeta.Name)
                raw, err := json.Marshal(&nodeMeta.serviceEp)
                if err != nil {
                        logger.Error("Could not create node service endpoint annotation", err)
                } else {
                        serviceEpAnnotation := string(raw)
                        if !epok || serviceEpAnnotation != epval {
                                node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation] =
                                        serviceEpAnnotation
                                nodeUpdated = true
                                cont.updateServicesForNode(node.ObjectMeta.Name)
                                cont.snatFullSync()
                        }
                }
        }

        nodePodNet, ok := cont.nodePodNetCache[node.ObjectMeta.Name]
        if !ok {
                nodePodNet = newNodePodNetMeta()
                cont.nodePodNetCache[node.ObjectMeta.Name] = nodePodNet
        }

        netval := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]
        if cont.nodeSyncEnabled {
                cont.checkNodePodNet(node.ObjectMeta.Name)
                if netval != nodePodNet.podNetIpsAnnotation {
                        logger.Debug("Overwriting existing pod network: ", netval)
                        node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation] =
                                nodePodNet.podNetIpsAnnotation
                        nodeUpdated = true
                }
        }
        cont.indexMutex.Unlock()

        cont.createNetPolForNode(node)
        cont.writeApicNode(node)

        if nodeUpdated {
                _, err := cont.updateNode(node)
                if err != nil {
                        if serr, ok := err.(*kubeerr.StatusError); ok {
                                if serr.ErrStatus.Code == http.StatusConflict {
                                        logger.Debug("Conflict updating node; ",
                                                "will retry on next update")
                                        return
                                }
                        }
                        logger.Error("Failed to update node: ", err)
                } else {
                        logger.WithFields(logrus.Fields{
                                "ServiceEpAnnotation": node.
                                        ObjectMeta.Annotations[metadata.ServiceEpAnnotation],
                                "PodNetworkRangeAnnotation": node.
                                        ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation],
                        }).Info("Updated node annotations")
                }
        }
}

func (cont *AciController) nodeDeleted(obj interface{}) {
        node, isNode := obj.(*v1.Node)
        if !isNode {
                deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        cont.log.Error("Received unexpected object: ", obj)
                        return
                }
                node, ok = deletedState.Obj.(*v1.Node)
                if !ok {
                        cont.log.Error("DeletedFinalStateUnknown contained non-Node object: ", deletedState.Obj)
                        return
                }
        }
        cont.apicConn.ClearApicObjects(cont.aciNameForKey("node", node.Name))
        cont.apicConn.ClearApicObjects(cont.aciNameForKey("node-vmm", node.Name))

        cont.indexMutex.Lock()
        defer cont.indexMutex.Unlock()

        if existing, ok := cont.nodeServiceMetaCache[node.ObjectMeta.Name]; ok {
                if existing.serviceEp.Ipv4 != nil {
                        cont.nodeServiceIps.V4.AddIp(existing.serviceEp.Ipv4)
                }
                if existing.serviceEp.Ipv6 != nil {
                        cont.nodeServiceIps.V6.AddIp(existing.serviceEp.Ipv6)
                }
        }
        delete(cont.nodeServiceMetaCache, node.ObjectMeta.Name)
        cont.updateServicesForNode(node.ObjectMeta.Name)
        cont.snatFullSync()
        if _, ok := cont.snatNodeInfoCache[node.ObjectMeta.Name]; ok {
                nodeinfo := cont.snatNodeInfoCache[node.ObjectMeta.Name]
                delete(cont.snatNodeInfoCache, node.ObjectMeta.Name)
                nodeinfokey, _ := cache.MetaNamespaceKeyFunc(nodeinfo)
                cont.queueNodeInfoUpdateByKey(nodeinfokey)
        }
        env := cont.env.(*K8sEnvironment)
        nodeinfocl := env.nodeInfoClient
        //TODO Add reconcile here on failure
        if nodeinfocl != nil {
                err := util.DeleteNodeInfoCR(*nodeinfocl, node.ObjectMeta.Name)
                if err != nil {
                        cont.log.Error("Could not delete the NodeInfo", node.ObjectMeta.Name)
                }
        }
}

// must have index lock
func (cont *AciController) addPodToNode(nodename string, key string) {
        existing, ok := cont.nodePodNetCache[nodename]
        if !ok {
                existing = newNodePodNetMeta()
                cont.nodePodNetCache[nodename] = existing
        }
        if _, ok = existing.nodePods[key]; !ok {
                existing.nodePods[key] = true
                cont.checkNodePodNet(nodename)
        }
}

// must have index lock
func (cont *AciController) removePodFromNode(nodename string, key string) {
        if existing, ok := cont.nodePodNetCache[nodename]; ok {
                delete(existing.nodePods, key)
                cont.checkNodePodNet(nodename)
        }
}

func (cont *AciController) recomputePodNetAnnotation(podnet *nodePodNetMeta) {
        raw, err := json.Marshal(&podnet.podNetIps)
        if err != nil {
                cont.log.Error("Could not create node pod network ",
                        "annotation", err)
        }
        podnet.podNetIpsAnnotation = string(raw)
}

// must have index lock
func (cont *AciController) mergePodNet(podnet *nodePodNetMeta, existingAnnotation string, logger *logrus.Entry) {
        existing := &metadata.NetIps{}
        err := json.Unmarshal([]byte(existingAnnotation), existing)
        if err != nil {
                cont.log.Error("Could not parse existing pod network ",
                        "annotation", err)
                return
        }

        logger.Debug("Merging existing pod network: ", existingAnnotation)

        {
                v4 := ipam.NewFromRanges(podnet.podNetIps.V4)
                v4.AddRanges(existing.V4)
                annSize := v4.GetSize()

                // validate existing against configured range
                v4 = v4.Intersect(cont.configuredPodNetworkIps.V4)
                if v4.GetSize() != annSize {
                        logger.Warnf("intersect: %+v, config: %+v", v4, cont.configuredPodNetworkIps.V4)
                        logger.Warn("Existing annotation outside configured",
                                "range", existingAnnotation)
                }

                // mark the existing as allocated
                prevSize := cont.podNetworkIps.V4.GetSize()
                cont.podNetworkIps.V4.RemoveRanges(existing.V4)

                // verify allocation was successful
                newSize := cont.podNetworkIps.V4.GetSize()
                if (newSize + annSize) != prevSize {
                        logger.Warn("Existing annotation failed allocation: ",
                                existingAnnotation)
                }

                if len(v4.FreeList) > 0 {
                        podnet.podNetIps.V4 = v4.FreeList
                } else {
                        podnet.podNetIps.V4 = nil
                }
        }

        {
                v6 := ipam.NewFromRanges(podnet.podNetIps.V6)
                v6.AddRanges(existing.V6)
                v6 = v6.Intersect(cont.configuredPodNetworkIps.V6)
                cont.podNetworkIps.V6.RemoveRanges(existing.V6)
                if len(v6.FreeList) > 0 {
                        podnet.podNetIps.V6 = v6.FreeList
                } else {
                        podnet.podNetIps.V6 = nil
                }
        }

        cont.recomputePodNetAnnotation(podnet)
}

func (cont *AciController) allocateIpChunk(podnet *nodePodNetMeta, v4 bool) bool {
        var podnetipam, ipa *ipam.IpAlloc
        changed := false
        if v4 {
                podnetipam = ipam.NewFromRanges(podnet.podNetIps.V4)
                ipa = cont.podNetworkIps.V4
        } else {
                podnetipam = ipam.NewFromRanges(podnet.podNetIps.V6)
                ipa = cont.podNetworkIps.V6
        }
        size := podnetipam.GetSize()
        if int64(len(podnet.nodePods)) >
                size-int64(cont.config.PodIpPoolChunkSize)/2 {
                // we have half a chunk left or less; allocate a new chunk
                r, err := ipa.GetIpChunk(int64(cont.config.PodIpPoolChunkSize))
                if err != nil {
                        cont.log.Error("Could not allocate address chunk: ", err)
                } else {
                        podnetipam.AddRanges(r)
                        if v4 {
                                podnet.podNetIps.V4 = podnetipam.FreeList
                        } else {
                                podnet.podNetIps.V6 = podnetipam.FreeList
                        }
                        cont.recomputePodNetAnnotation(podnet)
                        changed = true
                }
        }
        return changed
}

// must have index lock
func (cont *AciController) checkNodePodNet(nodename string) {
        v4changed, v6changed := false, false
        if podnet, ok := cont.nodePodNetCache[nodename]; ok {
                if !cont.configuredPodNetworkIps.V4.Empty() {
                        v4changed = cont.allocateIpChunk(podnet, true)
                }
                if !cont.configuredPodNetworkIps.V6.Empty() {
                        v6changed = cont.allocateIpChunk(podnet, false)
                }
        }
        if v4changed || v6changed {
                go cont.env.NodePodNetworkChanged(nodename)
        }

}

1	// Copyright 2017 Cisco Systems, Inc.
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	// Handlers for node updates.
16
17	package controller
18
19	import (
20	"crypto/rand"
21	"encoding/json"
22	"errors"
23	"fmt"
24	"net"
25	"net/http"
26
27	v1 "k8s.io/api/core/v1"
28	kubeerr "k8s.io/apimachinery/pkg/api/errors"
29	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
30	"k8s.io/apimachinery/pkg/fields"
31	"k8s.io/apimachinery/pkg/labels"
32	"k8s.io/client-go/kubernetes"
33	"k8s.io/client-go/tools/cache"
34
35	"github.com/sirupsen/logrus"
36
37	"github.com/noironetworks/aci-containers/pkg/apicapi"
38	crdv1 "github.com/noironetworks/aci-containers/pkg/gbpcrd/apis/acipolicy/v1"
39	"github.com/noironetworks/aci-containers/pkg/gbpserver/watchers"
40	"github.com/noironetworks/aci-containers/pkg/ipam"
41	"github.com/noironetworks/aci-containers/pkg/metadata"
42	"github.com/noironetworks/aci-containers/pkg/util"
43	)
44
45	const (
46	tunnelIDIncr = 2
47	)
48
49	type tunnelState struct {
50	stateDriver *watchers.K8sStateDriver
51	nodeToTunnel map[string]int64
52	nextID int64
53	}
54
55	func (cont *AciController) initNodeInformerFromClient(
56	kubeClient *kubernetes.Clientset) {	×
57		×
58	cont.initNodeInformerBase(	×
59	cache.NewListWatchFromClient(	×
60	kubeClient.CoreV1().RESTClient(), "nodes",	×
61	metav1.NamespaceAll, fields.Everything()))	×
62	}	×
63
64	func (cont AciController) initNodeInformerBase(listWatch cache.ListWatch) {	1✔
65	cont.nodeIndexer, cont.nodeInformer = cache.NewIndexerInformer(	1✔
66	listWatch, &v1.Node{}, 0,	1✔
67	cache.ResourceEventHandlerFuncs{	1✔
68	AddFunc: func(obj interface{}) {	2✔
69	cont.syncPodNet(obj) // update cache	1✔
70	cont.nodeChanged(obj)	1✔
71	},	1✔
72	UpdateFunc: func(_ interface{}, obj interface{}) {	1✔
73	cont.nodeChanged(obj)	1✔
74	},	1✔
75	DeleteFunc: func(obj interface{}) {	1✔
76	cont.nodeDeleted(obj)	1✔
77	},	1✔
78	},
79	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
80	)
81	}
82
83	func apicNodeNetPol(name string, tenantName string,
84	nodeIps map[string]bool) apicapi.ApicObject {	1✔
85		1✔
86	hpp := apicapi.NewHostprotPol(tenantName, name)	1✔
87	hppDn := hpp.GetDn()	1✔
88	nodeSubj := apicapi.NewHostprotSubj(hppDn, "local-node")	1✔
89	if len(nodeIps) > 0 {	2✔
90	nodeSubjDn := nodeSubj.GetDn()	1✔
91	outbound := apicapi.NewHostprotRule(nodeSubjDn, "allow-all-egress")	1✔
92	outbound.SetAttr("direction", "egress")	1✔
93	outbound.SetAttr("ethertype", "ipv4")	1✔
94	outbound.SetAttr("connTrack", "normal")	1✔
95		1✔
96	inbound := apicapi.NewHostprotRule(nodeSubjDn, "allow-all-ingress")	1✔
97	inbound.SetAttr("direction", "ingress")	1✔
98	inbound.SetAttr("ethertype", "ipv4")	1✔
99	inbound.SetAttr("connTrack", "normal")	1✔
100		1✔
101	for ip := range nodeIps {	2✔
102	outbound.AddChild(apicapi.NewHostprotRemoteIp(outbound.GetDn(), ip))	1✔
103	inbound.AddChild(apicapi.NewHostprotRemoteIp(inbound.GetDn(), ip))	1✔
104	}	1✔
105
106	nodeSubj.AddChild(inbound)	1✔
107	nodeSubj.AddChild(outbound)	1✔
108	}
109	hpp.AddChild(nodeSubj)	1✔
110	return hpp	1✔
111	}
112
113	func (cont AciController) createNetPolForNode(node v1.Node) {	1✔
114	nodeIps := make(map[string]bool)	1✔
115	for _, a := range node.Status.Addresses {	2✔
116	if a.Address != "" &&	1✔
117	(a.Type == "InternalIP" \|\| a.Type == "ExternalIP") {	2✔
118	nodeIps[a.Address] = true	1✔
119	}	1✔
120	}
121
122	sgName := cont.aciNameForKey("node", node.Name)	1✔
123	cont.apicConn.WriteApicObjects(sgName,	1✔
124	apicapi.ApicSlice{	1✔
125	apicNodeNetPol(sgName, cont.config.AciPolicyTenant, nodeIps),	1✔
126	})	1✔
127	}
128
129	func (cont AciController) createServiceEndpoint(existing metadata.ServiceEndpoint, ep *metadata.ServiceEndpoint, deviceMac string, nodeName string) error {	1✔
130		1✔
131	_, err := net.ParseMAC(deviceMac)	1✔
132	if err == nil && deviceMac != "00:00:00:00:00:00" {	2✔
133	ep.Mac = deviceMac	1✔
134	} else {	2✔
135	_, err := net.ParseMAC(existing.Mac)	1✔
136	if err == nil {	2✔
137	ep.Mac = existing.Mac	1✔
138	} else {	2✔
139	var mac net.HardwareAddr	1✔
140	mac = make([]byte, 6)	1✔
141	_, err := rand.Read(mac)	1✔
142	if err != nil {	1✔
143	return err	×
144	}	×
145
146	mac[0] = (mac[0] & 254) \| 2	1✔
147	ep.Mac = mac.String()	1✔
148	}
149	}
150
151	if ep.Ipv4 == nil && existing.Ipv4 != nil &&	1✔
152	cont.nodeServiceIps.V4.RemoveIp(existing.Ipv4) {	2✔
153	ep.Ipv4 = existing.Ipv4	1✔
154	}	1✔
155
156	if ep.Ipv4 == nil {	2✔
157	ipv4, err := cont.nodeServiceIps.V4.GetIp()	1✔
158	if err == nil {	2✔
159	ep.Ipv4 = ipv4	1✔
160	} else {	2✔
161	ep.Ipv4 = nil	1✔
162	}	1✔
163	}
164
165	if ep.Ipv6 == nil && existing.Ipv6 != nil &&	1✔
166	cont.nodeServiceIps.V6.RemoveIp(existing.Ipv6) {	1✔
167	ep.Ipv6 = existing.Ipv6	×
168	}	×
169
170	if ep.Ipv6 == nil {	2✔
171	ipv6, err := cont.nodeServiceIps.V6.GetIp()	1✔
172	if err == nil {	2✔
173	ep.Ipv6 = ipv6	1✔
174	} else {	2✔
175	ep.Ipv6 = nil	1✔
176	}	1✔
177	}
178
179	if ep.Ipv4 == nil && ep.Ipv6 == nil {	2✔
180	return errors.New("No IP addresses available for service endpoint")	1✔
181	}	1✔
182
183	if (ep.HealthGroupDn == "") && (cont.config.AciServiceMonitorInterval > 0) {	2✔
184	name := cont.aciNameForKey("svc", nodeName)	1✔
185	healthGroupObj := apicapi.NewVnsRedirectHealthGroup(cont.config.AciVrfTenant, name)	1✔
186	ep.HealthGroupDn = healthGroupObj.GetDn()	1✔
187	cont.apicConn.WriteApicObjects(name, apicapi.ApicSlice{healthGroupObj})	1✔
188	}	1✔
189
190	return nil	1✔
191	}
192
193	func (cont *AciController) nodeFullSync() {	1✔
194	cache.ListAll(cont.nodeIndexer, labels.Everything(),	1✔
195	func(nodeobj interface{}) {	2✔
196	cont.syncPodNet(nodeobj) // update the cache	1✔
197	cont.nodeChanged(nodeobj)	1✔
198	})	1✔
199	}
200
201	// syncPodNet syncs in the podnets from the node object's annotation
202	func (cont *AciController) syncPodNet(obj interface{}) {	1✔
203	cont.indexMutex.Lock()	1✔
204	defer cont.indexMutex.Unlock()	1✔
205		1✔
206	node := obj.(*v1.Node)	1✔
207	logger := cont.log.WithFields(logrus.Fields{	1✔
208	"Node": node.ObjectMeta.Name,	1✔
209	})	1✔
210		1✔
211	if node.ObjectMeta.Annotations == nil {	1✔
212	return // nothing to sync	×
213	}	×
214
215	netval, ok := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]	1✔
216	if !ok {	2✔
217	return // nothing to sync	1✔
218	}	1✔
219
220	nodePodNet := newNodePodNetMeta()	1✔
221	cont.mergePodNet(nodePodNet, netval, logger)	1✔
222	cont.nodePodNetCache[node.ObjectMeta.Name] = nodePodNet	1✔
223	}
224
225	func maxof(x1, x2 int64) int64 {	×
226	if x1 > x2 {	×
227	return x1	×
228	}	×
229
230	return x2	×
231	}
232
233	func (cont AciController) getTunnelID(node v1.Node) int64 {	1✔
234	if cont.config.LBType == lbTypeAci {	2✔
235	return 0	1✔
236	}	1✔
237
238	nodeIP := getNodeIP(node, v1.NodeInternalIP)	×
239	if nodeIP == "" {	×
240	cont.log.Errorf("Can't get IP for node %s", node.Name)	×
241	return 0	×
242	}	×
243
244	if cont.tunnelGetter == nil {	×
245	tunnelGetter := &tunnelState{}	×
246	tunnelGetter.stateDriver = &watchers.K8sStateDriver{}	×
247	err := tunnelGetter.stateDriver.Init(watchers.FieldTunnelID)	×
248	if err != nil {	×
249	cont.log.Warnf("Can't get tunnelID for %s", node.Name)	×
250	return 0	×
251	}	×
252
253	gbps, err := tunnelGetter.stateDriver.Get()	×
254	if err != nil {	×
255	cont.log.Warnf("Can't get tunnelID for %s", node.Name)	×
256	return 0	×
257	}	×
258
259	cont.tunnelGetter = tunnelGetter	×
260	if gbps.Status.TunnelIDs == nil {	×
261	cont.log.Infof("Initializing tunnelIDs")	×
262	tunnelGetter.nodeToTunnel = make(map[string]int64)	×
263	cont.tunnelGetter.nextID = 0	×
264	} else {	×
265	tunnelGetter.nodeToTunnel = gbps.Status.TunnelIDs	×
266	for _, val := range tunnelGetter.nodeToTunnel {	×
267	cont.tunnelGetter.nextID = maxof(cont.tunnelGetter.nextID, val)	×
268	}	×
269
270	cont.tunnelGetter.nextID = cont.tunnelGetter.nextID + tunnelIDIncr	×
271	}
272	}
273
274	id, found := cont.tunnelGetter.nodeToTunnel[nodeIP]	×
275	if found {	×
276	return id + int64(cont.config.CSRTunnelIDBase)	×
277	}	×
278
279	id = cont.tunnelGetter.nextID	×
280	if id >= int64(cont.config.MaxCSRTunnels*tunnelIDIncr) {	×
281	cont.log.Infof("Max tunnels %d reached", cont.config.MaxCSRTunnels)	×
282	return 0	×
283	}	×
284
285	// each allocated id packs two tunnels.
286	cont.tunnelGetter.nodeToTunnel[nodeIP] = id	×
287	cont.tunnelGetter.nextID = id + tunnelIDIncr	×
288	state := &crdv1.GBPSState{}	×
289	state.Status.TunnelIDs = cont.tunnelGetter.nodeToTunnel	×
290	cont.tunnelGetter.stateDriver.Update(state)	×
291	return id + int64(cont.config.CSRTunnelIDBase)	×
292	}
293
294	func (cont AciController) writeApicNode(node v1.Node) {	1✔
295	tunnelID := cont.getTunnelID(node)	1✔
296	//cont.log.Infof("=> Node: %s, tunnelID: %v", node.Name, tunnelID)	1✔
297	key := cont.aciNameForKey("node-vmm", node.Name)	1✔
298	aobj := apicapi.NewVmmInjectedHost(cont.vmmDomainProvider(),	1✔
299	cont.config.AciVmmDomain, cont.config.AciVmmController,	1✔
300	node.Name)	1✔
301	aobj.SetAttr("mgmtIp", getNodeIP(node, v1.NodeInternalIP))	1✔
302	aobj.SetAttr("os", node.Status.NodeInfo.OSImage)	1✔
303	aobj.SetAttr("kernelVer", node.Status.NodeInfo.KernelVersion)	1✔
304	if apicapi.ApicVersion >= "5.0" {	1✔
305	aobj.SetAttr("id", fmt.Sprintf("%v", tunnelID))	×
306	}	×
307	cont.apicConn.WriteApicObjects(key, apicapi.ApicSlice{aobj})	1✔
308	}
309
310	func getNodeIP(node *v1.Node, aType v1.NodeAddressType) string {	1✔
311	for _, a := range node.Status.Addresses {	2✔
312	if a.Type == aType {	2✔
313	return a.Address	1✔
314	}	1✔
315
316	}
317
318	return ""	1✔
319	}
320
321	func (cont *AciController) nodeChangedByName(nodeName string) {	1✔
322	node, exists, err := cont.nodeIndexer.GetByKey(nodeName)	1✔
323	if err != nil {	1✔
324	cont.log.Error("Could not lookup node: ", err)	×
325	return	×
326	}	×
327	if exists && node != nil {	2✔
328	cont.nodeChanged(node)	1✔
329	}	1✔
330	}
331
332	func (cont *AciController) nodeChanged(obj interface{}) {	1✔
333	cont.indexMutex.Lock()	1✔
334		1✔
335	node := obj.(*v1.Node)	1✔
336	logger := cont.log.WithFields(logrus.Fields{	1✔
337	"Node": node.ObjectMeta.Name,	1✔
338	})	1✔
339		1✔
340	nodeUpdated := false	1✔
341	if node.ObjectMeta.Annotations == nil {	1✔
342	node.ObjectMeta.Annotations = make(map[string]string)	×
343	}	×
344
345	nodeMeta, metaok := cont.nodeServiceMetaCache[node.ObjectMeta.Name]	1✔
346	epval, epok := node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation]	1✔
347	deviceMac, hasDevice := cont.deviceMacForNode(node.ObjectMeta.Name)	1✔
348		1✔
349	if cont.nodeSyncEnabled && hasDevice {	2✔
350	if !metaok {	2✔
351	nodeMeta = &nodeServiceMeta{}	1✔
352	cont.nodeServiceMetaCache[node.ObjectMeta.Name] = nodeMeta	1✔
353	}	1✔
354
355	existing := &metadata.ServiceEndpoint{}	1✔
356	if epok {	2✔
357	err := json.Unmarshal([]byte(epval), existing)	1✔
358	if err != nil {	1✔
359	logger.WithFields(logrus.Fields{	×
360	"epval": epval,	×
361	}).Warn("Could not parse existing node ",	×
362	"service endpoint annotation: ", err)	×
363	}	×
364	}
365
366	cont.createServiceEndpoint(existing, &nodeMeta.serviceEp, deviceMac, node.ObjectMeta.Name)	1✔
367	raw, err := json.Marshal(&nodeMeta.serviceEp)	1✔
368	if err != nil {	1✔
369	logger.Error("Could not create node service endpoint annotation", err)	×
370	} else {	1✔
371	serviceEpAnnotation := string(raw)	1✔
372	if !epok \|\| serviceEpAnnotation != epval {	2✔
373	node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation] =	1✔
374	serviceEpAnnotation	1✔
375	nodeUpdated = true	1✔
376	cont.updateServicesForNode(node.ObjectMeta.Name)	1✔
377	cont.snatFullSync()	1✔
378	}	1✔
379	}
380	}
381
382	nodePodNet, ok := cont.nodePodNetCache[node.ObjectMeta.Name]	1✔
383	if !ok {	2✔
384	nodePodNet = newNodePodNetMeta()	1✔
385	cont.nodePodNetCache[node.ObjectMeta.Name] = nodePodNet	1✔
386	}	1✔
387
388	netval := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]	1✔
389	if cont.nodeSyncEnabled {	2✔
390	cont.checkNodePodNet(node.ObjectMeta.Name)	1✔
391	if netval != nodePodNet.podNetIpsAnnotation {	2✔
392	logger.Debug("Overwriting existing pod network: ", netval)	1✔
393	node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation] =	1✔
394	nodePodNet.podNetIpsAnnotation	1✔
395	nodeUpdated = true	1✔
396	}	1✔
397	}
398	cont.indexMutex.Unlock()	1✔
399		1✔
400	cont.createNetPolForNode(node)	1✔
401	cont.writeApicNode(node)	1✔
402		1✔
403	if nodeUpdated {	2✔
404	_, err := cont.updateNode(node)	1✔
405	if err != nil {	1✔
406	if serr, ok := err.(*kubeerr.StatusError); ok {	×
407	if serr.ErrStatus.Code == http.StatusConflict {	×
408	logger.Debug("Conflict updating node; ",	×
409	"will retry on next update")	×
410	return	×
411	}	×
412	}
413	logger.Error("Failed to update node: ", err)	×
414	} else {	1✔
415	logger.WithFields(logrus.Fields{	1✔
416	"ServiceEpAnnotation": node.	1✔
417	ObjectMeta.Annotations[metadata.ServiceEpAnnotation],	1✔
418	"PodNetworkRangeAnnotation": node.	1✔
419	ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation],	1✔
420	}).Info("Updated node annotations")	1✔
421	}	1✔
422	}
423	}
424
425	func (cont *AciController) nodeDeleted(obj interface{}) {	1✔
426	node, isNode := obj.(*v1.Node)	1✔
427	if !isNode {	1✔
428	deletedState, ok := obj.(cache.DeletedFinalStateUnknown)	×
429	if !ok {	×
430	cont.log.Error("Received unexpected object: ", obj)	×
431	return	×
432	}	×
433	node, ok = deletedState.Obj.(*v1.Node)	×
434	if !ok {	×
435	cont.log.Error("DeletedFinalStateUnknown contained non-Node object: ", deletedState.Obj)	×
436	return	×
437	}	×
438	}
439	cont.apicConn.ClearApicObjects(cont.aciNameForKey("node", node.Name))	1✔
440	cont.apicConn.ClearApicObjects(cont.aciNameForKey("node-vmm", node.Name))	1✔
441		1✔
442	cont.indexMutex.Lock()	1✔
443	defer cont.indexMutex.Unlock()	1✔
444		1✔
445	if existing, ok := cont.nodeServiceMetaCache[node.ObjectMeta.Name]; ok {	2✔
446	if existing.serviceEp.Ipv4 != nil {	2✔
447	cont.nodeServiceIps.V4.AddIp(existing.serviceEp.Ipv4)	1✔
448	}	1✔
449	if existing.serviceEp.Ipv6 != nil {	1✔
450	cont.nodeServiceIps.V6.AddIp(existing.serviceEp.Ipv6)	×
451	}	×
452	}
453	delete(cont.nodeServiceMetaCache, node.ObjectMeta.Name)	1✔
454	cont.updateServicesForNode(node.ObjectMeta.Name)	1✔
455	cont.snatFullSync()	1✔
456	if _, ok := cont.snatNodeInfoCache[node.ObjectMeta.Name]; ok {	1✔
457	nodeinfo := cont.snatNodeInfoCache[node.ObjectMeta.Name]	×
458	delete(cont.snatNodeInfoCache, node.ObjectMeta.Name)	×
459	nodeinfokey, _ := cache.MetaNamespaceKeyFunc(nodeinfo)	×
460	cont.queueNodeInfoUpdateByKey(nodeinfokey)	×
461	}	×
462	env := cont.env.(*K8sEnvironment)	1✔
463	nodeinfocl := env.nodeInfoClient	1✔
464	//TODO Add reconcile here on failure	1✔
465	if nodeinfocl != nil {	1✔
466	err := util.DeleteNodeInfoCR(*nodeinfocl, node.ObjectMeta.Name)	×
467	if err != nil {	×
468	cont.log.Error("Could not delete the NodeInfo", node.ObjectMeta.Name)	×
469	}	×
470	}
471	}
472
473	// must have index lock
474	func (cont *AciController) addPodToNode(nodename string, key string) {	1✔
475	existing, ok := cont.nodePodNetCache[nodename]	1✔
476	if !ok {	2✔
477	existing = newNodePodNetMeta()	1✔
478	cont.nodePodNetCache[nodename] = existing	1✔
479	}	1✔
480	if _, ok = existing.nodePods[key]; !ok {	2✔
481	existing.nodePods[key] = true	1✔
482	cont.checkNodePodNet(nodename)	1✔
483	}	1✔
484	}
485
486	// must have index lock
487	func (cont *AciController) removePodFromNode(nodename string, key string) {	×
488	if existing, ok := cont.nodePodNetCache[nodename]; ok {	×
489	delete(existing.nodePods, key)	×
490	cont.checkNodePodNet(nodename)	×
491	}	×
492	}
493
494	func (cont AciController) recomputePodNetAnnotation(podnet nodePodNetMeta) {	1✔
495	raw, err := json.Marshal(&podnet.podNetIps)	1✔
496	if err != nil {	1✔
497	cont.log.Error("Could not create node pod network ",	×
498	"annotation", err)	×
499	}	×
500	podnet.podNetIpsAnnotation = string(raw)	1✔
501	}
502
503	// must have index lock
504	func (cont AciController) mergePodNet(podnet nodePodNetMeta, existingAnnotation string, logger *logrus.Entry) {	1✔
505	existing := &metadata.NetIps{}	1✔
506	err := json.Unmarshal([]byte(existingAnnotation), existing)	1✔
507	if err != nil {	1✔
508	cont.log.Error("Could not parse existing pod network ",	×
509	"annotation", err)	×
510	return	×
511	}	×
512
513	logger.Debug("Merging existing pod network: ", existingAnnotation)	1✔
514		1✔
515	{	2✔
516	v4 := ipam.NewFromRanges(podnet.podNetIps.V4)	1✔
517	v4.AddRanges(existing.V4)	1✔
518	annSize := v4.GetSize()	1✔
519		1✔
520	// validate existing against configured range	1✔
521	v4 = v4.Intersect(cont.configuredPodNetworkIps.V4)	1✔
522	if v4.GetSize() != annSize {	2✔
523	logger.Warnf("intersect: %+v, config: %+v", v4, cont.configuredPodNetworkIps.V4)	1✔
524	logger.Warn("Existing annotation outside configured",	1✔
525	"range", existingAnnotation)	1✔
526	}	1✔
527
528	// mark the existing as allocated
529	prevSize := cont.podNetworkIps.V4.GetSize()	1✔
530	cont.podNetworkIps.V4.RemoveRanges(existing.V4)	1✔
531		1✔
532	// verify allocation was successful	1✔
533	newSize := cont.podNetworkIps.V4.GetSize()	1✔
534	if (newSize + annSize) != prevSize {	2✔
535	logger.Warn("Existing annotation failed allocation: ",	1✔
536	existingAnnotation)	1✔
537	}	1✔
538
539	if len(v4.FreeList) > 0 {	2✔
540	podnet.podNetIps.V4 = v4.FreeList	1✔
541	} else {	1✔
542	podnet.podNetIps.V4 = nil	×
543	}	×
544	}
545
546	{	1✔
547	v6 := ipam.NewFromRanges(podnet.podNetIps.V6)	1✔
548	v6.AddRanges(existing.V6)	1✔
549	v6 = v6.Intersect(cont.configuredPodNetworkIps.V6)	1✔
550	cont.podNetworkIps.V6.RemoveRanges(existing.V6)	1✔
551	if len(v6.FreeList) > 0 {	1✔
552	podnet.podNetIps.V6 = v6.FreeList	×
553	} else {	1✔
554	podnet.podNetIps.V6 = nil	1✔
555	}	1✔
556	}
557
558	cont.recomputePodNetAnnotation(podnet)	1✔
559	}
560
561	func (cont AciController) allocateIpChunk(podnet nodePodNetMeta, v4 bool) bool {	1✔
562	var podnetipam, ipa *ipam.IpAlloc	1✔
563	changed := false	1✔
564	if v4 {	2✔
565	podnetipam = ipam.NewFromRanges(podnet.podNetIps.V4)	1✔
566	ipa = cont.podNetworkIps.V4	1✔
567	} else {	2✔
568	podnetipam = ipam.NewFromRanges(podnet.podNetIps.V6)	1✔
569	ipa = cont.podNetworkIps.V6	1✔
570	}	1✔
571	size := podnetipam.GetSize()	1✔
572	if int64(len(podnet.nodePods)) >	1✔
573	size-int64(cont.config.PodIpPoolChunkSize)/2 {	2✔
574	// we have half a chunk left or less; allocate a new chunk	1✔
575	r, err := ipa.GetIpChunk(int64(cont.config.PodIpPoolChunkSize))	1✔
576	if err != nil {	2✔
577	cont.log.Error("Could not allocate address chunk: ", err)	1✔
578	} else {	2✔
579	podnetipam.AddRanges(r)	1✔
580	if v4 {	2✔
581	podnet.podNetIps.V4 = podnetipam.FreeList	1✔
582	} else {	2✔
583	podnet.podNetIps.V6 = podnetipam.FreeList	1✔
584	}	1✔
585	cont.recomputePodNetAnnotation(podnet)	1✔
586	changed = true	1✔
587	}
588	}
589	return changed	1✔
590	}
591
592	// must have index lock
593	func (cont *AciController) checkNodePodNet(nodename string) {	1✔
594	v4changed, v6changed := false, false	1✔
595	if podnet, ok := cont.nodePodNetCache[nodename]; ok {	2✔
596	if !cont.configuredPodNetworkIps.V4.Empty() {	2✔
597	v4changed = cont.allocateIpChunk(podnet, true)	1✔
598	}	1✔
599	if !cont.configuredPodNetworkIps.V6.Empty() {	2✔
600	v6changed = cont.allocateIpChunk(podnet, false)	1✔
601	}	1✔
602	}
603	if v4changed \|\| v6changed {	2✔
604	go cont.env.NodePodNetworkChanged(nodename)	1✔
605	}	1✔
606
607	}

noironetworks / aci-containers / 4318

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous