6757

Committed 02 Aug 2022 01:46PM UTC coverage: 57.299%. First build

Build # 6757

Build Type

push

travis-ci-com

Committed by

akhilamohanan

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

Run Details

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

12070 of 21065 relevant lines covered (57.3%)

0.64 hits per line

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

70.31

/pkg/hostagent/agent.go

// Copyright 2017 Cisco Systems, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package hostagent

import (
        "net"
        "sync"
        "time"

        "github.com/containernetworking/cni/pkg/types"
        crdclientset "github.com/noironetworks/aci-containers/pkg/gbpcrd/clientset/versioned"
        aciv1 "github.com/noironetworks/aci-containers/pkg/gbpcrd/clientset/versioned/typed/acipolicy/v1"
        "github.com/noironetworks/aci-containers/pkg/index"
        "github.com/noironetworks/aci-containers/pkg/ipam"
        md "github.com/noironetworks/aci-containers/pkg/metadata"
        nodepodifclset "github.com/noironetworks/aci-containers/pkg/nodepodif/clientset/versioned"
        nodepodifv1 "github.com/noironetworks/aci-containers/pkg/nodepodif/clientset/versioned/typed/acipolicy/v1"
        snatpolicy "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
        "github.com/noironetworks/aci-containers/pkg/util"
        "github.com/sirupsen/logrus"
        "github.com/vishvananda/netlink"
        "golang.org/x/time/rate"
        v1 "k8s.io/api/core/v1"
        "k8s.io/apimachinery/pkg/util/wait"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/rest"
        "k8s.io/client-go/tools/cache"
        "k8s.io/client-go/util/workqueue"
)

type HostAgent struct {
        log    *logrus.Logger
        config *HostAgentConfig
        env    Environment

        indexMutex           sync.Mutex
        ipamMutex            sync.Mutex
        snatPolicyLabelMutex sync.RWMutex
        snatPolicyCacheMutex sync.RWMutex

        opflexEps             map[string][]*opflexEndpoint
        opflexServices        map[string]*opflexService
        epMetadata            map[string]map[string]*md.ContainerMetadata
        podIpToName           map[string]string
        cniToPodID            map[string]string
        podUidToName          map[string]string
        serviceEp             md.ServiceEndpoint
        crdClient             aciv1.AciV1Interface
        nodePodIFClient       nodepodifv1.AciV1Interface
        podInformer           cache.SharedIndexInformer
        endpointsInformer     cache.SharedIndexInformer
        serviceInformer       cache.SharedIndexInformer
        nodeInformer          cache.SharedIndexInformer
        nsInformer            cache.SharedIndexInformer
        netPolInformer        cache.SharedIndexInformer
        depInformer           cache.SharedIndexInformer
        rcInformer            cache.SharedIndexInformer
        snatGlobalInformer    cache.SharedIndexInformer
        controllerInformer    cache.SharedIndexInformer
        snatPolicyInformer    cache.SharedIndexInformer
        qosPolicyInformer     cache.SharedIndexInformer
        rdConfigInformer      cache.SharedIndexInformer
        qosPolPods            *index.PodSelectorIndex
        endpointSliceInformer cache.SharedIndexInformer
        netPolPods            *index.PodSelectorIndex
        depPods               *index.PodSelectorIndex
        rcPods                *index.PodSelectorIndex
        podNetAnnotation      string
        podIps                *ipam.IpCache
        usedIPs               map[string]string

        syncEnabled         bool
        opflexConfigWritten bool
        syncQueue           workqueue.RateLimitingInterface
        syncProcessors      map[string]func() bool

        ignoreOvsPorts        map[string][]string
        netNsFuncChan         chan func()
        vtepIP                string
        gbpServerIP           string
        opflexSnatGlobalInfos map[string][]*opflexSnatGlobalInfo
        opflexSnatLocalInfos  map[string]*opflexSnatLocalInfo
        //snatpods per snat policy
        snatPods map[string]map[string]ResourceType
        //Object Key and list of labels active for snatpolicy
        snatPolicyLabels map[string]map[string]ResourceType
        snatPolicyCache  map[string]*snatpolicy.SnatPolicy
        rdConfig         *opflexRdConfig
        poster           *EventPoster
        ocServices       []opflexOcService // OpenShiftservices
        serviceEndPoints ServiceEndPointType
        // Service to pod uids to track EPfiles aded with clusterIp
        servicetoPodUids map[string]map[string]struct{}
        // reverse map to get ServiceIp's from poduid
        podtoServiceUids map[string]map[string]string
        nodePodIfEPs     map[string]*opflexEndpoint
        // integration test checker
        integ_test *string `json:",omitempty"`
}

