4902

Committed 24 May 2021 04:00AM UTC coverage: 58.456%. First build

Build # 4902

Build Type

push

travis-ci-com

Committed by

snaiksat

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/<a class="double-link" href="https://git"><a class=hub.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639">bce736026">&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/bce7360261c19a2fbc1e84f5a098cbe2026ea639">bce736026</a><a href="https://github.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639">&lt;a href=&quot;https://github.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/noironetworks/aci-containers/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639&amp;quot;&amp;gt;bce736026&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/noironetworks/aci-containers/commit/bce7360261c19a2fbc1e84f5a098cbe2026ea639&amp;amp;quot;&amp;amp;gt;&amp;amp;amp;quot;&amp;amp;amp;gt;Revert &amp;amp;amp;amp;quot;podifs should always be created n&amp;lt;/a&amp;gt;ow&amp;a</a>mp;amp;quot;

This re... (continued)

Run Details

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

11234 of 19218 relevant lines covered (58.46%)

0.65 hits per line

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

61.43

/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 (
        "fmt"
        "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"
        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

        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
        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
        vtepIface             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
}

type Vtep struct {
        vtepIP    string
        vtepIface string
}

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

                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}

        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()
        }
        return ha
}

func getVtep() (Vtep, error) {
        var vtep Vtep
        ifaces, err := net.Interfaces()
        if err != nil {
                return vtep, err
        }
        for _, i := range ifaces {
                // FIXME -- hardcoded for now
                if i.Name != "enp0s8" {
                        continue
                }
                addrs, err := i.Addrs()
                if err != nil {
                        return vtep, err
                }
                for _, addr := range addrs {
                        var ip net.IP
                        switch v := addr.(type) {
                        case *net.IPAddr:
                                ip = v.IP
                                vtep.vtepIP = ip.String()
                                vtep.vtepIface = i.Name
                                return vtep, nil
                        }
                        // process IP address
                }
        }

        return vtep, fmt.Errorf("VTEP IP not found")
}

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

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) 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()
        }
        return
}

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

noironetworks / aci-containers / 4902

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