6688

Committed 01 Jun 2022 04:28PM UTC coverage: 57.321%. First build

Build # 6688

Build Type

push

travis-ci-com

Committed by

akhilamohanan

Commit Message

Never reconsider a failed snatpolicy

When multiple namespace snatpolicies have overlapping ips, State of
snatpolicies are correct when we configure but later Failed snatpolicies
go to Ready state.

Never reconsider a Failed policy (the only option would be to delete and recreate it)

(cherry picked from commit 0af14dc87)

Run Details

5 of 5 new or added lines in 1 file covered. (100.0%)

12074 of 21064 relevant lines covered (57.32%)

0.64 hits per line

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

40.14

/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"
        "net"
        "reflect"

        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"
)

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

noironetworks / aci-containers / 6688

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