8274

Committed 06 Dec 2023 10:05AM UTC coverage: 53.692% (-0.1%) from 53.79%

Build # 8274

Build Type

Pull #1206

travis-pro

Committed by

web-flow

Commit Message

Merge cbf87aef6 into afa25a6a5

Pull Request Pull Request #1206: Added paremeter enable-opflex-agent-reconnect

Run Details

5 of 29 new or added lines in 3 files covered. (17.24%)

17 existing lines in 3 files now uncovered.

13263 of 24702 relevant lines covered (53.69%)

0.6 hits per line

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

37.33

/pkg/hostagent/nodes.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.

// Handlers for node updates.

package hostagent

import (
        "encoding/json"
        "fmt"
        "os"
        "path/filepath"
        "reflect"
        "strings"
        "time"

        "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/runtime"
        "k8s.io/apimachinery/pkg/watch"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/tools/cache"
        "k8s.io/kubernetes/pkg/controller"

        "context"

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

const (
        hostVethEP   = "veth_host_ac.ep"
        hostVethName = "veth_host"
)

func (agent *HostAgent) initNodeInformerFromClient(
        kubeClient *kubernetes.Clientset) {
        agent.initNodeInformerBase(
                &cache.ListWatch{
                        ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                                options.FieldSelector =
                                        fields.Set{"metadata.name": agent.config.NodeName}.String()
                                obj, err := kubeClient.CoreV1().Nodes().List(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to list Nodes during initialization of NodeInformer: %s", err)
                                }
                                return obj, err
                        },
                        WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                                options.FieldSelector =
                                        fields.Set{"metadata.name": agent.config.NodeName}.String()
                                obj, err := kubeClient.CoreV1().Nodes().Watch(context.TODO(), options)
                                if err != nil {
                                        agent.log.Fatalf("Failed to watch Nodes during initialization of NodeInformer: %s", err)
                                }
                                return obj, err
                        },
                })
}

func (agent *HostAgent) initNodeInformerBase(listWatch *cache.ListWatch) {
        agent.nodeInformer = cache.NewSharedIndexInformer(
                listWatch,
                &v1.Node{},
                controller.NoResyncPeriodFunc(),
                cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
        )
        agent.nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: func(obj interface{}) {
                        agent.nodeChanged(obj)
                },
                UpdateFunc: func(oldobj interface{}, newobj interface{}) {
                        agent.nodeChanged(oldobj, newobj)
                },
                DeleteFunc: func(obj interface{}) {
                        agent.nodeDeleted(obj)
                },
        })
}

