5715

Committed 10 Sep 2021 06:59PM UTC coverage: 58.51%. First build

Build # 5715

Build Type

push

travis-ci-com

Committed by Tanya Tukade

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/9c1339cb8395d4f5744b10193f9561ac45da46bd">9c1339cb8">&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/9c1339cb8395d4f5744b10193f9561ac45da46bd">9c1339cb8</a><a href="https://github.com/noironetworks/aci-containers/commit/9c1339cb8395d4f5744b10193f9561ac45da46bd">&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/9c1339cb8395d4f5744b10193f9561ac45da46bd&quot;&gt;9c1339cb8&lt;/a&gt;&lt;a href=&quot;https://github.com/noironetworks/aci-containers/commit/9c1339cb8395d4f5744b10193f9561ac45da46bd&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/9c1339cb8395d4f5744b10193f9561ac45da46bd&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/noironetworks/aci-containers/commit/9c1339cb8395d4f5744b10193f9561a... (continued)

Run Details

51 of 51 new or added lines in 3 files covered. (100.0%)

11314 of 19337 relevant lines covered (58.51%)

0.65 hits per line

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

40.77

/pkg/controller/snats.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 (
        "fmt"
        nodeinfo "github.com/noironetworks/aci-containers/pkg/nodeinfo/apis/aci.snat/v1"
        snatglobalinfo "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/apis/aci.snat/v1"
        snatpolicy "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
        snatclientset "github.com/noironetworks/aci-containers/pkg/snatpolicy/clientset/versioned"
        "github.com/noironetworks/aci-containers/pkg/util"
        "github.com/sirupsen/logrus"
        v1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/client-go/tools/cache"
        "net"
        "reflect"
)

const snatGraphName = "svcgraph"

type ContPodSelector struct {
        Labels    map[string]string
        Namespace string
}

type ContPortRange struct {
        Start int `json:"start,omitempty"`
        End   int `json:"end,omitempty"`
}

type ContSnatPolicy struct {
        SnatIp            []string
        Selector          ContPodSelector
        PortRange         []ContPortRange
        Protocols         []string
        ExpandedSnatIps   []string
        ExpandedSnatPorts []snatglobalinfo.PortRange
}

func SnatPolicyLogger(log *logrus.Logger, snat *snatpolicy.SnatPolicy) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "namespace": snat.ObjectMeta.Namespace,
                "name":      snat.ObjectMeta.Name,
                "spec":      snat.Spec,
        })
}

func (cont *AciController) initSnatInformerFromClient(
        snatClient *snatclientset.Clientset) {
        cont.initSnatInformerBase(
                cache.NewListWatchFromClient(
                        snatClient.AciV1().RESTClient(), "snatpolicies",
                        metav1.NamespaceAll, fields.Everything()))
}

