8450

Committed 17 Jan 2024 10:02AM UTC coverage: 60.272% (+7.0%) from 53.227%

Build # 8450

Build Type

push

travis-pro

Committed by

web-flow

Commit Message

Merge pull request #1213 from noironetworks/shaikmas-uts

Unit Tests

Run Details

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

4 existing lines in 1 file now uncovered.

13745 of 22805 relevant lines covered (60.27%)

1.35 hits per line

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

44.78

/pkg/controller/netflow.go

// Copyright 2020 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 (
        "strconv"

        "github.com/sirupsen/logrus"

        netflowpolicy "github.com/noironetworks/aci-containers/pkg/netflowpolicy/apis/aci.netflow/v1alpha"
        netflowclientset "github.com/noironetworks/aci-containers/pkg/netflowpolicy/clientset/versioned"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        "k8s.io/client-go/tools/cache"

        "github.com/noironetworks/aci-containers/pkg/apicapi"
)

const (
        netflowCRDName = "netflowpolicies.aci.netflow"
)

func NetflowPolicyLogger(log *logrus.Logger, netflow *netflowpolicy.NetflowPolicy) *logrus.Entry {
        return log.WithFields(logrus.Fields{
                "name": netflow.ObjectMeta.Name,
                "spec": netflow.Spec,
        })
}

func netflowInit(cont *AciController, stopCh <-chan struct{}) {
        cont.log.Debug("Initializing netflow client")
        restconfig := cont.env.RESTConfig()
        netflowClient, err := netflowclientset.NewForConfig(restconfig)
        if err != nil {
                cont.log.Errorf("Failed to intialize netflow client")
                return
        }
        cont.initNetflowInformerFromClient(netflowClient)
        go cont.netflowInformer.Run(stopCh)
        go cont.processQueue(cont.netflowQueue, cont.netflowIndexer,
                func(obj interface{}) bool {
                        return cont.handleNetflowPolUpdate(obj)
                }, nil, nil, stopCh)
        cache.WaitForCacheSync(stopCh, cont.netflowInformer.HasSynced)
}

func (cont *AciController) initNetflowInformerFromClient(
        netflowClient *netflowclientset.Clientset) {
        cont.initNetflowInformerBase(
                cache.NewListWatchFromClient(
                        netflowClient.AciV1alpha().RESTClient(), "netflowpolicies",
                        metav1.NamespaceAll, fields.Everything()))
}

func (cont *AciController) initNetflowInformerBase(listWatch *cache.ListWatch) {
        cont.netflowIndexer, cont.netflowInformer = cache.NewIndexerInformer(
                listWatch,
                &netflowpolicy.NetflowPolicy{}, 0,
                cache.ResourceEventHandlerFuncs{
                        AddFunc: func(obj interface{}) {
                                cont.netflowPolicyUpdated(obj)
                        },
                        UpdateFunc: func(_, obj interface{}) {
                                cont.netflowPolicyUpdated(obj)
                        },
                        DeleteFunc: func(obj interface{}) {
                                cont.netflowPolicyDelete(obj)
                        },
                },
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        cont.log.Debug("Initializing Netflow Policy Informers")
}

func (cont *AciController) netflowPolicyUpdated(obj interface{}) {
        netflowPolicy, ok := obj.(*netflowpolicy.NetflowPolicy)
        if !ok {
                cont.log.Error("netflowPolicyUpdated: Bad object type")
                return
        }
        key, err := cache.MetaNamespaceKeyFunc(netflowPolicy)
        if err != nil {
                NetflowPolicyLogger(cont.log, netflowPolicy).
                        Error("Could not create key:" + err.Error())
                return
        }
        cont.queueNetflowUpdateByKey(key)
}

func (cont *AciController) queueNetflowUpdateByKey(key string) {
        cont.netflowQueue.Add(key)
}

func (cont *AciController) netflowPolicyDelete(obj interface{}) bool {
        nf, isNf := obj.(*netflowpolicy.NetflowPolicy)
        if !isNf {
                deletedState, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        NetflowPolicyLogger(cont.log, nf).
                                Error("Received unexpected object: ", obj)
                        return false
                }
                nf, ok = deletedState.Obj.(*netflowpolicy.NetflowPolicy)
                if !ok {
                        NetflowPolicyLogger(cont.log, nf).
                                Error("DeletedFinalStateUnknown contained non-netflow object: ", deletedState.Obj)
                        return false
                }
        }
        nfkey, err := cache.MetaNamespaceKeyFunc(nf)
        if err != nil {
                NetflowPolicyLogger(cont.log, nf).
                        Error("Could not create netflow key: ", err)
                return false
        }
        cont.apicConn.ClearApicObjects(cont.aciNameForKey("nfp", nfkey))
        return true
}