func (agent *HostAgent) nodeChanged(obj ...interface{}) {
        updateServices := false
        var node *v1.Node
        if len(obj) == 2 {
                oldnode := obj[0].(*v1.Node)
                node = obj[1].(*v1.Node)
                if !reflect.DeepEqual(node.ObjectMeta.Labels, oldnode.ObjectMeta.Labels) {
                        updateServices = true
                        agent.log.Infof("Node label changed for: %s, Updating services", node.ObjectMeta.Name)
                }
        } else {
                node = obj[0].(*v1.Node)
        }
        if node.ObjectMeta.Name != agent.config.NodeName {
                agent.log.Error("Got incorrect node update for ", node.ObjectMeta.Name)
                return
        }

        agent.indexMutex.Lock()

        if agent.config.EnableOpflexAgentReconnect {
                nodeAciPod, acipodok := node.ObjectMeta.Annotations[metadata.NodeAciPodAnnotation]
                if acipodok {
                        if agent.nodeAciPodAnnotation != nodeAciPod && nodeAciPod == "none" {
                                agent.informOpflexAgent(nodeAciPod)
                        }
                        agent.nodeAciPodAnnotation = nodeAciPod
                }
        }

        if agent.config.AciMultipod {
                aciPod, acipodok := node.ObjectMeta.Annotations[metadata.AciPodAnnotation]
                if acipodok {
                        if agent.aciPodAnnotation != aciPod {
                                agent.doDhcpRenew(aciPod)
                        }
                        agent.aciPodAnnotation = aciPod
                }
        }

        pnet, ok := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]
        if ok {
                agent.updateIpamAnnotation(pnet)
        }

        {
                var newServiceEp metadata.ServiceEndpoint
                epval, ok := node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation]
                if ok {
                        err := json.Unmarshal([]byte(epval), &newServiceEp)
                        if err != nil {
                                agent.log.WithFields(logrus.Fields{
                                        "epval": epval,
                                }).Warn("Could not parse node ",
                                        "service endpoint annotation: ", err)
                        }
                }
                if !reflect.DeepEqual(newServiceEp, agent.serviceEp) {
                        agent.log.WithFields(logrus.Fields{
                                "epval": epval,
                        }).Info("Updated service endpoint")
                        agent.serviceEp = newServiceEp
                        // this case can be posible when there is a default snatpolicy present
                        // And nodeinfo service EP is not annotated
                        if _, ok := agent.opflexServices[SnatService]; ok {
                                agent.opflexServices[SnatService].InterfaceIp = agent.serviceEp.Ipv4.String()
                                agent.log.Infof("Updated Snat service-ext file: %s", agent.serviceEp.Ipv4.String())
                        }
                        updateServices = true
                }
        }

        gotVtep := false
        if agent.vtepIP == "" {
                for _, a := range node.Status.Addresses {
                        if a.Type == v1.NodeInternalIP {
                                agent.vtepIP = a.Address
                                agent.log.Infof("vtepIP: %s", agent.vtepIP)
                                gotVtep = true
                        }
                }
        }

        agent.indexMutex.Unlock()
        if gotVtep {
                agent.routeInit()
                if agent.crdClient != nil {
                        agent.registerHostVeth()
                }
        }

        if updateServices {
                agent.updateAllServices()
        }
}

func (agent *HostAgent) registerHostVeth() {
        go func() {
                for {
                        ep := &opflexEndpoint{}
                        epfile := filepath.Join(agent.config.OpFlexEndpointDir, hostVethEP)
                        datacont, err := os.ReadFile(epfile)
                        if err != nil {
                                agent.log.Errorf("Unable to read %s - %v", epfile, err)
                                return
                        }

                        err = json.Unmarshal(datacont, ep)
                        if err != nil {
                                agent.log.Errorf("Unable to read %s - %v", epfile, err)
                                return
                        }

                        vmName := ep.Attributes["vm-name"]
                        if !strings.Contains(vmName, agent.vtepIP) {
                                vmName = fmt.Sprintf("%s.%s", vmName, agent.vtepIP)
                                ep.Attributes["vm-name"] = vmName
                        }
                        agent.log.Infof("-- Adding %+v to registry", ep)
                        agent.EPRegAdd(ep)
                        if ep.registered {
                                return
                        }
                        time.Sleep(5 * time.Second)
                }
        }()
}

func (agent *HostAgent) nodeDeleted(obj interface{}) {
        agent.indexMutex.Lock()
        defer agent.indexMutex.Unlock()
}

