4718

Committed 04 May 2021 06:42PM UTC coverage: 58.336%. First build

Build # 4718

Build Type

push

travis-ci-com

Committed by

jayaramsatya

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/556e984ac1bb318a491a697a10cd7f11e341191e">556e984ac">&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/&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/556e984ac1bb318a491a697a10cd7f11e341191e">556e984ac</a><a href="https://github.com/noironetworks/aci-containers/commit/556e984ac1bb318a491a697a10cd7f11e341191e">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/noironetworks/aci-containers/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/noironetworks/aci-containers/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/noironetworks/aci-containers/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/noironetworks/aci-containers/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/noironetworks/aci-containers/commit/556e984ac1bb318a491a... (continued)

Run Details

28 of 28 new or added lines in 2 files covered. (100.0%)

11162 of 19134 relevant lines covered (58.34%)

0.65 hits per line

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

53.54

/pkg/controller/snatglobalinfo.go

// Copyright 2019 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.

package controller

import (
        "context"
        uuid "github.com/google/uuid"
        nodeinfo "github.com/noironetworks/aci-containers/pkg/nodeinfo/apis/aci.snat/v1"
        nodeinfoclset "github.com/noironetworks/aci-containers/pkg/nodeinfo/clientset/versioned"
        snatglobalinfo "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/apis/aci.snat/v1"
        snatv1 "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
        "github.com/noironetworks/aci-containers/pkg/util"
        "k8s.io/api/core/v1"
        "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/apimachinery/pkg/runtime"
        "k8s.io/apimachinery/pkg/watch"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/tools/cache"
        "net"
        "reflect"
        "strconv"
)

type set map[string]bool

func (cont *AciController) initSnatNodeInformerFromClient(
        snatClient *nodeinfoclset.Clientset) {
        cont.initSnatNodeInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                return snatClient.AciV1().NodeInfos(metav1.NamespaceAll).List(context.TODO(), options)
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                return snatClient.AciV1().NodeInfos(metav1.NamespaceAll).Watch(context.TODO(), options)
                        },
                })
}

func (cont *AciController) initSnatNodeInformerBase(listWatch *cache.ListWatch) {
        cont.snatNodeInfoIndexer, cont.snatNodeInformer = cache.NewIndexerInformer(
                listWatch,
                &nodeinfo.NodeInfo{}, 0,
                cache.ResourceEventHandlerFuncs{
                        AddFunc: func(obj interface{}) {
                                cont.snatNodeInfoAdded(obj)
                        },
                        UpdateFunc: func(oldiobj interface{}, newobj interface{}) {
                                cont.snatNodeInfoUpdated(oldiobj, newobj)
                        },
                        DeleteFunc: func(obj interface{}) {
                                cont.snatNodeInfoDeleted(obj)
                        },
                },
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        cont.log.Debug("Initializing Node Informers: ")
}

func (cont *AciController) initSnatCfgFromClient(
        kubeClient kubernetes.Interface) {
        cont.initSnatCfgInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                options.FieldSelector =
                                        fields.Set{"metadata.name": "snat-operator-config"}.String()
                                return kubeClient.CoreV1().ConfigMaps(
                                        "aci-containers-system").List(context.TODO(), options)
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                options.FieldSelector =
                                        fields.Set{"metadata.name": "snat-operator-config"}.String()
                                return kubeClient.CoreV1().ConfigMaps(
                                        "aci-containers-system").Watch(context.TODO(), options)
                        },
                })
}

func (cont *AciController) initSnatCfgInformerBase(listWatch *cache.ListWatch) {
        _, cont.snatCfgInformer = cache.NewIndexerInformer(
                listWatch,
                &v1.ConfigMap{}, 0,
                cache.ResourceEventHandlerFuncs{
                        UpdateFunc: func(_, obj interface{}) {
                                cont.snatCfgUpdate(obj)
                        },
                },
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        cont.log.Debug("Initializing SnatCfg  Informers: ")
}

// Handle any changes to snatOperator Config
func (cont *AciController) snatCfgUpdate(obj interface{}) {
        snatcfg := obj.(*v1.ConfigMap)
        var portRange snatglobalinfo.PortRange
        cont.log.Debug("snatCfgUpdated: ", snatcfg)
        data := snatcfg.Data
        start, err1 := strconv.Atoi(data["start"])
        end, err2 := strconv.Atoi(data["end"])
        portsPerNode, err3 := strconv.Atoi(data["ports-per-node"])
        if err1 != nil || err2 != nil || err3 != nil ||
                start < 5000 || end > 65000 || start > end || portsPerNode > end-start+1 {
                return
        }
        portRange.Start = start
        portRange.End = end
        var currPortRange []snatglobalinfo.PortRange
        currPortRange = append(currPortRange, portRange)
        cont.indexMutex.Lock()
        nodeInfoKeys := make(map[string]bool)
        for name, info := range cont.snatPolicyCache {
                cont.clearSnatGlobalCache(name, "")
                info.ExpandedSnatPorts = util.ExpandPortRanges(currPortRange, portsPerNode)
                cont.getNodeInfoKeys(name, nodeInfoKeys)
        }
        cont.indexMutex.Unlock()
        for key := range nodeInfoKeys {
                cont.queueNodeInfoUpdateByKey(key)
        }
}

func (cont *AciController) snatNodeInfoAdded(obj interface{}) {
        nodeinfo := obj.(*nodeinfo.NodeInfo)
        nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)
        if err != nil {
                return
        }
        cont.log.Debug("Node Info Added: ", nodeinfokey)
        cont.indexMutex.Lock()
        cont.snatNodeInfoCache[nodeinfo.ObjectMeta.Name] = nodeinfo
        cont.indexMutex.Unlock()
        cont.queueNodeInfoUpdateByKey(nodeinfokey)
}