type ServiceEndPointType interface {
        InitClientInformer(kubeClient *kubernetes.Clientset)
        Run(stopCh <-chan struct{})
        SetOpflexService(ofas *opflexService, as *v1.Service,
                external bool, key string, sp v1.ServicePort) bool
}

type serviceEndpoint struct {
        agent *HostAgent
}
type serviceEndpointSlice struct {
        agent *HostAgent
}

func (sep *serviceEndpoint) InitClientInformer(kubeClient *kubernetes.Clientset) {
        sep.agent.initEndpointsInformerFromClient(kubeClient)
}

func (seps *serviceEndpointSlice) InitClientInformer(kubeClient *kubernetes.Clientset) {
        seps.agent.initEndpointSliceInformerFromClient(kubeClient)
}

func (sep *serviceEndpoint) Run(stopCh <-chan struct{}) {
        go sep.agent.endpointsInformer.Run(stopCh)
        cache.WaitForCacheSync(stopCh, sep.agent.endpointsInformer.HasSynced)
}

func (seps *serviceEndpointSlice) Run(stopCh <-chan struct{}) {
        go seps.agent.endpointSliceInformer.Run(stopCh)
        cache.WaitForCacheSync(stopCh, seps.agent.endpointSliceInformer.HasSynced)
}

func NewHostAgent(config *HostAgentConfig, env Environment, log *logrus.Logger) *HostAgent {
        ha := &HostAgent{
                log:            log,
                config:         config,
                env:            env,
                opflexEps:      make(map[string][]*opflexEndpoint),
                opflexServices: make(map[string]*opflexService),
                epMetadata:     make(map[string]map[string]*md.ContainerMetadata),
                podIpToName:    make(map[string]string),
                cniToPodID:     make(map[string]string),
                podUidToName:   make(map[string]string),
                nodePodIfEPs:   make(map[string]*opflexEndpoint),

                podIps: ipam.NewIpCache(),

                ignoreOvsPorts: make(map[string][]string),

                netNsFuncChan:         make(chan func()),
                opflexSnatGlobalInfos: make(map[string][]*opflexSnatGlobalInfo),
                opflexSnatLocalInfos:  make(map[string]*opflexSnatLocalInfo),
                snatPods:              make(map[string]map[string]ResourceType),
                snatPolicyLabels:      make(map[string]map[string]ResourceType),
                snatPolicyCache:       make(map[string]*snatpolicy.SnatPolicy),
                servicetoPodUids:      make(map[string]map[string]struct{}),
                podtoServiceUids:      make(map[string]map[string]string),
                syncQueue: workqueue.NewNamedRateLimitingQueue(
                        &workqueue.BucketRateLimiter{
                                Limiter: rate.NewLimiter(rate.Limit(10), int(10)),
                        }, "sync"),
                ocServices: []opflexOcService{
                        {
                                RouterInternalDefault,
                                OpenShiftIngressNs,
                        },
                },
        }

        ha.syncProcessors = map[string]func() bool{
                "eps":           ha.syncEps,
                "services":      ha.syncServices,
                "opflexServer":  ha.syncOpflexServer,
                "snat":          ha.syncSnat,
                "snatnodeInfo":  ha.syncSnatNodeInfo,
                "rdconfig":      ha.syncRdConfig,
                "snatLocalInfo": ha.UpdateLocalInfoCr,
                "nodepodifs":    ha.syncNodePodIfs}

        if ha.config.EPRegistry == "k8s" {
                cfg, err := rest.InClusterConfig()
                if err != nil {
                        log.Errorf("ERROR getting cluster config: %v", err)
                        return ha
                }
                aciawClient, err := crdclientset.NewForConfig(cfg)
                if err != nil {
                        log.Errorf("ERROR getting crd client for registry: %v", err)
                        return ha
                }
                ha.crdClient = aciawClient.AciV1()
        }
        if ha.config.EnableNodePodIF {
                cfg, err := rest.InClusterConfig()
                if err != nil {
                        log.Errorf("ERROR getting cluster config: %v", err)
                        return ha
                }
                nodepodifClient, err := nodepodifclset.NewForConfig(cfg)
                if err != nil {
                        log.Errorf("ERROR getting nodepodif client for enabling NodePodIF: %v", err)
                        return ha
                }
                ha.nodePodIFClient = nodepodifClient.AciV1()
        }
        return ha
}