// netflowPolObjs is used to build Netflow Policy objects
func (cont *AciController) netflowPolObjs(nfp *netflowpolicy.NetflowPolicy) apicapi.ApicSlice {
        key, err := cache.MetaNamespaceKeyFunc(nfp)
        if err != nil {
                NetflowPolicyLogger(cont.log, nfp).
                        Error("Could not create netflow key: ", err)
        }
        labelKey := cont.aciNameForKey("nfp", key)
        cont.log.Debug("create netflowpolicy")
        nf := apicapi.NewNetflowVmmExporterPol(labelKey)
        nfDn := nf.GetDn()
        apicSlice := apicapi.ApicSlice{nf}
        nf.SetAttr("dstAddr", nfp.Spec.FlowSamplingPolicy.DstAddr)
        nf.SetAttr("dstPort", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.DstPort))

        if nfp.Spec.FlowSamplingPolicy.FlowType == "netflow" {
                nf.SetAttr("ver", "v5")
        } else if nfp.Spec.FlowSamplingPolicy.FlowType == "ipfix" {
                nf.SetAttr("ver", "v9")
        }

        VmmVSwitch := apicapi.NewVmmVSwitchPolicyCont(cont.vmmDomainProvider(), cont.config.AciVmmDomain)
        RsVmmVSwitch := apicapi.NewVmmRsVswitchExporterPol(VmmVSwitch.GetDn(), nfDn)
        VmmVSwitch.AddChild(RsVmmVSwitch)
        RsVmmVSwitch.SetAttr("activeFlowTimeOut", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.ActiveFlowTimeOut))
        RsVmmVSwitch.SetAttr("idleFlowTimeOut", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.IdleFlowTimeOut))
        RsVmmVSwitch.SetAttr("samplingRate", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.SamplingRate))
        apicSlice = append(apicSlice, VmmVSwitch)

        cont.log.Info("Netflow ApicSlice: ", apicSlice)

        return apicapi.ApicSlice{nf, VmmVSwitch}
}

// func returns false if exceuted without error, true if the caller has to requeue.
func (cont *AciController) handleNetflowPolUpdate(obj interface{}) bool {
        // Ability to configure Netflow Policy is available only in APIC versions >= 5.0(x).
        // In APIC versions < 5.0(x), Netflow VMM Exporter Policy can be associated with VMware domains only.
        if apicapi.ApicVersion < "5.0" {
                cont.log.Error("Cannot create Netflow Policy in APIC versions < 5.0(x). Actual APIC version: ",
                        apicapi.ApicVersion)
                return false
        }
        nfp, ok := obj.(*netflowpolicy.NetflowPolicy)
        if !ok {
                cont.log.Error("handleNetflowPolUpdate: Bad object type")
                return false
        }
        logger := NetflowPolicyLogger(cont.log, nfp)
        key, err := cache.MetaNamespaceKeyFunc(nfp)
        if err != nil {
                logger.Error("Could not create netflow policy key: ", err)
                return false
        }
        labelKey := cont.aciNameForKey("nfp", key)
        cont.apicConn.WriteApicObjects(labelKey, cont.netflowPolObjs(nfp))

        return false
}