func (cont *AciController) queueNodeInfoUpdateByKey(key string) {
        cont.log.Debug("Node key Queued: ", key)
        cont.snatNodeInfoQueue.Add(key)
}

func (cont *AciController) snatNodeInfoUpdated(oldobj interface{}, newobj interface{}) {
        oldnodeinfo := oldobj.(*nodeinfo.NodeInfo)
        newnodeinfo := newobj.(*nodeinfo.NodeInfo)
        nodeinfokey, err := cache.MetaNamespaceKeyFunc(newnodeinfo)
        if err != nil {
                return
        }
        if reflect.DeepEqual(oldnodeinfo.Spec, newnodeinfo.Spec) {
                return
        }
        cont.indexMutex.Lock()
        cont.updateSnatIpandPorts(oldnodeinfo.Spec.SnatPolicyNames,
                newnodeinfo.Spec.SnatPolicyNames, newnodeinfo.ObjectMeta.Name)
        cont.snatNodeInfoCache[newnodeinfo.ObjectMeta.Name] = newnodeinfo
        cont.indexMutex.Unlock()
        cont.queueNodeInfoUpdateByKey(nodeinfokey)
}

func (cont *AciController) snatNodeInfoDeleted(obj interface{}) {
        nodeinfo := obj.(*nodeinfo.NodeInfo)
        nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)
        if err != nil {
                return
        }
        cont.indexMutex.Lock()
        delete(cont.snatNodeInfoCache, nodeinfo.ObjectMeta.Name)
        cont.indexMutex.Unlock()
        cont.queueNodeInfoUpdateByKey(nodeinfokey)
}

func (cont *AciController) setSnatPolicyStaus(snatPolicyName string, status snatv1.PolicyState) bool {
        obj, exists, err := cont.snatIndexer.GetByKey(snatPolicyName)
        if err == nil && exists && obj != nil {
                snatpolicy := obj.(*snatv1.SnatPolicy)
                if snatpolicy.Status.State != status {
                        snatpolicy.Status.State = status
                        env := cont.env.(*K8sEnvironment)
                        policycl := env.snatClient
                        if policycl != nil {
                                err = util.UpdateSnatPolicyCR(*policycl, snatpolicy)
                                if err != nil {
                                        cont.log.Info("Policy status update failed queue the request again: ", err)
                                        return true
                                }
                        }
                }
        }
        return false
}

func (cont *AciController) isSnatNodeInfoPresent(nodeName string) bool {
        cont.indexMutex.Lock()
        _, ok := cont.snatNodeInfoCache[nodeName]
        cont.indexMutex.Unlock()
        return ok
}

func (cont *AciController) checksnatPolicyPortExhausted(name string) bool {
        snatobj, exists, err := cont.snatIndexer.GetByKey(name)
        if err == nil && exists && snatobj != nil {
                snatpolicy := snatobj.(*snatv1.SnatPolicy)
                if snatpolicy.Status.State == snatv1.IpPortsExhausted {
                        return true
                }
        }
        return false
}

func (cont *AciController) handleSnatNodeInfo(nodeinfo *nodeinfo.NodeInfo) bool {
        cont.log.Debug("handle Node Info: ", nodeinfo)
        updated := false
        nodename := nodeinfo.ObjectMeta.Name
        ret := false
        // Cache needs to be updated
        if !cont.isSnatNodeInfoPresent(nodename) || len(nodeinfo.Spec.SnatPolicyNames) == 0 {
                ret = cont.deleteNodeinfoFromGlInfoCache(nodename)
                updated = true
        } else {
                // This case ignores any stale entry is present in nodeinfo
                _, _, err := cont.nodeIndexer.GetByKey(nodename)
                if err != nil {
                        cont.log.Info("Could not lookup node: ", err, "nodeName: ", nodename)
                        return false
                }
                allocfailed := make(map[string]bool)
                markready := make(map[string]bool)
                for name := range nodeinfo.Spec.SnatPolicyNames {
                        cont.indexMutex.Lock()
                        snatpolicy, ok := cont.snatPolicyCache[name]
                        cont.indexMutex.Unlock()
                        if !ok {
                                continue
                        }
                        cont.log.Debug("SnatPolicy Name: ", name)
                        if len(snatpolicy.SnatIp) != 0 {
                                snatIp, portrange, alloc := cont.getIpAndPortRange(nodename, snatpolicy, "")
                                cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)
                                if alloc == false {
                                        cont.log.Error("Port Range Exhausted: ", name)
                                        allocfailed[name] = true
                                        continue
                                } else {
                                        if cont.checksnatPolicyPortExhausted(name) {
                                                markready[name] = true
                                        }
                                }
                                cont.updateGlobalInfoforPolicy(portrange, snatIp, nodename,
                                        nodeinfo.Spec.Macaddress, name)
                                updated = true
                        } else {
                                snatIps := cont.getServiceIps(snatpolicy)
                                cont.log.Debug("Service Ips: ", snatIps)
                                for _, snatip := range snatIps {
                                        snatIp, portrange, alloc := cont.getIpAndPortRange(nodename, snatpolicy, snatip)
                                        cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)
                                        if alloc == false {
                                                cont.log.Error("Port Range Exhausted: ", name)
                                                allocfailed[name] = true
                                                continue
                                        } else {
                                                if cont.checksnatPolicyPortExhausted(name) {
                                                        markready[name] = true
                                                }
                                        }
                                        cont.updateGlobalInfoforPolicy(portrange, snatIp, nodename,
                                                nodeinfo.Spec.Macaddress, name)
                                        updated = true
                                }
                        }
                }
                ret = cont.setSnatPoliciesState(allocfailed, snatv1.IpPortsExhausted)
                ret = cont.setSnatPoliciesState(markready, snatv1.Ready)
        }

        if updated {
                cont.scheduleSyncGlobalInfo()
        }
        return ret
}