func addPodRoute(ipn types.IPNet, dev string, src string) error {
        link, err := netlink.LinkByName(dev)
        if err != nil {
                return err
        }
        if err := netlink.LinkSetUp(link); err != nil {
                return err
        }
        ipsrc := net.ParseIP(src)
        dst := &net.IPNet{
                IP:   ipn.IP,
                Mask: ipn.Mask,
        }
        route := netlink.Route{LinkIndex: link.Attrs().Index, Dst: dst, Src: ipsrc}
        return netlink.RouteAdd(&route)
}

func (agent *HostAgent) ReadSnatPolicyLabel(key string) (map[string]ResourceType, bool) {
        agent.snatPolicyLabelMutex.RLock()
        defer agent.snatPolicyLabelMutex.RUnlock()
        value, ok := agent.snatPolicyLabels[key]
        return value, ok
}

func (agent *HostAgent) WriteSnatPolicyLabel(key string, policy string, res ResourceType) {
        agent.snatPolicyLabelMutex.Lock()
        defer agent.snatPolicyLabelMutex.Unlock()
        agent.snatPolicyLabels[key][policy] = res
}

func (agent *HostAgent) WriteNewSnatPolicyLabel(key string) {
        agent.snatPolicyLabelMutex.Lock()
        defer agent.snatPolicyLabelMutex.Unlock()
        agent.snatPolicyLabels[key] = make(map[string]ResourceType)
}

func (agent *HostAgent) DeleteSnatPolicyLabelEntry(key string, policy string) {
        agent.snatPolicyLabelMutex.Lock()
        defer agent.snatPolicyLabelMutex.Unlock()
        delete(agent.snatPolicyLabels[key], policy)
}

func (agent *HostAgent) DeleteSnatPolicyLabel(key string) {
        agent.snatPolicyLabelMutex.Lock()
        defer agent.snatPolicyLabelMutex.Unlock()
        delete(agent.snatPolicyLabels, key)
}

func (agent *HostAgent) DeleteMatchingSnatPolicyLabel(policy string) {
        agent.snatPolicyLabelMutex.Lock()
        defer agent.snatPolicyLabelMutex.Unlock()
        for key, v := range agent.snatPolicyLabels {
                if _, ok := v[policy]; ok {
                        delete(agent.snatPolicyLabels[key], policy)
                }
        }
}

func (agent *HostAgent) Init() {
        agent.log.Debug("Initializing endpoint CNI metadata")
        err := md.LoadMetadata(agent.config.CniMetadataDir,
                agent.config.CniNetwork, &agent.epMetadata)
        if err != nil {
                panic(err.Error())
        }
        agent.log.Info("Loaded cached endpoint CNI metadata: ", len(agent.epMetadata))
        agent.buildUsedIPs()
        // check if the cluster supports endpoint slices
        // if cluster doesn't have the support fallback to endpoints
        kubeClient := agent.env.(*K8sEnvironment).kubeClient
        if util.IsEndPointSlicesSupported(kubeClient) {
                agent.serviceEndPoints = &serviceEndpointSlice{}
                agent.serviceEndPoints.(*serviceEndpointSlice).agent = agent
                agent.log.Info("Initializing ServiceEndpointSlices")
        } else {
                agent.serviceEndPoints = &serviceEndpoint{}
                agent.serviceEndPoints.(*serviceEndpoint).agent = agent
                agent.log.Info("Initializing ServiceEndpoints")
        }
        err = agent.env.Init(agent)
        if err != nil {
                panic(err.Error())
        }
}

func (agent *HostAgent) ScheduleSync(syncType string) {
        agent.syncQueue.AddRateLimited(syncType)
}

func (agent *HostAgent) scheduleSyncEps() {
        agent.ScheduleSync("eps")
        agent.scheduleSyncNodePodIfs()
}

func (agent *HostAgent) scheduleSyncServices() {
        agent.ScheduleSync("services")
}

func (agent *HostAgent) scheduleSyncSnats() {
        agent.ScheduleSync("snat")
}