func (cont *AciController) initSnatInformerBase(listWatch *cache.ListWatch) {
        cont.snatIndexer, cont.snatInformer = cache.NewIndexerInformer(
                listWatch,
                &snatpolicy.SnatPolicy{}, 0,
                cache.ResourceEventHandlerFuncs{
                        AddFunc: func(obj interface{}) {
                                cont.snatPolicyUpdated(obj)
                        },
                        UpdateFunc: func(_, obj interface{}) {
                                cont.snatPolicyUpdated(obj)
                        },
                        DeleteFunc: func(obj interface{}) {
                                cont.snatPolicyDelete(obj)
                        },
                },
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        cont.log.Debug("Initializing Snat Policy Informers")

}

func (cont *AciController) snatPolicyUpdated(obj interface{}) {
        snatPolicy := obj.(*snatpolicy.SnatPolicy)
        key, err := cache.MetaNamespaceKeyFunc(snatPolicy)
        if err != nil {
                SnatPolicyLogger(cont.log, snatPolicy).
                        Error("Could not create key:" + err.Error())
                return
        }
        cont.queueSnatUpdateByKey(key)

}

func (cont *AciController) queueSnatUpdateByKey(key string) {
        cont.log.Info("Added snatpolicy key: ", key)
        cont.snatQueue.Add(key)
}

func (cont *AciController) queueSnatUpdate(snatpolicy *snatpolicy.SnatPolicy) {
        key, err := cache.MetaNamespaceKeyFunc(snatpolicy)
        if err != nil {
                SnatPolicyLogger(cont.log, snatpolicy).
                        Error("Could not create key:" + err.Error())
                return
        }
        cont.log.Info("Add snatpolicy key: ", key)
        cont.snatQueue.Add(key)
}

func (cont *AciController) handleSnatUpdate(snatPolicy *snatpolicy.SnatPolicy) bool {
        _, err := cache.MetaNamespaceKeyFunc(snatPolicy)
        if err != nil {
                SnatPolicyLogger(cont.log, snatPolicy).
                        Error("Could not create key:" + err.Error())
                return false
        }
        cont.log.Info("Handle update for snatpolicy: ", snatPolicy.ObjectMeta.Name)
        policyName := snatPolicy.ObjectMeta.Name
        if snatPolicy.Status.State != snatpolicy.Ready {
                if snatPolicy.Status.State == snatpolicy.IpPortsExhausted {
                        cont.indexMutex.Lock()
                        if snatInfo, ok := cont.snatPolicyCache[policyName]; ok {
                                if reflect.DeepEqual(snatPolicy.Spec.SnatIp, snatInfo.SnatIp) {
                                        cont.indexMutex.Unlock()
                                        return false
                                }
                        }
                        cont.indexMutex.Unlock()
                }
                if status, err := cont.validateCr(snatPolicy); !status {
                        cont.log.Error("SnatPolicy Failed: ", err)
                        return cont.setSnatPolicyStaus(policyName, snatpolicy.Failed)
                }
                return cont.setSnatPolicyStaus(policyName, snatpolicy.Ready)
        }
        cont.updateSnatPolicyCache(policyName, snatPolicy)
        var requeue bool
        cont.indexMutex.Lock()
        if cont.snatSyncEnabled {
                cont.indexMutex.Unlock()

                if len(snatPolicy.Spec.SnatIp) == 0 {
                        err = cont.handleSnatPolicyForServices(snatPolicy)
                } else {
                        err = cont.updateServiceDeviceInstanceSnat(snatGraphName)
                }
                if err != nil {
                        requeue = true
                }
        } else {
                cont.indexMutex.Unlock()
        }
        return requeue
}

func (cont *AciController) getServicesBySelector(selector labels.Selector, ns string) []*v1.Service {
        var services []*v1.Service
        cache.ListAll(cont.serviceIndexer, selector,
                func(servobj interface{}) {
                        service := servobj.(*v1.Service)
                        if len(ns) == 0 || (len(ns) > 0 && ns == service.ObjectMeta.Namespace) {
                                if len(service.Status.LoadBalancer.Ingress) != 0 {
                                        services = append(services, service)
                                }
                        }
                })
        return services
}

func (cont *AciController) handleSnatPolicyForServices(snatpolicy *snatpolicy.SnatPolicy) error {
        ServiceList := cont.getServicesBySelector(labels.SelectorFromSet(
                labels.Set(snatpolicy.Spec.Selector.Labels)),
                snatpolicy.Spec.Selector.Namespace)

        if len(ServiceList) == 0 {
                return nil
        }
        for _, service := range ServiceList {
                servicekey, err := cache.MetaNamespaceKeyFunc(service)
                if err != nil {
                        servicekey = service.ObjectMeta.Namespace + "/" + service.ObjectMeta.Name
                }
                cont.indexMutex.Lock()
                if service.GetDeletionTimestamp() == nil {
                        cont.snatServices[servicekey] = true
                        cont.queueServiceUpdateByKey(servicekey)
                }
                cont.indexMutex.Unlock()
        }
        return nil
}

func (cont *AciController) updateSnatPolicyCache(key string, snatpolicy *snatpolicy.SnatPolicy) {
        cont.indexMutex.Lock()
        cont.log.Debug("Updating snatpolicy cache for policy: ", snatpolicy.ObjectMeta.Name)
        var policy ContSnatPolicy
        env := cont.env.(*K8sEnvironment)
        policy.SnatIp = snatpolicy.Spec.SnatIp
        policy.ExpandedSnatIps = util.ExpandCIDRs(policy.SnatIp)
        portRange := snatglobalinfo.PortRange{Start: 5000, End: 65000}
        portsPerNode := 3000
        if env.kubeClient != nil {
                portRange, portsPerNode = util.GetPortRangeFromConfigMap(env.kubeClient)
        }
        var currPortRange []snatglobalinfo.PortRange
        currPortRange = append(currPortRange, portRange)
        policy.ExpandedSnatPorts = util.ExpandPortRanges(currPortRange, portsPerNode)
        policy.Selector = ContPodSelector{Labels: snatpolicy.Spec.Selector.Labels, Namespace: snatpolicy.Spec.Selector.Namespace}
        Update := false
        if snatInfo, ok := cont.snatPolicyCache[key]; ok {
                cont.log.Debug("Snatpolicy already exists in cache: ", snatpolicy.ObjectMeta.Name)
                if !reflect.DeepEqual(policy.SnatIp, snatInfo.SnatIp) {
                        cont.log.Debug("Snatpolicy info found in cache marked to be updated: ", snatpolicy.ObjectMeta.Name)
                        cont.clearSnatGlobalCache(snatpolicy.ObjectMeta.Name, "")
                        Update = true
                }
        }
        cont.snatPolicyCache[key] = &policy
        nodeInfoKeys := make(map[string]bool)
        if Update {
                cont.getNodeInfoKeys(snatpolicy.ObjectMeta.Name, nodeInfoKeys)
        }
        cont.indexMutex.Unlock()
        for key := range nodeInfoKeys {
                cont.queueNodeInfoUpdateByKey(key)
        }
}

func (cont *AciController) snatPolicyDelete(snatobj interface{}) {
        snatpolicy := snatobj.(*snatpolicy.SnatPolicy)
        cont.log.Info("Snatpolicy marked for deletion: ", snatpolicy.ObjectMeta.Name)
        cont.indexMutex.Lock()
        cont.clearSnatGlobalCache(snatpolicy.ObjectMeta.Name, "")
        delete(cont.snatPolicyCache, snatpolicy.ObjectMeta.Name)

        if len(snatpolicy.Spec.SnatIp) == 0 {
                ServiceList := cont.getServicesBySelector(labels.SelectorFromSet(
                        labels.Set(snatpolicy.Spec.Selector.Labels)),
                        snatpolicy.Spec.Selector.Namespace)
                if len(ServiceList) > 0 {
                        for _, service := range ServiceList {
                                servicekey, err1 := cache.MetaNamespaceKeyFunc(service)
                                if err1 != nil {
                                        servicekey = service.ObjectMeta.Namespace + "/" + service.ObjectMeta.Name
                                }
                                delete(cont.snatServices, servicekey)
                                cont.queueServiceUpdateByKey(servicekey)
                        }
                }
        } else {
                if len(cont.snatPolicyCache) == 0 {
                        cont.log.Debug("No more snat policies, deleting graph")
                        graphName := cont.aciNameForKey("snat", snatGraphName)
                        go cont.apicConn.ClearApicObjects(graphName)
                } else {
                        go cont.updateServiceDeviceInstanceSnat(snatGraphName)
                }
        }
        cont.indexMutex.Unlock()
}

func (cont *AciController) createGlobalInfoCache(unittestmode bool) bool {
        cont.log.Info("Checking if snatglobalinfo CR already exists")

        var nodeInfos []*nodeinfo.NodeInfo
        cont.indexMutex.Lock()
        defer cont.indexMutex.Unlock()
        cache.ListAll(cont.snatNodeInfoIndexer, labels.Everything(),
                func(nodeInfoObj interface{}) {
                        nodeInfo := nodeInfoObj.(*nodeinfo.NodeInfo)
                        nodeInfos = append(nodeInfos, nodeInfo)
                })
        env := cont.env.(*K8sEnvironment)
        globalcl := env.snatGlobalClient
        if globalcl == nil {
                if unittestmode {
                        return false
                }
                cont.log.Fatalf("snatglobalinfo client not found")
        }
        snatglobalInfo, err := util.GetGlobalInfoCR(*globalcl)

        if err != nil {
                cont.log.Info("No existing snatglobalinfo CR found in controller bootstrap")
        } else {
                cont.log.Info("Syncing snatglobalinfo cache with existing CR")
                macAddressToNodeName := make(map[string]string)
                for _, value := range nodeInfos {
                        nodeName := value.ObjectMeta.Name
                        macAddress := value.Spec.Macaddress
                        macAddressToNodeName[macAddress] = nodeName
                }

                for _, glinfos := range snatglobalInfo.Spec.GlobalInfos {
                        for _, v := range glinfos {
                                nodeName := macAddressToNodeName[v.MacAddress]
                                snatIP := v.SnatIp
                                if _, ok := cont.snatGlobalInfoCache[snatIP]; !ok {
                                        cont.snatGlobalInfoCache[snatIP] = make(map[string]*snatglobalinfo.GlobalInfo)
                                }
                                cont.snatGlobalInfoCache[snatIP][nodeName] = &v
                        }
                }
        }
        return true
}

func (cont *AciController) snatFullSync() {
        cache.ListAll(cont.snatIndexer, labels.Everything(),
                func(sobj interface{}) {
                        cont.queueSnatUpdate(sobj.(*snatpolicy.SnatPolicy))
                })
}

func (cont *AciController) validateCr(cr *snatpolicy.SnatPolicy) (bool, string) {
        cont.indexMutex.Lock()
        snatPolicyCache := make(map[string]*ContSnatPolicy)
        for k, v := range cont.snatPolicyCache {
                snatPolicyCache[k] = v
        }
        cont.indexMutex.Unlock()
        if len(cont.snatPolicyCache) >= 1 {
                cr_labels := cr.Spec.Selector.Labels
                cr_ns := cr.Spec.Selector.Namespace
                for key, item := range snatPolicyCache {
                        if cr.ObjectMeta.Name != key {
                                if (len(item.Selector.Labels) == 0) && (len(cr_labels) == 0) &&
                                        (cr_ns == item.Selector.Namespace) {
                                        return false, fmt.Sprintf(
                                                "Namespace is conflicting with the snatpolicy %s",
                                                cr.ObjectMeta.Name)
                                }
                                for _, val := range item.SnatIp {
                                        _, net1, _ := parseIP(val)
                                        for _, ip := range cr.Spec.SnatIp {
                                                _, net2, err := parseIP(ip)
                                                if err != nil {
                                                        return false, fmt.Sprintf(
                                                                "Invalid incoming SnatIP %s", ip)
                                                }
                                                if net2.Contains(net1.IP) || net1.Contains(net2.IP) {
                                                        return false, fmt.Sprintf(
                                                                "SnatIP's are conflicting with the snatpolicy %s",
                                                                cr.ObjectMeta.Name)
                                                }
                                        }
                                }
                        }
                }
        } else {
                for _, ip := range cr.Spec.SnatIp {
                        _, _, err := parseIP(ip)
                        if err != nil {
                                return false, fmt.Sprintf(
                                        "Invalid incoming SnatIP %s", ip)
                        }
                }
        }
        for _, ip := range cr.Spec.DestIp {
                _, _, err := parseIP(ip)
                if err != nil {
                        return false, fmt.Sprintf(
                                "Invalid incoming DestIP %s", ip)
                }
        }

        return true, ""
}

func parseIP(cidr string) (net.IP, *net.IPNet, error) {
        ip, ipnet, err := net.ParseCIDR(cidr)
        if err != nil {
                ip_temp := net.ParseIP(cidr)
                if ip_temp != nil && ip_temp.To4() != nil {
                        cidr = cidr + "/32"
                        ip, ipnet, _ = net.ParseCIDR(cidr)
                        return ip, ipnet, nil
                } else if ip_temp != nil && ip_temp.To16() != nil {
                        cidr = cidr + "/128"
                        ip, ipnet, _ = net.ParseCIDR(cidr)
                        return ip, ipnet, nil
                } else {
                        return nil, nil, err
                }
        }
        return ip, ipnet, err
}

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	"fmt"
20	nodeinfo "github.com/noironetworks/aci-containers/pkg/nodeinfo/apis/aci.snat/v1"
21	snatglobalinfo "github.com/noironetworks/aci-containers/pkg/snatglobalinfo/apis/aci.snat/v1"
22	snatpolicy "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
23	snatclientset "github.com/noironetworks/aci-containers/pkg/snatpolicy/clientset/versioned"
24	"github.com/noironetworks/aci-containers/pkg/util"
25	"github.com/sirupsen/logrus"
26	v1 "k8s.io/api/core/v1"
27	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
28	"k8s.io/apimachinery/pkg/fields"
29	"k8s.io/apimachinery/pkg/labels"
30	"k8s.io/client-go/tools/cache"
31	"net"
32	"reflect"
33	)
34
35	const snatGraphName = "svcgraph"
36
37	type ContPodSelector struct {
38	Labels map[string]string
39	Namespace string
40	}
41
42	type ContPortRange struct {
43	Start int `json:"start,omitempty"`
44	End int `json:"end,omitempty"`
45	}
46
47	type ContSnatPolicy struct {
48	SnatIp []string
49	Selector ContPodSelector
50	PortRange []ContPortRange
51	Protocols []string
52	ExpandedSnatIps []string
53	ExpandedSnatPorts []snatglobalinfo.PortRange
54	}
55
56	func SnatPolicyLogger(log logrus.Logger, snat snatpolicy.SnatPolicy) *logrus.Entry {	×
57	return log.WithFields(logrus.Fields{	×
58	"namespace": snat.ObjectMeta.Namespace,	×
59	"name": snat.ObjectMeta.Name,	×
60	"spec": snat.Spec,	×
61	})	×
62	}	×
63
64	func (cont *AciController) initSnatInformerFromClient(
65	snatClient *snatclientset.Clientset) {	×
66	cont.initSnatInformerBase(	×
67	cache.NewListWatchFromClient(	×
68	snatClient.AciV1().RESTClient(), "snatpolicies",	×
69	metav1.NamespaceAll, fields.Everything()))	×
70	}	×
71
72	func (cont AciController) initSnatInformerBase(listWatch cache.ListWatch) {	1✔
73	cont.snatIndexer, cont.snatInformer = cache.NewIndexerInformer(	1✔
74	listWatch,	1✔
75	&snatpolicy.SnatPolicy{}, 0,	1✔
76	cache.ResourceEventHandlerFuncs{	1✔
77	AddFunc: func(obj interface{}) {	2✔
78	cont.snatPolicyUpdated(obj)	1✔
79	},	1✔
80	UpdateFunc: func(_, obj interface{}) {	1✔
81	cont.snatPolicyUpdated(obj)	1✔
82	},	1✔
83	DeleteFunc: func(obj interface{}) {	1✔
84	cont.snatPolicyDelete(obj)	1✔
85	},	1✔
86	},
87	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
88	)
89	cont.log.Debug("Initializing Snat Policy Informers")	1✔
90
91	}
92
93	func (cont *AciController) snatPolicyUpdated(obj interface{}) {	1✔
94	snatPolicy := obj.(*snatpolicy.SnatPolicy)	1✔
95	key, err := cache.MetaNamespaceKeyFunc(snatPolicy)	1✔
96	if err != nil {	1✔
97	SnatPolicyLogger(cont.log, snatPolicy).	×
98	Error("Could not create key:" + err.Error())	×
99	return	×
100	}	×
101	cont.queueSnatUpdateByKey(key)	1✔
102
103	}
104
105	func (cont *AciController) queueSnatUpdateByKey(key string) {	1✔
106	cont.log.Info("Added snatpolicy key: ", key)	1✔
107	cont.snatQueue.Add(key)	1✔
108	}	1✔
109
110	func (cont AciController) queueSnatUpdate(snatpolicy snatpolicy.SnatPolicy) {	1✔
111	key, err := cache.MetaNamespaceKeyFunc(snatpolicy)	1✔
112	if err != nil {	1✔
113	SnatPolicyLogger(cont.log, snatpolicy).	×
114	Error("Could not create key:" + err.Error())	×
115	return	×
116	}	×
117	cont.log.Info("Add snatpolicy key: ", key)	1✔
118	cont.snatQueue.Add(key)	1✔
119	}
120
121	func (cont AciController) handleSnatUpdate(snatPolicy snatpolicy.SnatPolicy) bool {	1✔
122	_, err := cache.MetaNamespaceKeyFunc(snatPolicy)	1✔
123	if err != nil {	1✔
124	SnatPolicyLogger(cont.log, snatPolicy).	×
125	Error("Could not create key:" + err.Error())	×
126	return false	×
127	}	×
128	cont.log.Info("Handle update for snatpolicy: ", snatPolicy.ObjectMeta.Name)	1✔
129	policyName := snatPolicy.ObjectMeta.Name	1✔
130	if snatPolicy.Status.State != snatpolicy.Ready {	1✔
131	if snatPolicy.Status.State == snatpolicy.IpPortsExhausted {	×
132	cont.indexMutex.Lock()	×
133	if snatInfo, ok := cont.snatPolicyCache[policyName]; ok {	×
134	if reflect.DeepEqual(snatPolicy.Spec.SnatIp, snatInfo.SnatIp) {	×
135	cont.indexMutex.Unlock()	×
136	return false	×
137	}	×
138	}
139	cont.indexMutex.Unlock()	×
140	}
141	if status, err := cont.validateCr(snatPolicy); !status {	×
142	cont.log.Error("SnatPolicy Failed: ", err)	×
143	return cont.setSnatPolicyStaus(policyName, snatpolicy.Failed)	×
144	}	×
145	return cont.setSnatPolicyStaus(policyName, snatpolicy.Ready)	×
146	}
147	cont.updateSnatPolicyCache(policyName, snatPolicy)	1✔
148	var requeue bool	1✔
149	cont.indexMutex.Lock()	1✔
150	if cont.snatSyncEnabled {	2✔
151	cont.indexMutex.Unlock()	1✔
152		1✔
153	if len(snatPolicy.Spec.SnatIp) == 0 {	1✔
154	err = cont.handleSnatPolicyForServices(snatPolicy)	×
155	} else {	1✔
156	err = cont.updateServiceDeviceInstanceSnat(snatGraphName)	1✔
157	}	1✔
158	if err != nil {	1✔
159	requeue = true	×
160	}	×
161	} else {	1✔
162	cont.indexMutex.Unlock()	1✔
163	}	1✔
164	return requeue	1✔
165	}
166
167	func (cont AciController) getServicesBySelector(selector labels.Selector, ns string) []v1.Service {	×
168	var services []*v1.Service	×
169	cache.ListAll(cont.serviceIndexer, selector,	×
170	func(servobj interface{}) {	×
171	service := servobj.(*v1.Service)	×
172	if len(ns) == 0 \|\| (len(ns) > 0 && ns == service.ObjectMeta.Namespace) {	×
173	if len(service.Status.LoadBalancer.Ingress) != 0 {	×
174	services = append(services, service)	×
175	}	×
176	}
177	})
178	return services	×
179	}
180
181	func (cont AciController) handleSnatPolicyForServices(snatpolicy snatpolicy.SnatPolicy) error {	×
182	ServiceList := cont.getServicesBySelector(labels.SelectorFromSet(	×
183	labels.Set(snatpolicy.Spec.Selector.Labels)),	×
184	snatpolicy.Spec.Selector.Namespace)	×
185		×
186	if len(ServiceList) == 0 {	×
187	return nil	×
188	}	×
189	for _, service := range ServiceList {	×
190	servicekey, err := cache.MetaNamespaceKeyFunc(service)	×
191	if err != nil {	×
192	servicekey = service.ObjectMeta.Namespace + "/" + service.ObjectMeta.Name	×
193	}	×
194	cont.indexMutex.Lock()	×
195	if service.GetDeletionTimestamp() == nil {	×
196	cont.snatServices[servicekey] = true	×
197	cont.queueServiceUpdateByKey(servicekey)	×
198	}	×
199	cont.indexMutex.Unlock()	×
200	}
201	return nil	×
202	}
203
204	func (cont AciController) updateSnatPolicyCache(key string, snatpolicy snatpolicy.SnatPolicy) {	1✔
205	cont.indexMutex.Lock()	1✔
206	cont.log.Debug("Updating snatpolicy cache for policy: ", snatpolicy.ObjectMeta.Name)	1✔
207	var policy ContSnatPolicy	1✔
208	env := cont.env.(*K8sEnvironment)	1✔
209	policy.SnatIp = snatpolicy.Spec.SnatIp	1✔
210	policy.ExpandedSnatIps = util.ExpandCIDRs(policy.SnatIp)	1✔
211	portRange := snatglobalinfo.PortRange{Start: 5000, End: 65000}	1✔
212	portsPerNode := 3000	1✔
213	if env.kubeClient != nil {	1✔
214	portRange, portsPerNode = util.GetPortRangeFromConfigMap(env.kubeClient)	×
215	}	×
216	var currPortRange []snatglobalinfo.PortRange	1✔
217	currPortRange = append(currPortRange, portRange)	1✔
218	policy.ExpandedSnatPorts = util.ExpandPortRanges(currPortRange, portsPerNode)	1✔
219	policy.Selector = ContPodSelector{Labels: snatpolicy.Spec.Selector.Labels, Namespace: snatpolicy.Spec.Selector.Namespace}	1✔
220	Update := false	1✔
221	if snatInfo, ok := cont.snatPolicyCache[key]; ok {	2✔
222	cont.log.Debug("Snatpolicy already exists in cache: ", snatpolicy.ObjectMeta.Name)	1✔
223	if !reflect.DeepEqual(policy.SnatIp, snatInfo.SnatIp) {	2✔
224	cont.log.Debug("Snatpolicy info found in cache marked to be updated: ", snatpolicy.ObjectMeta.Name)	1✔
225	cont.clearSnatGlobalCache(snatpolicy.ObjectMeta.Name, "")	1✔
226	Update = true	1✔
227	}	1✔
228	}
229	cont.snatPolicyCache[key] = &policy	1✔
230	nodeInfoKeys := make(map[string]bool)	1✔
231	if Update {	2✔
232	cont.getNodeInfoKeys(snatpolicy.ObjectMeta.Name, nodeInfoKeys)	1✔
233	}	1✔
234	cont.indexMutex.Unlock()	1✔
235	for key := range nodeInfoKeys {	2✔
236	cont.queueNodeInfoUpdateByKey(key)	1✔
237	}	1✔
238	}
239
240	func (cont *AciController) snatPolicyDelete(snatobj interface{}) {	1✔
241	snatpolicy := snatobj.(*snatpolicy.SnatPolicy)	1✔
242	cont.log.Info("Snatpolicy marked for deletion: ", snatpolicy.ObjectMeta.Name)	1✔
243	cont.indexMutex.Lock()	1✔
244	cont.clearSnatGlobalCache(snatpolicy.ObjectMeta.Name, "")	1✔
245	delete(cont.snatPolicyCache, snatpolicy.ObjectMeta.Name)	1✔
246		1✔
247	if len(snatpolicy.Spec.SnatIp) == 0 {	1✔
248	ServiceList := cont.getServicesBySelector(labels.SelectorFromSet(	×
249	labels.Set(snatpolicy.Spec.Selector.Labels)),	×
250	snatpolicy.Spec.Selector.Namespace)	×
251	if len(ServiceList) > 0 {	×
252	for _, service := range ServiceList {	×
253	servicekey, err1 := cache.MetaNamespaceKeyFunc(service)	×
254	if err1 != nil {	×
255	servicekey = service.ObjectMeta.Namespace + "/" + service.ObjectMeta.Name	×
256	}	×
257	delete(cont.snatServices, servicekey)	×
258	cont.queueServiceUpdateByKey(servicekey)	×
259	}
260	}
261	} else {	1✔
262	if len(cont.snatPolicyCache) == 0 {	2✔
263	cont.log.Debug("No more snat policies, deleting graph")	1✔
264	graphName := cont.aciNameForKey("snat", snatGraphName)	1✔
265	go cont.apicConn.ClearApicObjects(graphName)	1✔
266	} else {	2✔
267	go cont.updateServiceDeviceInstanceSnat(snatGraphName)	1✔
268	}	1✔
269	}
270	cont.indexMutex.Unlock()	1✔
271	}
272
273	func (cont *AciController) createGlobalInfoCache(unittestmode bool) bool {	1✔
274	cont.log.Info("Checking if snatglobalinfo CR already exists")	1✔
275		1✔
276	var nodeInfos []*nodeinfo.NodeInfo	1✔
277	cont.indexMutex.Lock()	1✔
278	defer cont.indexMutex.Unlock()	1✔
279	cache.ListAll(cont.snatNodeInfoIndexer, labels.Everything(),	1✔
280	func(nodeInfoObj interface{}) {	1✔
281	nodeInfo := nodeInfoObj.(*nodeinfo.NodeInfo)	×
282	nodeInfos = append(nodeInfos, nodeInfo)	×
283	})	×
284	env := cont.env.(*K8sEnvironment)	1✔
285	globalcl := env.snatGlobalClient	1✔
286	if globalcl == nil {	2✔
287	if unittestmode {	2✔
288	return false	1✔
289	}	1✔
290	cont.log.Fatalf("snatglobalinfo client not found")	×
291	}
292	snatglobalInfo, err := util.GetGlobalInfoCR(*globalcl)	×
293		×
294	if err != nil {	×
295	cont.log.Info("No existing snatglobalinfo CR found in controller bootstrap")	×
296	} else {	×
297	cont.log.Info("Syncing snatglobalinfo cache with existing CR")	×
298	macAddressToNodeName := make(map[string]string)	×
299	for _, value := range nodeInfos {	×
300	nodeName := value.ObjectMeta.Name	×
301	macAddress := value.Spec.Macaddress	×
302	macAddressToNodeName[macAddress] = nodeName	×
303	}	×
304
305	for _, glinfos := range snatglobalInfo.Spec.GlobalInfos {	×
306	for _, v := range glinfos {	×
307	nodeName := macAddressToNodeName[v.MacAddress]	×
308	snatIP := v.SnatIp	×
309	if _, ok := cont.snatGlobalInfoCache[snatIP]; !ok {	×
310	cont.snatGlobalInfoCache[snatIP] = make(map[string]*snatglobalinfo.GlobalInfo)	×
311	}	×
312	cont.snatGlobalInfoCache[snatIP][nodeName] = &v	×
313	}
314	}
315	}
316	return true	×
317	}
318
319	func (cont *AciController) snatFullSync() {	1✔
320	cache.ListAll(cont.snatIndexer, labels.Everything(),	1✔
321	func(sobj interface{}) {	2✔
322	cont.queueSnatUpdate(sobj.(*snatpolicy.SnatPolicy))	1✔
323	})	1✔
324	}
325
326	func (cont AciController) validateCr(cr snatpolicy.SnatPolicy) (bool, string) {	×
327	cont.indexMutex.Lock()	×
328	snatPolicyCache := make(map[string]*ContSnatPolicy)	×
329	for k, v := range cont.snatPolicyCache {	×
330	snatPolicyCache[k] = v	×
331	}	×
332	cont.indexMutex.Unlock()	×
333	if len(cont.snatPolicyCache) >= 1 {	×
334	cr_labels := cr.Spec.Selector.Labels	×
335	cr_ns := cr.Spec.Selector.Namespace	×
336	for key, item := range snatPolicyCache {	×
337	if cr.ObjectMeta.Name != key {	×
338	if (len(item.Selector.Labels) == 0) && (len(cr_labels) == 0) &&	×
339	(cr_ns == item.Selector.Namespace) {	×
340	return false, fmt.Sprintf(	×
341	"Namespace is conflicting with the snatpolicy %s",	×
342	cr.ObjectMeta.Name)	×
343	}	×
344	for _, val := range item.SnatIp {	×
345	_, net1, _ := parseIP(val)	×
346	for _, ip := range cr.Spec.SnatIp {	×
347	_, net2, err := parseIP(ip)	×
348	if err != nil {	×
349	return false, fmt.Sprintf(	×
350	"Invalid incoming SnatIP %s", ip)	×
351	}	×
352	if net2.Contains(net1.IP) \|\| net1.Contains(net2.IP) {	×
353	return false, fmt.Sprintf(	×
354	"SnatIP's are conflicting with the snatpolicy %s",	×
355	cr.ObjectMeta.Name)	×
356	}	×
357	}
358	}
359	}
360	}
361	} else {	×
362	for _, ip := range cr.Spec.SnatIp {	×
363	_, _, err := parseIP(ip)	×
364	if err != nil {	×
365	return false, fmt.Sprintf(	×
366	"Invalid incoming SnatIP %s", ip)	×
367	}	×
368	}
369	}
370	for _, ip := range cr.Spec.DestIp {	×
371	_, _, err := parseIP(ip)	×
372	if err != nil {	×
373	return false, fmt.Sprintf(	×
374	"Invalid incoming DestIP %s", ip)	×
375	}	×
376	}
377
378	return true, ""	×
379	}
380
381	func parseIP(cidr string) (net.IP, *net.IPNet, error) {	×
382	ip, ipnet, err := net.ParseCIDR(cidr)	×
383	if err != nil {	×
384	ip_temp := net.ParseIP(cidr)	×
385	if ip_temp != nil && ip_temp.To4() != nil {	×
386	cidr = cidr + "/32"	×
387	ip, ipnet, _ = net.ParseCIDR(cidr)	×
388	return ip, ipnet, nil	×
389	} else if ip_temp != nil && ip_temp.To16() != nil {	×
390	cidr = cidr + "/128"	×
391	ip, ipnet, _ = net.ParseCIDR(cidr)	×
392	return ip, ipnet, nil	×
393	} else {	×
394	return nil, nil, err	×
395	}	×
396	}
397	return ip, ipnet, err	×
398	}

noironetworks / aci-containers / 5715

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