1	// Copyright 2020 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	"strconv"
20
21	"github.com/sirupsen/logrus"
22
23	netflowpolicy "github.com/noironetworks/aci-containers/pkg/netflowpolicy/apis/aci.netflow/v1alpha"
24	netflowclientset "github.com/noironetworks/aci-containers/pkg/netflowpolicy/clientset/versioned"
25	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26	"k8s.io/apimachinery/pkg/fields"
27	"k8s.io/client-go/tools/cache"
28
29	"github.com/noironetworks/aci-containers/pkg/apicapi"
30	)
31
32	const (
33	netflowCRDName = "netflowpolicies.aci.netflow"
34	)
35
36	func NetflowPolicyLogger(log logrus.Logger, netflow netflowpolicy.NetflowPolicy) *logrus.Entry {	2✔
37	return log.WithFields(logrus.Fields{	2✔
38	"name": netflow.ObjectMeta.Name,	2✔
39	"spec": netflow.Spec,	2✔
40	})	2✔
41	}	2✔
42
43	func netflowInit(cont *AciController, stopCh <-chan struct{}) {	×
44	cont.log.Debug("Initializing netflow client")	×
45	restconfig := cont.env.RESTConfig()	×
46	netflowClient, err := netflowclientset.NewForConfig(restconfig)	×
47	if err != nil {	×
48	cont.log.Errorf("Failed to intialize netflow client")	×
49	return	×
50	}	×
51	cont.initNetflowInformerFromClient(netflowClient)	×
52	go cont.netflowInformer.Run(stopCh)	×
53	go cont.processQueue(cont.netflowQueue, cont.netflowIndexer,	×
54	func(obj interface{}) bool {	×
55	return cont.handleNetflowPolUpdate(obj)	×
56	}, nil, nil, stopCh)	×
57	cache.WaitForCacheSync(stopCh, cont.netflowInformer.HasSynced)	×
58	}
59
60	func (cont *AciController) initNetflowInformerFromClient(
61	netflowClient *netflowclientset.Clientset) {	×
62	cont.initNetflowInformerBase(	×
63	cache.NewListWatchFromClient(	×
64	netflowClient.AciV1alpha().RESTClient(), "netflowpolicies",	×
65	metav1.NamespaceAll, fields.Everything()))	×
66	}	×
67
68	func (cont AciController) initNetflowInformerBase(listWatch cache.ListWatch) {	2✔
69	cont.netflowIndexer, cont.netflowInformer = cache.NewIndexerInformer(	2✔
70	listWatch,	2✔
71	&netflowpolicy.NetflowPolicy{}, 0,	2✔
72	cache.ResourceEventHandlerFuncs{	2✔
73	AddFunc: func(obj interface{}) {	2✔
74	cont.netflowPolicyUpdated(obj)	×
75	},	×
76	UpdateFunc: func(_, obj interface{}) {	×
77	cont.netflowPolicyUpdated(obj)	×
78	},	×
79	DeleteFunc: func(obj interface{}) {	×
80	cont.netflowPolicyDelete(obj)	×
81	},	×
82	},
83	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
84	)
85	cont.log.Debug("Initializing Netflow Policy Informers")	2✔
86	}
87
88	func (cont *AciController) netflowPolicyUpdated(obj interface{}) {	×
89	netflowPolicy, ok := obj.(*netflowpolicy.NetflowPolicy)	×
90	if !ok {	×
91	cont.log.Error("netflowPolicyUpdated: Bad object type")	×
92	return	×
93	}	×
94	key, err := cache.MetaNamespaceKeyFunc(netflowPolicy)	×
95	if err != nil {	×
96	NetflowPolicyLogger(cont.log, netflowPolicy).	×
97	Error("Could not create key:" + err.Error())	×
98	return	×
99	}	×
100	cont.queueNetflowUpdateByKey(key)	×
101	}
102
103	func (cont *AciController) queueNetflowUpdateByKey(key string) {	×
104	cont.netflowQueue.Add(key)	×
105	}	×
106
107	func (cont *AciController) netflowPolicyDelete(obj interface{}) bool {	2✔
108	nf, isNf := obj.(*netflowpolicy.NetflowPolicy)	2✔
109	if !isNf {	2✔
110	deletedState, ok := obj.(cache.DeletedFinalStateUnknown)	×
111	if !ok {	×
112	NetflowPolicyLogger(cont.log, nf).	×
113	Error("Received unexpected object: ", obj)	×
114	return false	×
115	}	×
116	nf, ok = deletedState.Obj.(*netflowpolicy.NetflowPolicy)	×
117	if !ok {	×
118	NetflowPolicyLogger(cont.log, nf).	×
119	Error("DeletedFinalStateUnknown contained non-netflow object: ", deletedState.Obj)	×
120	return false	×
121	}	×
122	}
123	nfkey, err := cache.MetaNamespaceKeyFunc(nf)	2✔
124	if err != nil {	2✔
125	NetflowPolicyLogger(cont.log, nf).	×
126	Error("Could not create netflow key: ", err)	×
127	return false	×
128	}	×
129	cont.apicConn.ClearApicObjects(cont.aciNameForKey("nfp", nfkey))	2✔
130	return true	2✔
131	}
132
133	// netflowPolObjs is used to build Netflow Policy objects
134	func (cont AciController) netflowPolObjs(nfp netflowpolicy.NetflowPolicy) apicapi.ApicSlice {	2✔
135	key, err := cache.MetaNamespaceKeyFunc(nfp)	2✔
136	if err != nil {	2✔
137	NetflowPolicyLogger(cont.log, nfp).	×
138	Error("Could not create netflow key: ", err)	×
139	}	×
140	labelKey := cont.aciNameForKey("nfp", key)	2✔
141	cont.log.Debug("create netflowpolicy")	2✔
142	nf := apicapi.NewNetflowVmmExporterPol(labelKey)	2✔
143	nfDn := nf.GetDn()	2✔
144	apicSlice := apicapi.ApicSlice{nf}	2✔
145	nf.SetAttr("dstAddr", nfp.Spec.FlowSamplingPolicy.DstAddr)	2✔
146	nf.SetAttr("dstPort", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.DstPort))	2✔
147		2✔
148	if nfp.Spec.FlowSamplingPolicy.FlowType == "netflow" {	4✔
149	nf.SetAttr("ver", "v5")	2✔
150	} else if nfp.Spec.FlowSamplingPolicy.FlowType == "ipfix" {	6✔
151	nf.SetAttr("ver", "v9")	2✔
152	}	2✔
153
154	VmmVSwitch := apicapi.NewVmmVSwitchPolicyCont(cont.vmmDomainProvider(), cont.config.AciVmmDomain)	2✔
155	RsVmmVSwitch := apicapi.NewVmmRsVswitchExporterPol(VmmVSwitch.GetDn(), nfDn)	2✔
156	VmmVSwitch.AddChild(RsVmmVSwitch)	2✔
157	RsVmmVSwitch.SetAttr("activeFlowTimeOut", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.ActiveFlowTimeOut))	2✔
158	RsVmmVSwitch.SetAttr("idleFlowTimeOut", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.IdleFlowTimeOut))	2✔
159	RsVmmVSwitch.SetAttr("samplingRate", strconv.Itoa(nfp.Spec.FlowSamplingPolicy.SamplingRate))	2✔
160	apicSlice = append(apicSlice, VmmVSwitch)	2✔
161		2✔
162	cont.log.Info("Netflow ApicSlice: ", apicSlice)	2✔
163		2✔
164	return apicapi.ApicSlice{nf, VmmVSwitch}	2✔
165	}
166
167	// func returns false if exceuted without error, true if the caller has to requeue.
168	func (cont *AciController) handleNetflowPolUpdate(obj interface{}) bool {	2✔
169	// Ability to configure Netflow Policy is available only in APIC versions >= 5.0(x).	2✔
170	// In APIC versions < 5.0(x), Netflow VMM Exporter Policy can be associated with VMware domains only.	2✔
171	if apicapi.ApicVersion < "5.0" {	2✔
UNCOV 172	cont.log.Error("Cannot create Netflow Policy in APIC versions < 5.0(x). Actual APIC version: ",	×
UNCOV 173	apicapi.ApicVersion)	×
UNCOV 174	return false	×
UNCOV 175	}	×
176	nfp, ok := obj.(*netflowpolicy.NetflowPolicy)	2✔
177	if !ok {	2✔
178	cont.log.Error("handleNetflowPolUpdate: Bad object type")	×
179	return false	×
180	}	×
181	logger := NetflowPolicyLogger(cont.log, nfp)	2✔
182	key, err := cache.MetaNamespaceKeyFunc(nfp)	2✔
183	if err != nil {	2✔
184	logger.Error("Could not create netflow policy key: ", err)	×
185	return false	×
186	}	×
187	labelKey := cont.aciNameForKey("nfp", key)	2✔
188	cont.apicConn.WriteApicObjects(labelKey, cont.netflowPolObjs(nfp))	2✔
189		2✔
190	return false	2✔
191	}

noironetworks / aci-containers / 8450

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