func (cont *AciController) syncSnatGlobalInfo() bool {
        env := cont.env.(*K8sEnvironment)
        globalcl := env.snatGlobalClient
        if globalcl == nil {
                return false
        }
        cont.indexMutex.Lock()
        cont.log.Debug("syncSnatGlobalInfo")
        glInfoCache := make(map[string]snatglobalinfo.GlobalInfoList)
        for _, glinfos := range cont.snatGlobalInfoCache {
                for name, v := range glinfos {
                        globalinfo := &snatglobalinfo.GlobalInfo{}
                        util.DeepCopyObj(*v, globalinfo)
                        glInfoCache[name] = append(glInfoCache[name], *globalinfo)
                }
        }
        cont.indexMutex.Unlock()
        snatglobalInfo, err := util.GetGlobalInfoCR(*globalcl)
        if errors.IsNotFound(err) {
                spec := snatglobalinfo.SnatGlobalInfoSpec{
                        GlobalInfos: glInfoCache,
                }
                if globalcl != nil {
                        err := util.CreateSnatGlobalInfoCR(*globalcl, spec)
                        if err != nil {
                                cont.log.Info("Create failed requeue the request", err)
                                return true
                        }
                }
                return false
        } else if err != nil {
                cont.log.Info("Create failed requeue the request-1: ", err)
                return true
        }
        if reflect.DeepEqual(snatglobalInfo.Spec.GlobalInfos, glInfoCache) {
                return false
        }
        snatglobalInfo.Spec.GlobalInfos = glInfoCache
        cont.log.Info("Update GlobalInfo: ", glInfoCache)
        err = util.UpdateGlobalInfoCR(*globalcl, snatglobalInfo)
        if err != nil {
                cont.log.Info("Update Failed: ", err)
                return true
        }
        return false
}

func (cont *AciController) updateGlobalInfoforPolicy(portrange snatglobalinfo.PortRange,
        snatIp, nodename, macaddr, plcyname string) {
        portlist := []snatglobalinfo.PortRange{}
        portlist = append(portlist, portrange)
        ip := net.ParseIP(snatIp)
        snatIpUuid, _ := uuid.FromBytes(ip)
        cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)
        cont.indexMutex.Lock()
        glinfo := &snatglobalinfo.GlobalInfo{
                MacAddress:     macaddr,
                PortRanges:     portlist,
                SnatIp:         snatIp,
                SnatIpUid:      snatIpUuid.String(),
                SnatPolicyName: plcyname,
        }
        if _, ok := cont.snatGlobalInfoCache[snatIp]; !ok {
                cont.snatGlobalInfoCache[snatIp] = make(map[string]*snatglobalinfo.GlobalInfo)
        }
        cont.snatGlobalInfoCache[snatIp][nodename] = glinfo
        cont.indexMutex.Unlock()
        cont.log.Debug("Node name and globalinfo: ", nodename, glinfo)
}

func (cont *AciController) getIpAndPortRange(nodename string, snatpolicy *ContSnatPolicy, serviceIp string) (string,
        snatglobalinfo.PortRange, bool) {
        expandedsnatports := snatpolicy.ExpandedSnatPorts
        if len(snatpolicy.SnatIp) != 0 {
                snatIps := snatpolicy.ExpandedSnatIps
                if len(snatIps) == 0 {
                        return "", snatglobalinfo.PortRange{}, false
                }
                return cont.allocateIpSnatPortRange(snatIps, nodename, expandedsnatports)
        } else {
                var snatIps []string
                snatIps = append(snatIps, serviceIp)
                return cont.allocateIpSnatPortRange(snatIps, nodename, expandedsnatports)
        }
        return "", snatglobalinfo.PortRange{}, false
}

func (cont *AciController) allocateIpSnatPortRange(snatIps []string, nodename string,
        expandedsnatports []snatglobalinfo.PortRange) (string, snatglobalinfo.PortRange, bool) {
        if len(expandedsnatports) < 1 {
                return "", snatglobalinfo.PortRange{}, false
        }
        for _, snatip := range snatIps {
                cont.indexMutex.Lock()
                globalInfo, ok := cont.snatGlobalInfoCache[snatip]
                if !ok {
                        cont.indexMutex.Unlock()
                        return snatip, expandedsnatports[0], true
                } else if len(globalInfo) < len(expandedsnatports) {
                        if _, ok := globalInfo[nodename]; !ok {
                                seen := make(map[int]int)
                                for _, val := range globalInfo {
                                        seen[val.PortRanges[0].Start] = val.PortRanges[0].End
                                }
                                for _, val := range expandedsnatports {
                                        if _, ok := seen[val.Start]; !ok {
                                                cont.indexMutex.Unlock()
                                                return snatip, val, true
                                        }
                                }
                        } else {
                                ip := globalInfo[nodename].SnatIp
                                portrange := &snatglobalinfo.PortRange{}
                                util.DeepCopyObj(globalInfo[nodename].PortRanges[0], portrange)
                                cont.indexMutex.Unlock()
                                return ip, *portrange, true
                        }
                }
                cont.indexMutex.Unlock()
        }
        return "", snatglobalinfo.PortRange{}, false
}