func (agent *HostAgent) routeInit() {
        for _, nc := range agent.config.NetConfig {
                err := addPodRoute(nc.Subnet, hostVethName, agent.vtepIP)
                if err != nil {
                        agent.log.Errorf("### Could not add route for subnet %+v reason: %s", nc.Subnet, err)
                        continue
                }
                agent.log.Infof("VtepIP: %s, subnet: %+v, interface: %s", agent.vtepIP, nc.Subnet, hostVethName)
        }
}

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	// Handlers for node updates.
16
17	package hostagent
18
19	import (
20	"encoding/json"
21	"fmt"
22	"os"
23	"path/filepath"
24	"reflect"
25	"strings"
26	"time"
27
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/runtime"
33	"k8s.io/apimachinery/pkg/watch"
34	"k8s.io/client-go/kubernetes"
35	"k8s.io/client-go/tools/cache"
36	"k8s.io/kubernetes/pkg/controller"
37
38	"context"
39
40	"github.com/noironetworks/aci-containers/pkg/metadata"
41	)
42
43	const (
44	hostVethEP = "veth_host_ac.ep"
45	hostVethName = "veth_host"
46	)
47
48	func (agent *HostAgent) initNodeInformerFromClient(
49	kubeClient *kubernetes.Clientset) {	×
50	agent.initNodeInformerBase(	×
51	&cache.ListWatch{	×
52	ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {	×
53	options.FieldSelector =	×
54	fields.Set{"metadata.name": agent.config.NodeName}.String()	×
55	obj, err := kubeClient.CoreV1().Nodes().List(context.TODO(), options)	×
56	if err != nil {	×
57	agent.log.Fatalf("Failed to list Nodes during initialization of NodeInformer: %s", err)	×
58	}	×
59	return obj, err	×
60	},
61	WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {	×
62	options.FieldSelector =	×
63	fields.Set{"metadata.name": agent.config.NodeName}.String()	×
64	obj, err := kubeClient.CoreV1().Nodes().Watch(context.TODO(), options)	×
65	if err != nil {	×
66	agent.log.Fatalf("Failed to watch Nodes during initialization of NodeInformer: %s", err)	×
67	}	×
68	return obj, err	×
69	},
70	})
71	}
72
73	func (agent HostAgent) initNodeInformerBase(listWatch cache.ListWatch) {	1✔
74	agent.nodeInformer = cache.NewSharedIndexInformer(	1✔
75	listWatch,	1✔
76	&v1.Node{},	1✔
77	controller.NoResyncPeriodFunc(),	1✔
78	cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},	1✔
79	)	1✔
80	agent.nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{	1✔
81	AddFunc: func(obj interface{}) {	2✔
82	agent.nodeChanged(obj)	1✔
83	},	1✔
84	UpdateFunc: func(oldobj interface{}, newobj interface{}) {	1✔
85	agent.nodeChanged(oldobj, newobj)	1✔
86	},	1✔
87	DeleteFunc: func(obj interface{}) {	×
88	agent.nodeDeleted(obj)	×
89	},	×
90	})
91	}
92
93	func (agent *HostAgent) nodeChanged(obj ...interface{}) {	1✔
94	updateServices := false	1✔
95	var node *v1.Node	1✔
96	if len(obj) == 2 {	2✔
97	oldnode := obj[0].(*v1.Node)	1✔
98	node = obj[1].(*v1.Node)	1✔
99	if !reflect.DeepEqual(node.ObjectMeta.Labels, oldnode.ObjectMeta.Labels) {	1✔
100	updateServices = true	×
101	agent.log.Infof("Node label changed for: %s, Updating services", node.ObjectMeta.Name)	×
102	}	×
103	} else {	1✔
104	node = obj[0].(*v1.Node)	1✔
105	}	1✔
106	if node.ObjectMeta.Name != agent.config.NodeName {	1✔
107	agent.log.Error("Got incorrect node update for ", node.ObjectMeta.Name)	×
108	return	×
109	}	×
110
111	agent.indexMutex.Lock()	1✔
112		1✔
113	if agent.config.EnableOpflexAgentReconnect {	1✔
NEW 114	nodeAciPod, acipodok := node.ObjectMeta.Annotations[metadata.NodeAciPodAnnotation]	×
NEW 115	if acipodok {	×
NEW 116	if agent.nodeAciPodAnnotation != nodeAciPod && nodeAciPod == "none" {	×
NEW 117	agent.informOpflexAgent(nodeAciPod)	×
NEW 118	}	×
NEW 119	agent.nodeAciPodAnnotation = nodeAciPod	×
120	}
121	}
122
123	if agent.config.AciMultipod {	1✔
124	aciPod, acipodok := node.ObjectMeta.Annotations[metadata.AciPodAnnotation]	×
125	if acipodok {	×
126	if agent.aciPodAnnotation != aciPod {	×
127	agent.doDhcpRenew(aciPod)	×
128	}	×
129	agent.aciPodAnnotation = aciPod	×
130	}
131	}
132
133	pnet, ok := node.ObjectMeta.Annotations[metadata.PodNetworkRangeAnnotation]	1✔
134	if ok {	2✔
135	agent.updateIpamAnnotation(pnet)	1✔
136	}	1✔
137
138	{	1✔
139	var newServiceEp metadata.ServiceEndpoint	1✔
140	epval, ok := node.ObjectMeta.Annotations[metadata.ServiceEpAnnotation]	1✔
141	if ok {	2✔
142	err := json.Unmarshal([]byte(epval), &newServiceEp)	1✔
143	if err != nil {	1✔
144	agent.log.WithFields(logrus.Fields{	×
145	"epval": epval,	×
146	}).Warn("Could not parse node ",	×
147	"service endpoint annotation: ", err)	×
148	}	×
149	}
150	if !reflect.DeepEqual(newServiceEp, agent.serviceEp) {	2✔
151	agent.log.WithFields(logrus.Fields{	1✔
152	"epval": epval,	1✔
153	}).Info("Updated service endpoint")	1✔
154	agent.serviceEp = newServiceEp	1✔
155	// this case can be posible when there is a default snatpolicy present	1✔
156	// And nodeinfo service EP is not annotated	1✔
157	if _, ok := agent.opflexServices[SnatService]; ok {	1✔
158	agent.opflexServices[SnatService].InterfaceIp = agent.serviceEp.Ipv4.String()	×
159	agent.log.Infof("Updated Snat service-ext file: %s", agent.serviceEp.Ipv4.String())	×
160	}	×
161	updateServices = true	1✔
162	}
163	}
164
165	gotVtep := false	1✔
166	if agent.vtepIP == "" {	2✔
167	for _, a := range node.Status.Addresses {	1✔
168	if a.Type == v1.NodeInternalIP {	×
169	agent.vtepIP = a.Address	×
170	agent.log.Infof("vtepIP: %s", agent.vtepIP)	×
171	gotVtep = true	×
172	}	×
173	}
174	}
175
176	agent.indexMutex.Unlock()	1✔
177	if gotVtep {	1✔
178	agent.routeInit()	×
179	if agent.crdClient != nil {	×
180	agent.registerHostVeth()	×
181	}	×
182	}
183
184	if updateServices {	2✔
185	agent.updateAllServices()	1✔
186	}	1✔
187	}
188
189	func (agent *HostAgent) registerHostVeth() {	×
190	go func() {	×
191	for {	×
192	ep := &opflexEndpoint{}	×
193	epfile := filepath.Join(agent.config.OpFlexEndpointDir, hostVethEP)	×
194	datacont, err := os.ReadFile(epfile)	×
195	if err != nil {	×
196	agent.log.Errorf("Unable to read %s - %v", epfile, err)	×
197	return	×
198	}	×
199
200	err = json.Unmarshal(datacont, ep)	×
201	if err != nil {	×
202	agent.log.Errorf("Unable to read %s - %v", epfile, err)	×
203	return	×
204	}	×
205
206	vmName := ep.Attributes["vm-name"]	×
207	if !strings.Contains(vmName, agent.vtepIP) {	×
208	vmName = fmt.Sprintf("%s.%s", vmName, agent.vtepIP)	×
209	ep.Attributes["vm-name"] = vmName	×
210	}	×
211	agent.log.Infof("-- Adding %+v to registry", ep)	×
212	agent.EPRegAdd(ep)	×
213	if ep.registered {	×
214	return	×
215	}	×
216	time.Sleep(5 * time.Second)	×
217	}
218	}()
219	}
220
221	func (agent *HostAgent) nodeDeleted(obj interface{}) {	×
222	agent.indexMutex.Lock()	×
223	defer agent.indexMutex.Unlock()	×
224	}	×
225
226	func (agent *HostAgent) routeInit() {	×
227	for _, nc := range agent.config.NetConfig {	×
228	err := addPodRoute(nc.Subnet, hostVethName, agent.vtepIP)	×
229	if err != nil {	×
230	agent.log.Errorf("### Could not add route for subnet %+v reason: %s", nc.Subnet, err)	×
231	continue	×
232	}
233	agent.log.Infof("VtepIP: %s, subnet: %+v, interface: %s", agent.vtepIP, nc.Subnet, hostVethName)	×
234	}
235	}

noironetworks / aci-containers / 8274

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