func (agent *HostAgent) scheduleSyncOpflexServer() {
        agent.ScheduleSync("opflexServer")
}
func (agent *HostAgent) scheduleSyncNodeInfo() {
        agent.ScheduleSync("snatnodeInfo")
}
func (agent *HostAgent) scheduleSyncRdConfig() {
        agent.ScheduleSync("rdconfig")
}
func (agent *HostAgent) scheduleSyncLocalInfo() {
        agent.ScheduleSync("snatLocalInfo")
}
func (agent *HostAgent) scheduleSyncNodePodIfs() {
        agent.ScheduleSync("nodepodifs")
}

func (agent *HostAgent) runTickers(stopCh <-chan struct{}) {
        ticker := time.NewTicker(time.Second * 30)
        defer ticker.Stop()

        for {
                select {
                case <-ticker.C:
                        agent.updateOpflexConfig()
                case <-stopCh:
                        return
                }
        }
}

func (agent *HostAgent) processSyncQueue(queue workqueue.RateLimitingInterface,
        queueStop <-chan struct{}) {

        go wait.Until(func() {
                for {
                        syncType, quit := queue.Get()
                        if quit {
                                break
                        }

                        var requeue bool
                        switch syncType := syncType.(type) {
                        case string:
                                if f, ok := agent.syncProcessors[syncType]; ok {
                                        requeue = f()
                                }
                        }
                        if requeue {
                                queue.AddRateLimited(syncType)
                        } else {
                                queue.Forget(syncType)
                        }
                        queue.Done(syncType)

                }
        }, time.Second, queueStop)
        <-queueStop
        queue.ShutDown()
}

func (agent *HostAgent) EnableSync() (changed bool) {
        changed = false
        agent.indexMutex.Lock()
        if agent.syncEnabled == false {
                agent.syncEnabled = true
                changed = true
        }
        agent.indexMutex.Unlock()
        if changed {
                agent.log.Info("Enabling OpFlex endpoint and service sync")
                agent.scheduleSyncServices()
                agent.scheduleSyncEps()
                agent.scheduleSyncSnats()
                agent.scheduleSyncNodeInfo()
                agent.scheduleSyncNodePodIfs()
        }
        return
}

func (agent *HostAgent) Run(stopCh <-chan struct{}) {
        err := agent.populateSnatLocalInfos()
        if err != nil {
                agent.log.Error("Failed to populate opflexSnatLocalInfos ", err.Error())
                panic(err.Error())
        }
        syncEnabled, err := agent.env.PrepareRun(stopCh)
        if err != nil {
                panic(err.Error())
        }
        if agent.config.OpFlexEndpointDir == "" ||
                agent.config.OpFlexServiceDir == "" ||
                agent.config.OpFlexSnatDir == "" {
                agent.log.Warn("OpFlex endpoint,service or snat directories not set")

        } else {
                if syncEnabled {
                        agent.EnableSync()
                }
                go agent.processSyncQueue(agent.syncQueue, stopCh)
        }

        agent.log.Info("Starting endpoint RPC")
        err = agent.runEpRPC(stopCh)
        if err != nil {
                panic(err.Error())
        }

        agent.cleanupSetup()
}