func (cont *AciController) deleteNodeinfoFromGlInfoCache(nodename string) bool {
        cont.indexMutex.Lock()
        defer cont.indexMutex.Unlock()
        for snatip, glinfos := range cont.snatGlobalInfoCache {
                if v, ok := glinfos[nodename]; ok {
                        if cont.checksnatPolicyPortExhausted(v.SnatPolicyName) {
                                if cont.setSnatPolicyStaus(v.SnatPolicyName, snatv1.Ready) == true {
                                        return true
                                }
                        }
                        delete(glinfos, nodename)
                        if len(glinfos) == 0 {
                                delete(cont.snatGlobalInfoCache, snatip)
                        }
                }
        }
        return false
}

func (cont *AciController) getServiceIps(policy *ContSnatPolicy) (serviceIps []string) {
        services := cont.getServicesBySelector(labels.SelectorFromSet(
                labels.Set(policy.Selector.Labels)),
                policy.Selector.Namespace)
        for _, service := range services {
                serviceIps = append(serviceIps, service.Status.LoadBalancer.Ingress[0].IP)
        }
        return
}

func (cont *AciController) updateSnatIpandPorts(oldPolicyNames map[string]bool,
        newPolicynames map[string]bool, nodename string) {
        for oldkey := range oldPolicyNames {
                if _, ok := newPolicynames[oldkey]; !ok {
                        cont.clearSnatGlobalCache(oldkey, nodename)
                }
        }
}

func (cont *AciController) clearSnatGlobalCache(policyName string, nodename string) {
        var expandedIps []string
        contSnatPolicy, ok := cont.snatPolicyCache[policyName]
        if !ok {
                return
        }
        if len(contSnatPolicy.SnatIp) > 0 {
                expandedIps = contSnatPolicy.ExpandedSnatIps
        } else {
                expandedIps = cont.getServiceIps(contSnatPolicy)
        }
        for _, snatip := range expandedIps {
                if v, ok := cont.snatGlobalInfoCache[snatip]; ok {
                        if len(nodename) > 0 {
                                if _, exists := v[nodename]; exists {
                                        delete(v, nodename)
                                        if len(v) == 0 {
                                                delete(cont.snatGlobalInfoCache, snatip)
                                        }
                                        break
                                }
                        } else {
                                delete(cont.snatGlobalInfoCache, snatip)
                        }
                } else {
                        break
                }
        }
}

func (cont *AciController) getNodeInfoKeys(policyName string, nodeinfokeys map[string]bool) {
        for _, nodeinfo := range cont.snatNodeInfoCache {
                if _, ok := nodeinfo.Spec.SnatPolicyNames[policyName]; ok {
                        nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)
                        if err != nil {
                                continue
                        }
                        nodeinfokeys[nodeinfokey] = true
                }
        }
}

func (cont *AciController) setSnatPoliciesState(names map[string]bool, status snatv1.PolicyState) bool {
        // Any alloc failures mark the policy with Status IpPortsExhausted
        ret := false
        for name := range names {
                if cont.setSnatPolicyStaus(name, status) == true {
                        ret = true
                }
        }
        return ret
}

1	// Copyright 2019 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
12	// implied.
13	// See the License for the specific language governing permissions and
14	// limitations under the License.
15
16	package controller
17
18	import (
19	"context"
20	uuid "github.com/google/uuid"
21	nodeinfo "github.com/noironetworks/aci-containers/pkg/nodeinfo/apis/aci.snat/v1"
22	nodeinfoclset "github.com/noironetworks/aci-containers/pkg/nodeinfo/clientset/versioned"
23	snatglobalinfo "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/apis/aci.snat/v1"
24	snatv1 "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
25	"github.com/noironetworks/aci-containers/pkg/util"
26	"k8s.io/api/core/v1"
27	"k8s.io/apimachinery/pkg/api/errors"
28	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
29	"k8s.io/apimachinery/pkg/fields"
30	"k8s.io/apimachinery/pkg/labels"
31	"k8s.io/apimachinery/pkg/runtime"
32	"k8s.io/apimachinery/pkg/watch"
33	"k8s.io/client-go/kubernetes"
34	"k8s.io/client-go/tools/cache"
35	"net"
36	"reflect"
37	"strconv"
38	)
39
40	type set map[string]bool
41
42	func (cont *AciController) initSnatNodeInformerFromClient(
43	snatClient *nodeinfoclset.Clientset) {	×
44	cont.initSnatNodeInformerBase(	×
45	&cache.ListWatch{	×
46	ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {	×
47	return snatClient.AciV1().NodeInfos(metav1.NamespaceAll).List(context.TODO(), options)	×
48	},	×
49	WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {	×
50	return snatClient.AciV1().NodeInfos(metav1.NamespaceAll).Watch(context.TODO(), options)	×
51	},	×
52	})
53	}
54
55	func (cont AciController) initSnatNodeInformerBase(listWatch cache.ListWatch) {	1✔
56	cont.snatNodeInfoIndexer, cont.snatNodeInformer = cache.NewIndexerInformer(	1✔
57	listWatch,	1✔
58	&nodeinfo.NodeInfo{}, 0,	1✔
59	cache.ResourceEventHandlerFuncs{	1✔
60	AddFunc: func(obj interface{}) {	2✔
61	cont.snatNodeInfoAdded(obj)	1✔
62	},	1✔
63	UpdateFunc: func(oldiobj interface{}, newobj interface{}) {	1✔
64	cont.snatNodeInfoUpdated(oldiobj, newobj)	1✔
65	},	1✔
66	DeleteFunc: func(obj interface{}) {	×
67	cont.snatNodeInfoDeleted(obj)	×
68	},	×
69	},
70	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
71	)
72	cont.log.Debug("Initializing Node Informers: ")	1✔
73	}
74
75	func (cont *AciController) initSnatCfgFromClient(
76	kubeClient kubernetes.Interface) {	×
77	cont.initSnatCfgInformerBase(	×
78	&cache.ListWatch{	×
79	ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {	×
80	options.FieldSelector =	×
81	fields.Set{"metadata.name": "snat-operator-config"}.String()	×
82	return kubeClient.CoreV1().ConfigMaps(	×
83	"aci-containers-system").List(context.TODO(), options)	×
84	},	×
85	WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {	×
86	options.FieldSelector =	×
87	fields.Set{"metadata.name": "snat-operator-config"}.String()	×
88	return kubeClient.CoreV1().ConfigMaps(	×
89	"aci-containers-system").Watch(context.TODO(), options)	×
90	},	×
91	})
92	}
93
94	func (cont AciController) initSnatCfgInformerBase(listWatch cache.ListWatch) {	1✔
95	_, cont.snatCfgInformer = cache.NewIndexerInformer(	1✔
96	listWatch,	1✔
97	&v1.ConfigMap{}, 0,	1✔
98	cache.ResourceEventHandlerFuncs{	1✔
99	UpdateFunc: func(_, obj interface{}) {	2✔
100	cont.snatCfgUpdate(obj)	1✔
101	},	1✔
102	},
103	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
104	)
105	cont.log.Debug("Initializing SnatCfg Informers: ")	1✔
106	}
107
108	// Handle any changes to snatOperator Config
109	func (cont *AciController) snatCfgUpdate(obj interface{}) {	1✔
110	snatcfg := obj.(*v1.ConfigMap)	1✔
111	var portRange snatglobalinfo.PortRange	1✔
112	cont.log.Debug("snatCfgUpdated: ", snatcfg)	1✔
113	data := snatcfg.Data	1✔
114	start, err1 := strconv.Atoi(data["start"])	1✔
115	end, err2 := strconv.Atoi(data["end"])	1✔
116	portsPerNode, err3 := strconv.Atoi(data["ports-per-node"])	1✔
117	if err1 != nil \|\| err2 != nil \|\| err3 != nil \|\|	1✔
118	start < 5000 \|\| end > 65000 \|\| start > end \|\| portsPerNode > end-start+1 {	1✔
119	return	×
120	}	×
121	portRange.Start = start	1✔
122	portRange.End = end	1✔
123	var currPortRange []snatglobalinfo.PortRange	1✔
124	currPortRange = append(currPortRange, portRange)	1✔
125	cont.indexMutex.Lock()	1✔
126	nodeInfoKeys := make(map[string]bool)	1✔
127	for name, info := range cont.snatPolicyCache {	2✔
128	cont.clearSnatGlobalCache(name, "")	1✔
129	info.ExpandedSnatPorts = util.ExpandPortRanges(currPortRange, portsPerNode)	1✔
130	cont.getNodeInfoKeys(name, nodeInfoKeys)	1✔
131	}	1✔
132	cont.indexMutex.Unlock()	1✔
133	for key := range nodeInfoKeys {	2✔
134	cont.queueNodeInfoUpdateByKey(key)	1✔
135	}	1✔
136	}
137
138	func (cont *AciController) snatNodeInfoAdded(obj interface{}) {	1✔
139	nodeinfo := obj.(*nodeinfo.NodeInfo)	1✔
140	nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)	1✔
141	if err != nil {	1✔
142	return	×
143	}	×
144	cont.log.Debug("Node Info Added: ", nodeinfokey)	1✔
145	cont.indexMutex.Lock()	1✔
146	cont.snatNodeInfoCache[nodeinfo.ObjectMeta.Name] = nodeinfo	1✔
147	cont.indexMutex.Unlock()	1✔
148	cont.queueNodeInfoUpdateByKey(nodeinfokey)	1✔
149	}
150
151	func (cont *AciController) queueNodeInfoUpdateByKey(key string) {	1✔
152	cont.log.Debug("Node key Queued: ", key)	1✔
153	cont.snatNodeInfoQueue.Add(key)	1✔
154	}	1✔
155
156	func (cont *AciController) snatNodeInfoUpdated(oldobj interface{}, newobj interface{}) {	1✔
157	oldnodeinfo := oldobj.(*nodeinfo.NodeInfo)	1✔
158	newnodeinfo := newobj.(*nodeinfo.NodeInfo)	1✔
159	nodeinfokey, err := cache.MetaNamespaceKeyFunc(newnodeinfo)	1✔
160	if err != nil {	1✔
161	return	×
162	}	×
163	if reflect.DeepEqual(oldnodeinfo.Spec, newnodeinfo.Spec) {	1✔
164	return	×
165	}	×
166	cont.indexMutex.Lock()	1✔
167	cont.updateSnatIpandPorts(oldnodeinfo.Spec.SnatPolicyNames,	1✔
168	newnodeinfo.Spec.SnatPolicyNames, newnodeinfo.ObjectMeta.Name)	1✔
169	cont.snatNodeInfoCache[newnodeinfo.ObjectMeta.Name] = newnodeinfo	1✔
170	cont.indexMutex.Unlock()	1✔
171	cont.queueNodeInfoUpdateByKey(nodeinfokey)	1✔
172	}
173
174	func (cont *AciController) snatNodeInfoDeleted(obj interface{}) {	×
175	nodeinfo := obj.(*nodeinfo.NodeInfo)	×
176	nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)	×
177	if err != nil {	×
178	return	×
179	}	×
180	cont.indexMutex.Lock()	×
181	delete(cont.snatNodeInfoCache, nodeinfo.ObjectMeta.Name)	×
182	cont.indexMutex.Unlock()	×
183	cont.queueNodeInfoUpdateByKey(nodeinfokey)	×
184	}
185
186	func (cont *AciController) setSnatPolicyStaus(snatPolicyName string, status snatv1.PolicyState) bool {	×
187	obj, exists, err := cont.snatIndexer.GetByKey(snatPolicyName)	×
188	if err == nil && exists && obj != nil {	×
189	snatpolicy := obj.(*snatv1.SnatPolicy)	×
190	if snatpolicy.Status.State != status {	×
191	snatpolicy.Status.State = status	×
192	env := cont.env.(*K8sEnvironment)	×
193	policycl := env.snatClient	×
194	if policycl != nil {	×
195	err = util.UpdateSnatPolicyCR(*policycl, snatpolicy)	×
196	if err != nil {	×
197	cont.log.Info("Policy status update failed queue the request again: ", err)	×
198	return true	×
199	}	×
200	}
201	}
202	}
203	return false	×
204	}
205
206	func (cont *AciController) isSnatNodeInfoPresent(nodeName string) bool {	1✔
207	cont.indexMutex.Lock()	1✔
208	_, ok := cont.snatNodeInfoCache[nodeName]	1✔
209	cont.indexMutex.Unlock()	1✔
210	return ok	1✔
211	}	1✔
212
213	func (cont *AciController) checksnatPolicyPortExhausted(name string) bool {	1✔
214	snatobj, exists, err := cont.snatIndexer.GetByKey(name)	1✔
215	if err == nil && exists && snatobj != nil {	2✔
216	snatpolicy := snatobj.(*snatv1.SnatPolicy)	1✔
217	if snatpolicy.Status.State == snatv1.IpPortsExhausted {	1✔
218	return true	×
219	}	×
220	}
221	return false	1✔
222	}
223
224	func (cont AciController) handleSnatNodeInfo(nodeinfo nodeinfo.NodeInfo) bool {	1✔
225	cont.log.Debug("handle Node Info: ", nodeinfo)	1✔
226	updated := false	1✔
227	nodename := nodeinfo.ObjectMeta.Name	1✔
228	ret := false	1✔
229	// Cache needs to be updated	1✔
230	if !cont.isSnatNodeInfoPresent(nodename) \|\| len(nodeinfo.Spec.SnatPolicyNames) == 0 {	1✔
231	ret = cont.deleteNodeinfoFromGlInfoCache(nodename)	×
232	updated = true	×
233	} else {	1✔
234	// This case ignores any stale entry is present in nodeinfo	1✔
235	_, _, err := cont.nodeIndexer.GetByKey(nodename)	1✔
236	if err != nil {	1✔
237	cont.log.Info("Could not lookup node: ", err, "nodeName: ", nodename)	×
238	return false	×
239	}	×
240	allocfailed := make(map[string]bool)	1✔
241	markready := make(map[string]bool)	1✔
242	for name := range nodeinfo.Spec.SnatPolicyNames {	2✔
243	cont.indexMutex.Lock()	1✔
244	snatpolicy, ok := cont.snatPolicyCache[name]	1✔
245	cont.indexMutex.Unlock()	1✔
246	if !ok {	1✔
247	continue	×
248	}
249	cont.log.Debug("SnatPolicy Name: ", name)	1✔
250	if len(snatpolicy.SnatIp) != 0 {	2✔
251	snatIp, portrange, alloc := cont.getIpAndPortRange(nodename, snatpolicy, "")	1✔
252	cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)	1✔
253	if alloc == false {	1✔
254	cont.log.Error("Port Range Exhausted: ", name)	×
255	allocfailed[name] = true	×
256	continue	×
257	} else {	1✔
258	if cont.checksnatPolicyPortExhausted(name) {	1✔
259	markready[name] = true	×
260	}	×
261	}
262	cont.updateGlobalInfoforPolicy(portrange, snatIp, nodename,	1✔
263	nodeinfo.Spec.Macaddress, name)	1✔
264	updated = true	1✔
265	} else {	×
266	snatIps := cont.getServiceIps(snatpolicy)	×
267	cont.log.Debug("Service Ips: ", snatIps)	×
268	for _, snatip := range snatIps {	×
269	snatIp, portrange, alloc := cont.getIpAndPortRange(nodename, snatpolicy, snatip)	×
270	cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)	×
271	if alloc == false {	×
272	cont.log.Error("Port Range Exhausted: ", name)	×
273	allocfailed[name] = true	×
274	continue	×
275	} else {	×
276	if cont.checksnatPolicyPortExhausted(name) {	×
277	markready[name] = true	×
278	}	×
279	}
280	cont.updateGlobalInfoforPolicy(portrange, snatIp, nodename,	×
281	nodeinfo.Spec.Macaddress, name)	×
282	updated = true	×
283	}
284	}
285	}
286	ret = cont.setSnatPoliciesState(allocfailed, snatv1.IpPortsExhausted)	1✔
287	ret = cont.setSnatPoliciesState(markready, snatv1.Ready)	1✔
288	}
289
290	if updated {	2✔
291	cont.scheduleSyncGlobalInfo()	1✔
292	}	1✔
293	return ret	1✔
294	}
295
296	func (cont *AciController) syncSnatGlobalInfo() bool {	1✔
297	env := cont.env.(*K8sEnvironment)	1✔
298	globalcl := env.snatGlobalClient	1✔
299	if globalcl == nil {	2✔
300	return false	1✔
301	}	1✔
302	cont.indexMutex.Lock()	×
303	cont.log.Debug("syncSnatGlobalInfo")	×
304	glInfoCache := make(map[string]snatglobalinfo.GlobalInfoList)	×
305	for _, glinfos := range cont.snatGlobalInfoCache {	×
306	for name, v := range glinfos {	×
307	globalinfo := &snatglobalinfo.GlobalInfo{}	×
308	util.DeepCopyObj(*v, globalinfo)	×
309	glInfoCache[name] = append(glInfoCache[name], *globalinfo)	×
310	}	×
311	}
312	cont.indexMutex.Unlock()	×
313	snatglobalInfo, err := util.GetGlobalInfoCR(*globalcl)	×
314	if errors.IsNotFound(err) {	×
315	spec := snatglobalinfo.SnatGlobalInfoSpec{	×
316	GlobalInfos: glInfoCache,	×
317	}	×
318	if globalcl != nil {	×
319	err := util.CreateSnatGlobalInfoCR(*globalcl, spec)	×
320	if err != nil {	×
321	cont.log.Info("Create failed requeue the request", err)	×
322	return true	×
323	}	×
324	}
325	return false	×
326	} else if err != nil {	×
327	cont.log.Info("Create failed requeue the request-1: ", err)	×
328	return true	×
329	}	×
330	if reflect.DeepEqual(snatglobalInfo.Spec.GlobalInfos, glInfoCache) {	×
331	return false	×
332	}	×
333	snatglobalInfo.Spec.GlobalInfos = glInfoCache	×
334	cont.log.Info("Update GlobalInfo: ", glInfoCache)	×
335	err = util.UpdateGlobalInfoCR(*globalcl, snatglobalInfo)	×
336	if err != nil {	×
337	cont.log.Info("Update Failed: ", err)	×
338	return true	×
339	}	×
340	return false	×
341	}
342
343	func (cont *AciController) updateGlobalInfoforPolicy(portrange snatglobalinfo.PortRange,
344	snatIp, nodename, macaddr, plcyname string) {	1✔
345	portlist := []snatglobalinfo.PortRange{}	1✔
346	portlist = append(portlist, portrange)	1✔
347	ip := net.ParseIP(snatIp)	1✔
348	snatIpUuid, _ := uuid.FromBytes(ip)	1✔
349	cont.log.Debug("SnatIP and Port range: ", snatIp, portrange)	1✔
350	cont.indexMutex.Lock()	1✔
351	glinfo := &snatglobalinfo.GlobalInfo{	1✔
352	MacAddress: macaddr,	1✔
353	PortRanges: portlist,	1✔
354	SnatIp: snatIp,	1✔
355	SnatIpUid: snatIpUuid.String(),	1✔
356	SnatPolicyName: plcyname,	1✔
357	}	1✔
358	if _, ok := cont.snatGlobalInfoCache[snatIp]; !ok {	2✔
359	cont.snatGlobalInfoCache[snatIp] = make(map[string]*snatglobalinfo.GlobalInfo)	1✔
360	}	1✔
361	cont.snatGlobalInfoCache[snatIp][nodename] = glinfo	1✔
362	cont.indexMutex.Unlock()	1✔
363	cont.log.Debug("Node name and globalinfo: ", nodename, glinfo)	1✔
364	}
365
366	func (cont AciController) getIpAndPortRange(nodename string, snatpolicy ContSnatPolicy, serviceIp string) (string,
367	snatglobalinfo.PortRange, bool) {	1✔
368	expandedsnatports := snatpolicy.ExpandedSnatPorts	1✔
369	if len(snatpolicy.SnatIp) != 0 {	2✔
370	snatIps := snatpolicy.ExpandedSnatIps	1✔
371	if len(snatIps) == 0 {	1✔
372	return "", snatglobalinfo.PortRange{}, false	×
373	}	×
374	return cont.allocateIpSnatPortRange(snatIps, nodename, expandedsnatports)	1✔
375	} else {	×
376	var snatIps []string	×
377	snatIps = append(snatIps, serviceIp)	×
378	return cont.allocateIpSnatPortRange(snatIps, nodename, expandedsnatports)	×
379	}	×
380	return "", snatglobalinfo.PortRange{}, false	×
381	}
382
383	func (cont *AciController) allocateIpSnatPortRange(snatIps []string, nodename string,
384	expandedsnatports []snatglobalinfo.PortRange) (string, snatglobalinfo.PortRange, bool) {	1✔
385	if len(expandedsnatports) < 1 {	1✔
386	return "", snatglobalinfo.PortRange{}, false	×
387	}	×
388	for _, snatip := range snatIps {	2✔
389	cont.indexMutex.Lock()	1✔
390	globalInfo, ok := cont.snatGlobalInfoCache[snatip]	1✔
391	if !ok {	2✔
392	cont.indexMutex.Unlock()	1✔
393	return snatip, expandedsnatports[0], true	1✔
394	} else if len(globalInfo) < len(expandedsnatports) {	3✔
395	if _, ok := globalInfo[nodename]; !ok {	2✔
396	seen := make(map[int]int)	1✔
397	for _, val := range globalInfo {	2✔
398	seen[val.PortRanges[0].Start] = val.PortRanges[0].End	1✔
399	}	1✔
400	for _, val := range expandedsnatports {	2✔
401	if _, ok := seen[val.Start]; !ok {	2✔
402	cont.indexMutex.Unlock()	1✔
403	return snatip, val, true	1✔
404	}	1✔
405	}
406	} else {	1✔
407	ip := globalInfo[nodename].SnatIp	1✔
408	portrange := &snatglobalinfo.PortRange{}	1✔
409	util.DeepCopyObj(globalInfo[nodename].PortRanges[0], portrange)	1✔
410	cont.indexMutex.Unlock()	1✔
411	return ip, *portrange, true	1✔
412	}	1✔
413	}
414	cont.indexMutex.Unlock()	×
415	}
416	return "", snatglobalinfo.PortRange{}, false	×
417	}
418
419	func (cont *AciController) deleteNodeinfoFromGlInfoCache(nodename string) bool {	×
420	cont.indexMutex.Lock()	×
421	defer cont.indexMutex.Unlock()	×
422	for snatip, glinfos := range cont.snatGlobalInfoCache {	×
423	if v, ok := glinfos[nodename]; ok {	×
424	if cont.checksnatPolicyPortExhausted(v.SnatPolicyName) {	×
425	if cont.setSnatPolicyStaus(v.SnatPolicyName, snatv1.Ready) == true {	×
426	return true	×
427	}	×
428	}
429	delete(glinfos, nodename)	×
430	if len(glinfos) == 0 {	×
431	delete(cont.snatGlobalInfoCache, snatip)	×
432	}	×
433	}
434	}
435	return false	×
436	}
437
438	func (cont AciController) getServiceIps(policy ContSnatPolicy) (serviceIps []string) {	×
439	services := cont.getServicesBySelector(labels.SelectorFromSet(	×
440	labels.Set(policy.Selector.Labels)),	×
441	policy.Selector.Namespace)	×
442	for _, service := range services {	×
443	serviceIps = append(serviceIps, service.Status.LoadBalancer.Ingress[0].IP)	×
444	}	×
445	return	×
446	}
447
448	func (cont *AciController) updateSnatIpandPorts(oldPolicyNames map[string]bool,
449	newPolicynames map[string]bool, nodename string) {	1✔
450	for oldkey := range oldPolicyNames {	2✔
451	if _, ok := newPolicynames[oldkey]; !ok {	1✔
452	cont.clearSnatGlobalCache(oldkey, nodename)	×
453	}	×
454	}
455	}
456
457	func (cont *AciController) clearSnatGlobalCache(policyName string, nodename string) {	1✔
458	var expandedIps []string	1✔
459	contSnatPolicy, ok := cont.snatPolicyCache[policyName]	1✔
460	if !ok {	1✔
461	return	×
462	}	×
463	if len(contSnatPolicy.SnatIp) > 0 {	2✔
464	expandedIps = contSnatPolicy.ExpandedSnatIps	1✔
465	} else {	1✔
466	expandedIps = cont.getServiceIps(contSnatPolicy)	×
467	}	×
468	for _, snatip := range expandedIps {	2✔
469	if v, ok := cont.snatGlobalInfoCache[snatip]; ok {	2✔
470	if len(nodename) > 0 {	1✔
471	if _, exists := v[nodename]; exists {	×
472	delete(v, nodename)	×
473	if len(v) == 0 {	×
474	delete(cont.snatGlobalInfoCache, snatip)	×
475	}	×
476	break	×
477	}
478	} else {	1✔
479	delete(cont.snatGlobalInfoCache, snatip)	1✔
480	}	1✔
481	} else {	1✔
482	break	1✔
483	}
484	}
485	}
486
487	func (cont *AciController) getNodeInfoKeys(policyName string, nodeinfokeys map[string]bool) {	1✔
488	for _, nodeinfo := range cont.snatNodeInfoCache {	2✔
489	if _, ok := nodeinfo.Spec.SnatPolicyNames[policyName]; ok {	2✔
490	nodeinfokey, err := cache.MetaNamespaceKeyFunc(nodeinfo)	1✔
491	if err != nil {	1✔
492	continue	×
493	}
494	nodeinfokeys[nodeinfokey] = true	1✔
495	}
496	}
497	}
498
499	func (cont *AciController) setSnatPoliciesState(names map[string]bool, status snatv1.PolicyState) bool {	1✔
500	// Any alloc failures mark the policy with Status IpPortsExhausted	1✔
501	ret := false	1✔
502	for name := range names {	1✔
503	if cont.setSnatPolicyStaus(name, status) == true {	×
504	ret = true	×
505	}	×
506	}
507	return ret	1✔
508	}

noironetworks / aci-containers / 4718

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