1	// Copyright 2017 Cisco Systems, Inc.
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package hostagent
16
17	import (
18	"net"
19	"sync"
20	"time"
21
22	"github.com/containernetworking/cni/pkg/types"
23	crdclientset "github.com/noironetworks/aci-containers/pkg/gbpcrd/clientset/versioned"
24	aciv1 "github.com/noironetworks/aci-containers/pkg/gbpcrd/clientset/versioned/typed/acipolicy/v1"
25	"github.com/noironetworks/aci-containers/pkg/index"
26	"github.com/noironetworks/aci-containers/pkg/ipam"
27	md "github.com/noironetworks/aci-containers/pkg/metadata"
28	nodepodifclset "github.com/noironetworks/aci-containers/pkg/nodepodif/clientset/versioned"
29	nodepodifv1 "github.com/noironetworks/aci-containers/pkg/nodepodif/clientset/versioned/typed/acipolicy/v1"
30	snatpolicy "github.com/noironetworks/aci-containers/pkg/snatpolicy/apis/aci.snat/v1"
31	"github.com/noironetworks/aci-containers/pkg/util"
32	"github.com/sirupsen/logrus"
33	"github.com/vishvananda/netlink"
34	"golang.org/x/time/rate"
35	v1 "k8s.io/api/core/v1"
36	"k8s.io/apimachinery/pkg/util/wait"
37	"k8s.io/client-go/kubernetes"
38	"k8s.io/client-go/rest"
39	"k8s.io/client-go/tools/cache"
40	"k8s.io/client-go/util/workqueue"
41	)
42
43	type HostAgent struct {
44	log *logrus.Logger
45	config *HostAgentConfig
46	env Environment
47
48	indexMutex sync.Mutex
49	ipamMutex sync.Mutex
50	snatPolicyLabelMutex sync.RWMutex
51	snatPolicyCacheMutex sync.RWMutex
52
53	opflexEps map[string][]*opflexEndpoint
54	opflexServices map[string]*opflexService
55	epMetadata map[string]map[string]*md.ContainerMetadata
56	podIpToName map[string]string
57	cniToPodID map[string]string
58	podUidToName map[string]string
59	serviceEp md.ServiceEndpoint
60	crdClient aciv1.AciV1Interface
61	nodePodIFClient nodepodifv1.AciV1Interface
62	podInformer cache.SharedIndexInformer
63	endpointsInformer cache.SharedIndexInformer
64	serviceInformer cache.SharedIndexInformer
65	nodeInformer cache.SharedIndexInformer
66	nsInformer cache.SharedIndexInformer
67	netPolInformer cache.SharedIndexInformer
68	depInformer cache.SharedIndexInformer
69	rcInformer cache.SharedIndexInformer
70	snatGlobalInformer cache.SharedIndexInformer
71	controllerInformer cache.SharedIndexInformer
72	snatPolicyInformer cache.SharedIndexInformer
73	qosPolicyInformer cache.SharedIndexInformer
74	rdConfigInformer cache.SharedIndexInformer
75	qosPolPods *index.PodSelectorIndex
76	endpointSliceInformer cache.SharedIndexInformer
77	netPolPods *index.PodSelectorIndex
78	depPods *index.PodSelectorIndex
79	rcPods *index.PodSelectorIndex
80	podNetAnnotation string
81	podIps *ipam.IpCache
82	usedIPs map[string]string
83
84	syncEnabled bool
85	opflexConfigWritten bool
86	syncQueue workqueue.RateLimitingInterface
87	syncProcessors map[string]func() bool
88
89	ignoreOvsPorts map[string][]string
90	netNsFuncChan chan func()
91	vtepIP string
92	gbpServerIP string
93	opflexSnatGlobalInfos map[string][]*opflexSnatGlobalInfo
94	opflexSnatLocalInfos map[string]*opflexSnatLocalInfo
95	//snatpods per snat policy
96	snatPods map[string]map[string]ResourceType
97	//Object Key and list of labels active for snatpolicy
98	snatPolicyLabels map[string]map[string]ResourceType
99	snatPolicyCache map[string]*snatpolicy.SnatPolicy
100	rdConfig *opflexRdConfig
101	poster *EventPoster
102	ocServices []opflexOcService // OpenShiftservices
103	serviceEndPoints ServiceEndPointType
104	// Service to pod uids to track EPfiles aded with clusterIp
105	servicetoPodUids map[string]map[string]struct{}
106	// reverse map to get ServiceIp's from poduid
107	podtoServiceUids map[string]map[string]string
108	nodePodIfEPs map[string]*opflexEndpoint
109	// integration test checker
110	integ_test *string `json:",omitempty"`
111	}
112
113	type ServiceEndPointType interface {
114	InitClientInformer(kubeClient *kubernetes.Clientset)
115	Run(stopCh <-chan struct{})
116	SetOpflexService(ofas opflexService, as v1.Service,
117	external bool, key string, sp v1.ServicePort) bool
118	}
119
120	type serviceEndpoint struct {
121	agent *HostAgent
122	}
123	type serviceEndpointSlice struct {
124	agent *HostAgent
125	}
126
127	func (sep serviceEndpoint) InitClientInformer(kubeClient kubernetes.Clientset) {	×
128	sep.agent.initEndpointsInformerFromClient(kubeClient)	×
129	}	×
130
131	func (seps serviceEndpointSlice) InitClientInformer(kubeClient kubernetes.Clientset) {	×
132	seps.agent.initEndpointSliceInformerFromClient(kubeClient)	×
133	}	×
134
135	func (sep *serviceEndpoint) Run(stopCh <-chan struct{}) {	1✔
136	go sep.agent.endpointsInformer.Run(stopCh)	1✔
137	cache.WaitForCacheSync(stopCh, sep.agent.endpointsInformer.HasSynced)	1✔
138	}	1✔
139
140	func (seps *serviceEndpointSlice) Run(stopCh <-chan struct{}) {	1✔
141	go seps.agent.endpointSliceInformer.Run(stopCh)	1✔
142	cache.WaitForCacheSync(stopCh, seps.agent.endpointSliceInformer.HasSynced)	1✔
143	}	1✔
144
145	func NewHostAgent(config HostAgentConfig, env Environment, log logrus.Logger) *HostAgent {	1✔
146	ha := &HostAgent{	1✔
147	log: log,	1✔
148	config: config,	1✔
149	env: env,	1✔
150	opflexEps: make(map[string][]*opflexEndpoint),	1✔
151	opflexServices: make(map[string]*opflexService),	1✔
152	epMetadata: make(map[string]map[string]*md.ContainerMetadata),	1✔
153	podIpToName: make(map[string]string),	1✔
154	cniToPodID: make(map[string]string),	1✔
155	podUidToName: make(map[string]string),	1✔
156	nodePodIfEPs: make(map[string]*opflexEndpoint),	1✔
157		1✔
158	podIps: ipam.NewIpCache(),	1✔
159		1✔
160	ignoreOvsPorts: make(map[string][]string),	1✔
161		1✔
162	netNsFuncChan: make(chan func()),	1✔
163	opflexSnatGlobalInfos: make(map[string][]*opflexSnatGlobalInfo),	1✔
164	opflexSnatLocalInfos: make(map[string]*opflexSnatLocalInfo),	1✔
165	snatPods: make(map[string]map[string]ResourceType),	1✔
166	snatPolicyLabels: make(map[string]map[string]ResourceType),	1✔
167	snatPolicyCache: make(map[string]*snatpolicy.SnatPolicy),	1✔
168	servicetoPodUids: make(map[string]map[string]struct{}),	1✔
169	podtoServiceUids: make(map[string]map[string]string),	1✔
170	syncQueue: workqueue.NewNamedRateLimitingQueue(	1✔
171	&workqueue.BucketRateLimiter{	1✔
172	Limiter: rate.NewLimiter(rate.Limit(10), int(10)),	1✔
173	}, "sync"),	1✔
174	ocServices: []opflexOcService{	1✔
175	{	1✔
176	RouterInternalDefault,	1✔
177	OpenShiftIngressNs,	1✔
178	},	1✔
179	},	1✔
180	}	1✔
181		1✔
182	ha.syncProcessors = map[string]func() bool{	1✔
183	"eps": ha.syncEps,	1✔
184	"services": ha.syncServices,	1✔
185	"opflexServer": ha.syncOpflexServer,	1✔
186	"snat": ha.syncSnat,	1✔
187	"snatnodeInfo": ha.syncSnatNodeInfo,	1✔
188	"rdconfig": ha.syncRdConfig,	1✔
189	"snatLocalInfo": ha.UpdateLocalInfoCr,	1✔
190	"nodepodifs": ha.syncNodePodIfs}	1✔
191		1✔
192	if ha.config.EPRegistry == "k8s" {	1✔
193	cfg, err := rest.InClusterConfig()	×
194	if err != nil {	×
195	log.Errorf("ERROR getting cluster config: %v", err)	×
196	return ha	×
197	}	×
198	aciawClient, err := crdclientset.NewForConfig(cfg)	×
199	if err != nil {	×
200	log.Errorf("ERROR getting crd client for registry: %v", err)	×
201	return ha	×
202	}	×
203	ha.crdClient = aciawClient.AciV1()	×
204	}
205	if ha.config.EnableNodePodIF {	1✔
206	cfg, err := rest.InClusterConfig()	×
207	if err != nil {	×
208	log.Errorf("ERROR getting cluster config: %v", err)	×
209	return ha	×
210	}	×
211	nodepodifClient, err := nodepodifclset.NewForConfig(cfg)	×
212	if err != nil {	×
213	log.Errorf("ERROR getting nodepodif client for enabling NodePodIF: %v", err)	×
214	return ha	×
215	}	×
216	ha.nodePodIFClient = nodepodifClient.AciV1()	×
217	}
218	return ha	1✔
219	}
220
221	func addPodRoute(ipn types.IPNet, dev string, src string) error {	×
222	link, err := netlink.LinkByName(dev)	×
223	if err != nil {	×
224	return err	×
225	}	×
226	if err := netlink.LinkSetUp(link); err != nil {	×
227	return err	×
228	}	×
229	ipsrc := net.ParseIP(src)	×
230	dst := &net.IPNet{	×
231	IP: ipn.IP,	×
232	Mask: ipn.Mask,	×
233	}	×
234	route := netlink.Route{LinkIndex: link.Attrs().Index, Dst: dst, Src: ipsrc}	×
235	return netlink.RouteAdd(&route)	×
236	}
237
238	func (agent *HostAgent) ReadSnatPolicyLabel(key string) (map[string]ResourceType, bool) {	1✔
239	agent.snatPolicyLabelMutex.RLock()	1✔
240	defer agent.snatPolicyLabelMutex.RUnlock()	1✔
241	value, ok := agent.snatPolicyLabels[key]	1✔
242	return value, ok	1✔
243	}	1✔
244
245	func (agent *HostAgent) WriteSnatPolicyLabel(key string, policy string, res ResourceType) {	1✔
246	agent.snatPolicyLabelMutex.Lock()	1✔
247	defer agent.snatPolicyLabelMutex.Unlock()	1✔
248	agent.snatPolicyLabels[key][policy] = res	1✔
249	}	1✔
250
251	func (agent *HostAgent) WriteNewSnatPolicyLabel(key string) {	1✔
252	agent.snatPolicyLabelMutex.Lock()	1✔
253	defer agent.snatPolicyLabelMutex.Unlock()	1✔
254	agent.snatPolicyLabels[key] = make(map[string]ResourceType)	1✔
255	}	1✔
256
257	func (agent *HostAgent) DeleteSnatPolicyLabelEntry(key string, policy string) {	1✔
258	agent.snatPolicyLabelMutex.Lock()	1✔
259	defer agent.snatPolicyLabelMutex.Unlock()	1✔
260	delete(agent.snatPolicyLabels[key], policy)	1✔
261	}	1✔
262
263	func (agent *HostAgent) DeleteSnatPolicyLabel(key string) {	1✔
264	agent.snatPolicyLabelMutex.Lock()	1✔
265	defer agent.snatPolicyLabelMutex.Unlock()	1✔
266	delete(agent.snatPolicyLabels, key)	1✔
267	}	1✔
268
269	func (agent *HostAgent) DeleteMatchingSnatPolicyLabel(policy string) {	1✔
270	agent.snatPolicyLabelMutex.Lock()	1✔
271	defer agent.snatPolicyLabelMutex.Unlock()	1✔
272	for key, v := range agent.snatPolicyLabels {	2✔
273	if _, ok := v[policy]; ok {	2✔
274	delete(agent.snatPolicyLabels[key], policy)	1✔
275	}	1✔
276	}
277	}
278
279	func (agent *HostAgent) Init() {	×
280	agent.log.Debug("Initializing endpoint CNI metadata")	×
281	err := md.LoadMetadata(agent.config.CniMetadataDir,	×
282	agent.config.CniNetwork, &agent.epMetadata)	×
283	if err != nil {	×
284	panic(err.Error())	×
285	}
286	agent.log.Info("Loaded cached endpoint CNI metadata: ", len(agent.epMetadata))	×
287	agent.buildUsedIPs()	×
288	// check if the cluster supports endpoint slices	×
289	// if cluster doesn't have the support fallback to endpoints	×
290	kubeClient := agent.env.(*K8sEnvironment).kubeClient	×
291	if util.IsEndPointSlicesSupported(kubeClient) {	×
292	agent.serviceEndPoints = &serviceEndpointSlice{}	×
293	agent.serviceEndPoints.(*serviceEndpointSlice).agent = agent	×
294	agent.log.Info("Initializing ServiceEndpointSlices")	×
295	} else {	×
296	agent.serviceEndPoints = &serviceEndpoint{}	×
297	agent.serviceEndPoints.(*serviceEndpoint).agent = agent	×
298	agent.log.Info("Initializing ServiceEndpoints")	×
299	}	×
300	err = agent.env.Init(agent)	×
301	if err != nil {	×
302	panic(err.Error())	×
303	}
304	}
305
306	func (agent *HostAgent) ScheduleSync(syncType string) {	1✔
307	agent.syncQueue.AddRateLimited(syncType)	1✔
308	}	1✔
309
310	func (agent *HostAgent) scheduleSyncEps() {	1✔
311	agent.ScheduleSync("eps")	1✔
312	agent.scheduleSyncNodePodIfs()	1✔
313	}	1✔
314
315	func (agent *HostAgent) scheduleSyncServices() {	1✔
316	agent.ScheduleSync("services")	1✔
317	}	1✔
318
319	func (agent *HostAgent) scheduleSyncSnats() {	1✔
320	agent.ScheduleSync("snat")	1✔
321	}	1✔
322
323	func (agent *HostAgent) scheduleSyncOpflexServer() {	×
324	agent.ScheduleSync("opflexServer")	×
325	}	×
326	func (agent *HostAgent) scheduleSyncNodeInfo() {	1✔
327	agent.ScheduleSync("snatnodeInfo")	1✔
328	}	1✔
329	func (agent *HostAgent) scheduleSyncRdConfig() {	1✔
330	agent.ScheduleSync("rdconfig")	1✔
331	}	1✔
332	func (agent *HostAgent) scheduleSyncLocalInfo() {	1✔
333	agent.ScheduleSync("snatLocalInfo")	1✔
334	}	1✔
335	func (agent *HostAgent) scheduleSyncNodePodIfs() {	1✔
336	agent.ScheduleSync("nodepodifs")	1✔
337	}	1✔
338
339	func (agent *HostAgent) runTickers(stopCh <-chan struct{}) {	1✔
340	ticker := time.NewTicker(time.Second * 30)	1✔
341	defer ticker.Stop()	1✔
342		1✔
343	for {	2✔
344	select {	1✔
345	case <-ticker.C:	×
346	agent.updateOpflexConfig()	×
347	case <-stopCh:	1✔
348	return	1✔
349	}
350	}
351	}
352
353	func (agent *HostAgent) processSyncQueue(queue workqueue.RateLimitingInterface,
354	queueStop <-chan struct{}) {	1✔
355		1✔
356	go wait.Until(func() {	2✔
357	for {	2✔
358	syncType, quit := queue.Get()	1✔
359	if quit {	2✔
360	break	1✔
361	}
362
363	var requeue bool	1✔
364	switch syncType := syncType.(type) {	1✔
365	case string:	1✔
366	if f, ok := agent.syncProcessors[syncType]; ok {	2✔
367	requeue = f()	1✔
368	}	1✔
369	}
370	if requeue {	1✔
371	queue.AddRateLimited(syncType)	×
372	} else {	1✔
373	queue.Forget(syncType)	1✔
374	}	1✔
375	queue.Done(syncType)	1✔
376
377	}
378	}, time.Second, queueStop)
379	<-queueStop	1✔
380	queue.ShutDown()	1✔
381	}
382
383	func (agent *HostAgent) EnableSync() (changed bool) {	1✔
384	changed = false	1✔
385	agent.indexMutex.Lock()	1✔
386	if agent.syncEnabled == false {	2✔
387	agent.syncEnabled = true	1✔
388	changed = true	1✔
389	}	1✔
390	agent.indexMutex.Unlock()	1✔
391	if changed {	2✔
392	agent.log.Info("Enabling OpFlex endpoint and service sync")	1✔
393	agent.scheduleSyncServices()	1✔
394	agent.scheduleSyncEps()	1✔
395	agent.scheduleSyncSnats()	1✔
396	agent.scheduleSyncNodeInfo()	1✔
397	agent.scheduleSyncNodePodIfs()	1✔
398	}	1✔
399	return	1✔
400	}
401
402	func (agent *HostAgent) Run(stopCh <-chan struct{}) {	1✔
403	err := agent.populateSnatLocalInfos()	1✔
404	if err != nil {	1✔
405	agent.log.Error("Failed to populate opflexSnatLocalInfos ", err.Error())	×
406	panic(err.Error())	×
407	}
408	syncEnabled, err := agent.env.PrepareRun(stopCh)	1✔
409	if err != nil {	1✔
410	panic(err.Error())	×
411	}
412	if agent.config.OpFlexEndpointDir == "" \|\|	1✔
413	agent.config.OpFlexServiceDir == "" \|\|	1✔
414	agent.config.OpFlexSnatDir == "" {	2✔
415	agent.log.Warn("OpFlex endpoint,service or snat directories not set")	1✔
416		1✔
417	} else {	2✔
418	if syncEnabled {	2✔
419	agent.EnableSync()	1✔
420	}	1✔
421	go agent.processSyncQueue(agent.syncQueue, stopCh)	1✔
422	}
423
424	agent.log.Info("Starting endpoint RPC")	1✔
425	err = agent.runEpRPC(stopCh)	1✔
426	if err != nil {	1✔
427	panic(err.Error())	×
428	}
429
430	agent.cleanupSetup()	1✔
431	}

noironetworks / aci-containers / 6757

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