• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

kubeovn / kube-ovn / 19527788804

20 Nov 2025 06:25AM UTC coverage: 21.99% (+0.08%) from 21.909%
19527788804

Pull #5543

github

zbb88888
fix: 使用 any 替换 interface{} (Go 1.18+)

- 修复 golangci-lint modernize 检查
- 在 subnet_bigint_test.go 中替换 3 处 interface{}
- 在 subnet_bigint_e2e_test.go 中替换 4 处 interface{}
- 保持测试逻辑不变

Signed-off-by: zbb88888 <jmdxjsjgcxy@gmail.com>
Pull Request #5543: Controller gen

66 of 123 new or added lines in 9 files covered. (53.66%)

2 existing lines in 1 file now uncovered.

11187 of 50873 relevant lines covered (21.99%)

0.26 hits per line

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

5.8
/pkg/controller/pod.go
1
package controller
2

3
import (
4
        "context"
5
        "encoding/json"
6
        "errors"
7
        "fmt"
8
        "maps"
9
        "net"
10
        "slices"
11
        "strconv"
12
        "strings"
13
        "sync"
14
        "time"
15

16
        nadv1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
17
        nadutils "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/utils"
18
        "github.com/scylladb/go-set/strset"
19
        v1 "k8s.io/api/core/v1"
20
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
21
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
22
        "k8s.io/apimachinery/pkg/labels"
23
        "k8s.io/apimachinery/pkg/types"
24
        utilruntime "k8s.io/apimachinery/pkg/util/runtime"
25
        "k8s.io/client-go/kubernetes"
26
        "k8s.io/client-go/tools/cache"
27
        "k8s.io/klog/v2"
28
        "k8s.io/utils/ptr"
29

30
        kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
31
        "github.com/kubeovn/kube-ovn/pkg/ipam"
32
        "github.com/kubeovn/kube-ovn/pkg/ovs"
33
        "github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
34
        "github.com/kubeovn/kube-ovn/pkg/request"
35
        "github.com/kubeovn/kube-ovn/pkg/util"
36
)
37

38
type NamedPort struct {
39
        mutex sync.RWMutex
40
        // first key is namespace, second key is portName
41
        namedPortMap map[string]map[string]*util.NamedPortInfo
42
}
43

44
func NewNamedPort() *NamedPort {
×
45
        return &NamedPort{
×
46
                mutex:        sync.RWMutex{},
×
47
                namedPortMap: map[string]map[string]*util.NamedPortInfo{},
×
48
        }
×
49
}
×
50

51
func (n *NamedPort) AddNamedPortByPod(pod *v1.Pod) {
×
52
        n.mutex.Lock()
×
53
        defer n.mutex.Unlock()
×
54
        ns := pod.Namespace
×
55
        podName := pod.Name
×
56

×
57
        restartableInitContainers := make([]v1.Container, 0, len(pod.Spec.InitContainers))
×
58
        for i := range pod.Spec.InitContainers {
×
59
                if pod.Spec.InitContainers[i].RestartPolicy != nil &&
×
60
                        *pod.Spec.InitContainers[i].RestartPolicy == v1.ContainerRestartPolicyAlways {
×
61
                        restartableInitContainers = append(restartableInitContainers, pod.Spec.InitContainers[i])
×
62
                }
×
63
        }
64

65
        containers := slices.Concat(restartableInitContainers, pod.Spec.Containers)
×
66
        if len(containers) == 0 {
×
67
                return
×
68
        }
×
69

70
        for _, container := range containers {
×
71
                if len(container.Ports) == 0 {
×
72
                        continue
×
73
                }
74

75
                for _, port := range container.Ports {
×
76
                        if port.Name == "" || port.ContainerPort == 0 {
×
77
                                continue
×
78
                        }
79

80
                        if _, ok := n.namedPortMap[ns]; ok {
×
81
                                if _, ok := n.namedPortMap[ns][port.Name]; ok {
×
82
                                        if n.namedPortMap[ns][port.Name].PortID == port.ContainerPort {
×
83
                                                n.namedPortMap[ns][port.Name].Pods.Add(podName)
×
84
                                        } else {
×
85
                                                klog.Warningf("named port %s has already be defined with portID %d",
×
86
                                                        port.Name, n.namedPortMap[ns][port.Name].PortID)
×
87
                                        }
×
88
                                        continue
×
89
                                }
90
                        } else {
×
91
                                n.namedPortMap[ns] = make(map[string]*util.NamedPortInfo)
×
92
                        }
×
93
                        n.namedPortMap[ns][port.Name] = &util.NamedPortInfo{
×
94
                                PortID: port.ContainerPort,
×
95
                                Pods:   strset.New(podName),
×
96
                        }
×
97
                }
98
        }
99
}
100

101
func (n *NamedPort) DeleteNamedPortByPod(pod *v1.Pod) {
×
102
        n.mutex.Lock()
×
103
        defer n.mutex.Unlock()
×
104

×
105
        ns := pod.Namespace
×
106
        podName := pod.Name
×
107

×
108
        if pod.Spec.Containers == nil {
×
109
                return
×
110
        }
×
111

112
        for _, container := range pod.Spec.Containers {
×
113
                if container.Ports == nil {
×
114
                        continue
×
115
                }
116

117
                for _, port := range container.Ports {
×
118
                        if port.Name == "" {
×
119
                                continue
×
120
                        }
121

122
                        if _, ok := n.namedPortMap[ns]; !ok {
×
123
                                continue
×
124
                        }
125

126
                        if _, ok := n.namedPortMap[ns][port.Name]; !ok {
×
127
                                continue
×
128
                        }
129

130
                        if !n.namedPortMap[ns][port.Name].Pods.Has(podName) {
×
131
                                continue
×
132
                        }
133

134
                        n.namedPortMap[ns][port.Name].Pods.Remove(podName)
×
135
                        if n.namedPortMap[ns][port.Name].Pods.Size() == 0 {
×
136
                                delete(n.namedPortMap[ns], port.Name)
×
137
                                if len(n.namedPortMap[ns]) == 0 {
×
138
                                        delete(n.namedPortMap, ns)
×
139
                                }
×
140
                        }
141
                }
142
        }
143
}
144

145
func (n *NamedPort) GetNamedPortByNs(namespace string) map[string]*util.NamedPortInfo {
×
146
        n.mutex.RLock()
×
147
        defer n.mutex.RUnlock()
×
148

×
149
        if result, ok := n.namedPortMap[namespace]; ok {
×
150
                for portName, info := range result {
×
151
                        klog.Infof("namespace %s's namedPort portname is %s with info %v", namespace, portName, info)
×
152
                }
×
153
                return result
×
154
        }
155
        return nil
×
156
}
157

158
func isPodAlive(p *v1.Pod) bool {
×
159
        if !p.DeletionTimestamp.IsZero() && p.DeletionGracePeriodSeconds != nil {
×
160
                now := time.Now()
×
161
                deletionTime := p.DeletionTimestamp.Time
×
162
                gracePeriod := time.Duration(*p.DeletionGracePeriodSeconds) * time.Second
×
163
                if now.After(deletionTime.Add(gracePeriod)) {
×
164
                        return false
×
165
                }
×
166
        }
167
        return isPodStatusPhaseAlive(p)
×
168
}
169

170
func isPodStatusPhaseAlive(p *v1.Pod) bool {
×
171
        if p.Status.Phase == v1.PodSucceeded && p.Spec.RestartPolicy != v1.RestartPolicyAlways {
×
172
                return false
×
173
        }
×
174

175
        if p.Status.Phase == v1.PodFailed && p.Spec.RestartPolicy == v1.RestartPolicyNever {
×
176
                return false
×
177
        }
×
178

179
        if p.Status.Phase == v1.PodFailed && p.Status.Reason == "Evicted" {
×
180
                return false
×
181
        }
×
182
        return true
×
183
}
184

185
func (c *Controller) enqueueAddPod(obj any) {
×
186
        p := obj.(*v1.Pod)
×
187
        if p.Spec.HostNetwork {
×
188
                return
×
189
        }
×
190

191
        // Pod might be targeted by manual endpoints and we need to recompute its port mappings
192
        c.enqueueStaticEndpointUpdateInNamespace(p.Namespace)
×
193

×
194
        // TODO: we need to find a way to reduce duplicated np added to the queue
×
195
        if c.config.EnableNP {
×
196
                c.namedPort.AddNamedPortByPod(p)
×
197
                if p.Status.PodIP != "" {
×
198
                        for _, np := range c.podMatchNetworkPolicies(p) {
×
199
                                klog.V(3).Infof("enqueue update network policy %s", np)
×
200
                                c.updateNpQueue.Add(np)
×
201
                        }
×
202
                }
203
        }
204

205
        key := cache.MetaObjectToName(p).String()
×
206
        if !isPodAlive(p) {
×
207
                isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(p)
×
208
                isVMPod, vmName := isVMPod(p)
×
209
                if isStateful || (isVMPod && c.config.EnableKeepVMIP) {
×
210
                        if isStateful && isStatefulSetPodToDel(c.config.KubeClient, p, statefulSetName, statefulSetUID) {
×
211
                                klog.V(3).Infof("enqueue delete pod %s", key)
×
212
                                c.deletingPodObjMap.Store(key, p)
×
213
                                c.deletePodQueue.Add(key)
×
214
                        }
×
215
                        if isVMPod && c.isVMToDel(p, vmName) {
×
216
                                klog.V(3).Infof("enqueue delete pod %s", key)
×
217
                                c.deletingPodObjMap.Store(key, p)
×
218
                                c.deletePodQueue.Add(key)
×
219
                        }
×
220
                } else {
×
221
                        klog.V(3).Infof("enqueue delete pod %s", key)
×
222
                        c.deletingPodObjMap.Store(key, p)
×
223
                        c.deletePodQueue.Add(key)
×
224
                }
×
225
                return
×
226
        }
227

228
        need, err := c.podNeedSync(p)
×
229
        if err != nil {
×
230
                klog.Errorf("invalid pod net: %v", err)
×
231
                return
×
232
        }
×
233
        if need {
×
234
                klog.Infof("enqueue add pod %s", key)
×
235
                c.addOrUpdatePodQueue.Add(key)
×
236
        }
×
237

238
        if err = c.handlePodEventForVpcEgressGateway(p); err != nil {
×
239
                klog.Errorf("failed to handle pod event for vpc egress gateway: %v", err)
×
240
        }
×
241
}
242

243
func (c *Controller) enqueueDeletePod(obj any) {
×
244
        var p *v1.Pod
×
245
        switch t := obj.(type) {
×
246
        case *v1.Pod:
×
247
                p = t
×
248
        case cache.DeletedFinalStateUnknown:
×
249
                pod, ok := t.Obj.(*v1.Pod)
×
250
                if !ok {
×
251
                        klog.Warningf("unexpected object type: %T", t.Obj)
×
252
                        return
×
253
                }
×
254
                p = pod
×
255
        default:
×
256
                klog.Warningf("unexpected type: %T", obj)
×
257
                return
×
258
        }
259

260
        if p.Spec.HostNetwork {
×
261
                return
×
262
        }
×
263

264
        // Pod might be targeted by manual endpoints and we need to recompute its port mappings
265
        c.enqueueStaticEndpointUpdateInNamespace(p.Namespace)
×
266

×
267
        if c.config.EnableNP {
×
268
                c.namedPort.DeleteNamedPortByPod(p)
×
269
                for _, np := range c.podMatchNetworkPolicies(p) {
×
270
                        c.updateNpQueue.Add(np)
×
271
                }
×
272
        }
273

274
        if c.config.EnableANP {
×
275
                podNs, _ := c.namespacesLister.Get(p.Namespace)
×
276
                c.updateAnpsByLabelsMatch(podNs.Labels, p.Labels)
×
277
        }
×
278

279
        key := cache.MetaObjectToName(p).String()
×
280
        klog.Infof("enqueue delete pod %s", key)
×
281
        c.deletingPodObjMap.Store(key, p)
×
282
        c.deletePodQueue.Add(key)
×
283
}
284

285
func (c *Controller) enqueueUpdatePod(oldObj, newObj any) {
×
286
        oldPod := oldObj.(*v1.Pod)
×
287
        newPod := newObj.(*v1.Pod)
×
288

×
289
        // Pod might be targeted by manual endpoints and we need to recompute its port mappings
×
290
        c.enqueueStaticEndpointUpdateInNamespace(oldPod.Namespace)
×
291

×
292
        if oldPod.Annotations[util.AAPsAnnotation] != "" || newPod.Annotations[util.AAPsAnnotation] != "" {
×
293
                oldAAPs := strings.Split(oldPod.Annotations[util.AAPsAnnotation], ",")
×
294
                newAAPs := strings.Split(newPod.Annotations[util.AAPsAnnotation], ",")
×
295
                var vipNames []string
×
296
                for _, vipName := range oldAAPs {
×
297
                        vipNames = append(vipNames, strings.TrimSpace(vipName))
×
298
                }
×
299
                for _, vipName := range newAAPs {
×
300
                        vipName = strings.TrimSpace(vipName)
×
301
                        if !slices.Contains(vipNames, vipName) {
×
302
                                vipNames = append(vipNames, vipName)
×
303
                        }
×
304
                }
305
                for _, vipName := range vipNames {
×
306
                        if vip, err := c.virtualIpsLister.Get(vipName); err == nil {
×
307
                                if vip.Spec.Namespace != newPod.Namespace {
×
308
                                        continue
×
309
                                }
310
                                klog.Infof("enqueue update virtual parents for %s", vipName)
×
311
                                c.updateVirtualParentsQueue.Add(vipName)
×
312
                        }
313
                }
314
        }
315

316
        if oldPod.ResourceVersion == newPod.ResourceVersion {
×
317
                return
×
318
        }
×
319
        if newPod.Spec.HostNetwork {
×
320
                return
×
321
        }
×
322

323
        podNets, err := c.getPodKubeovnNets(newPod)
×
324
        if err != nil {
×
325
                klog.Errorf("failed to get newPod nets %v", err)
×
326
                return
×
327
        }
×
328

329
        key := cache.MetaObjectToName(newPod).String()
×
330
        if c.config.EnableNP {
×
331
                c.namedPort.AddNamedPortByPod(newPod)
×
332
                newNp := c.podMatchNetworkPolicies(newPod)
×
333
                if !maps.Equal(oldPod.Labels, newPod.Labels) {
×
334
                        oldNp := c.podMatchNetworkPolicies(oldPod)
×
335
                        for _, np := range util.DiffStringSlice(oldNp, newNp) {
×
336
                                c.updateNpQueue.Add(np)
×
337
                        }
×
338
                }
339

340
                for _, podNet := range podNets {
×
341
                        oldAllocated := oldPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]
×
342
                        newAllocated := newPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]
×
343
                        if oldAllocated != newAllocated {
×
344
                                for _, np := range newNp {
×
345
                                        klog.V(3).Infof("enqueue update network policy %s for pod %s", np, key)
×
346
                                        c.updateNpQueue.Add(np)
×
347
                                }
×
348
                                break
×
349
                        }
350
                }
351
        }
352

353
        if c.config.EnableANP {
×
354
                podNs, _ := c.namespacesLister.Get(newPod.Namespace)
×
355
                if !maps.Equal(oldPod.Labels, newPod.Labels) {
×
356
                        c.updateAnpsByLabelsMatch(podNs.Labels, newPod.Labels)
×
357
                }
×
358

359
                for _, podNet := range podNets {
×
360
                        oldAllocated := oldPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]
×
361
                        newAllocated := newPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]
×
362
                        if oldAllocated != newAllocated {
×
363
                                c.updateAnpsByLabelsMatch(podNs.Labels, newPod.Labels)
×
364
                                break
×
365
                        }
366
                }
367
        }
368

369
        isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(newPod)
×
370
        isVMPod, vmName := isVMPod(newPod)
×
371
        if !isPodStatusPhaseAlive(newPod) && !isStateful && !isVMPod {
×
372
                klog.V(3).Infof("enqueue delete pod %s", key)
×
373
                c.deletingPodObjMap.Store(key, newPod)
×
374
                c.deletePodQueue.Add(key)
×
375
                return
×
376
        }
×
377

378
        // enqueue delay
379
        var delay time.Duration
×
380
        if newPod.Spec.TerminationGracePeriodSeconds != nil {
×
381
                if !newPod.DeletionTimestamp.IsZero() {
×
382
                        delay = time.Until(newPod.DeletionTimestamp.Add(time.Duration(*newPod.Spec.TerminationGracePeriodSeconds) * time.Second))
×
383
                } else {
×
384
                        delay = time.Duration(*newPod.Spec.TerminationGracePeriodSeconds) * time.Second
×
385
                }
×
386
        }
387

388
        if !newPod.DeletionTimestamp.IsZero() && !isStateful && !isVMPod {
×
389
                go func() {
×
390
                        // In case node get lost and pod can not be deleted,
×
391
                        // the ip address will not be recycled
×
392
                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
×
393
                        c.deletingPodObjMap.Store(key, newPod)
×
394
                        c.deletePodQueue.AddAfter(key, delay)
×
395
                }()
×
396
                return
×
397
        }
398

399
        if err = c.handlePodEventForVpcEgressGateway(newPod); err != nil {
×
400
                klog.Errorf("failed to handle pod event for vpc egress gateway: %v", err)
×
401
        }
×
402

403
        // do not delete statefulset pod unless ownerReferences is deleted
404
        if isStateful && isStatefulSetPodToDel(c.config.KubeClient, newPod, statefulSetName, statefulSetUID) {
×
405
                go func() {
×
406
                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
×
407
                        c.deletingPodObjMap.Store(key, newPod)
×
408
                        c.deletePodQueue.AddAfter(key, delay)
×
409
                }()
×
410
                return
×
411
        }
412
        if isVMPod && c.isVMToDel(newPod, vmName) {
×
413
                go func() {
×
414
                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)
×
415
                        c.deletingPodObjMap.Store(key, newPod)
×
416
                        c.deletePodQueue.AddAfter(key, delay)
×
417
                }()
×
418
                return
×
419
        }
420
        klog.Infof("enqueue update pod %s", key)
×
421
        c.addOrUpdatePodQueue.Add(key)
×
422

×
423
        // security policy changed
×
424
        for _, podNet := range podNets {
×
425
                oldSecurity := oldPod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)]
×
426
                newSecurity := newPod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)]
×
427
                oldSg := oldPod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]
×
428
                newSg := newPod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]
×
429
                oldVips := oldPod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]
×
430
                newVips := newPod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]
×
431
                oldAAPs := oldPod.Annotations[util.AAPsAnnotation]
×
432
                newAAPs := newPod.Annotations[util.AAPsAnnotation]
×
433
                if oldSecurity != newSecurity || oldSg != newSg || oldVips != newVips || oldAAPs != newAAPs {
×
434
                        c.updatePodSecurityQueue.Add(key)
×
435
                        break
×
436
                }
437
        }
438
}
439

440
func (c *Controller) getPodKubeovnNets(pod *v1.Pod) ([]*kubeovnNet, error) {
1✔
441
        attachmentNets, err := c.getPodAttachmentNet(pod)
1✔
442
        if err != nil {
1✔
443
                klog.Error(err)
×
444
                return nil, err
×
445
        }
×
446

447
        podNets := attachmentNets
1✔
448
        // When Kube-OVN is run as non-primary CNI, we do not add default network configuration to pod.
1✔
449
        // We only add network attachment defined by the user to pod.
1✔
450
        if c.config.EnableNonPrimaryCNI {
2✔
451
                return podNets, nil
1✔
452
        }
1✔
453

454
        defaultSubnet, err := c.getPodDefaultSubnet(pod)
1✔
455
        if err != nil {
1✔
456
                klog.Error(err)
×
457
                return nil, err
×
458
        }
×
459

460
        // pod annotation default subnet not found
461
        if defaultSubnet == nil {
1✔
462
                klog.Errorf("pod %s/%s has no default subnet, skip adding default network", pod.Namespace, pod.Name)
×
463
                return attachmentNets, nil
×
464
        }
×
465

466
        if _, hasOtherDefaultNet := pod.Annotations[util.DefaultNetworkAnnotation]; !hasOtherDefaultNet {
2✔
467
                podNets = append(attachmentNets, &kubeovnNet{
1✔
468
                        Type:         providerTypeOriginal,
1✔
469
                        ProviderName: util.OvnProvider,
1✔
470
                        Subnet:       defaultSubnet,
1✔
471
                        IsDefault:    true,
1✔
472
                })
1✔
473
        }
1✔
474

475
        return podNets, nil
1✔
476
}
477

478
func (c *Controller) handleAddOrUpdatePod(key string) (err error) {
×
479
        now := time.Now()
×
480
        klog.Infof("handle add/update pod %s", key)
×
481

×
482
        namespace, name, err := cache.SplitMetaNamespaceKey(key)
×
483
        if err != nil {
×
484
                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
×
485
                return nil
×
486
        }
×
487

488
        c.podKeyMutex.LockKey(key)
×
489
        defer func() {
×
490
                _ = c.podKeyMutex.UnlockKey(key)
×
491
                last := time.Since(now)
×
492
                klog.Infof("take %d ms to handle add or update pod %s", last.Milliseconds(), key)
×
493
        }()
×
494

495
        pod, err := c.podsLister.Pods(namespace).Get(name)
×
496
        if err != nil {
×
497
                if k8serrors.IsNotFound(err) {
×
498
                        return nil
×
499
                }
×
500
                klog.Error(err)
×
501
                return err
×
502
        }
503
        if err := util.ValidatePodNetwork(pod.Annotations); err != nil {
×
504
                klog.Errorf("validate pod %s/%s failed: %v", namespace, name, err)
×
505
                c.recorder.Eventf(pod, v1.EventTypeWarning, "ValidatePodNetworkFailed", err.Error())
×
506
                return err
×
507
        }
×
508

509
        podNets, err := c.getPodKubeovnNets(pod)
×
510
        if err != nil {
×
511
                klog.Errorf("failed to get pod nets %v", err)
×
512
                return err
×
513
        }
×
514

515
        // check and do hotnoplug nic
516
        if pod, err = c.syncKubeOvnNet(pod, podNets); err != nil {
×
517
                klog.Errorf("failed to sync pod nets %v", err)
×
518
                return err
×
519
        }
×
520
        if pod == nil {
×
521
                // pod has been deleted
×
522
                return nil
×
523
        }
×
524
        needAllocatePodNets := needAllocateSubnets(pod, podNets)
×
525
        if len(needAllocatePodNets) != 0 {
×
526
                if pod, err = c.reconcileAllocateSubnets(pod, needAllocatePodNets); err != nil {
×
527
                        klog.Error(err)
×
528
                        return err
×
529
                }
×
530
                if pod == nil {
×
531
                        // pod has been deleted
×
532
                        return nil
×
533
                }
×
534
        }
535

536
        isVpcNatGw, vpcGwName := c.checkIsPodVpcNatGw(pod)
×
537
        if isVpcNatGw {
×
538
                if needRestartNatGatewayPod(pod) {
×
539
                        klog.Infof("restarting vpc nat gateway %s", vpcGwName)
×
540
                        c.addOrUpdateVpcNatGatewayQueue.Add(vpcGwName)
×
541
                }
×
542
        }
543

544
        // check if route subnet is need.
545
        return c.reconcileRouteSubnets(pod, needRouteSubnets(pod, podNets))
×
546
}
547

548
// do the same thing as add pod
549
func (c *Controller) reconcileAllocateSubnets(pod *v1.Pod, needAllocatePodNets []*kubeovnNet) (*v1.Pod, error) {
×
550
        namespace := pod.Namespace
×
551
        name := pod.Name
×
552
        klog.Infof("sync pod %s/%s allocated", namespace, name)
×
553

×
554
        vipsMap := c.getVirtualIPs(pod, needAllocatePodNets)
×
555
        isVMPod, vmName := isVMPod(pod)
×
556
        podType := getPodType(pod)
×
557
        podName := c.getNameByPod(pod)
×
558
        // todo: isVmPod, getPodType, getNameByPod has duplicated logic
×
559

×
560
        var err error
×
561
        var vmKey string
×
562
        if isVMPod && c.config.EnableKeepVMIP {
×
563
                vmKey = fmt.Sprintf("%s/%s", namespace, vmName)
×
564
        }
×
565
        // Avoid create lsp for already running pod in ovn-nb when controller restart
566
        patch := util.KVPatch{}
×
567
        for _, podNet := range needAllocatePodNets {
×
568
                // the subnet may changed when alloc static ip from the latter subnet after ns supports multi subnets
×
569
                v4IP, v6IP, mac, subnet, err := c.acquireAddress(pod, podNet)
×
570
                if err != nil {
×
571
                        c.recorder.Eventf(pod, v1.EventTypeWarning, "AcquireAddressFailed", err.Error())
×
572
                        klog.Error(err)
×
573
                        return nil, err
×
574
                }
×
575
                ipStr := util.GetStringIP(v4IP, v6IP)
×
576
                patch[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] = ipStr
×
577
                if mac == "" {
×
578
                        patch[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = nil
×
579
                } else {
×
580
                        patch[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = mac
×
581
                }
×
582
                patch[fmt.Sprintf(util.CidrAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.CIDRBlock
×
583
                patch[fmt.Sprintf(util.GatewayAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.Gateway
×
584
                if isOvnSubnet(podNet.Subnet) {
×
585
                        patch[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName)] = subnet.Name
×
586
                        if pod.Annotations[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] == "" {
×
587
                                patch[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] = c.config.PodNicType
×
588
                        }
×
589
                } else {
×
590
                        patch[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName)] = nil
×
591
                        patch[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] = nil
×
592
                }
×
593
                patch[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] = "true"
×
594
                if vmKey != "" {
×
595
                        patch[fmt.Sprintf(util.VMAnnotationTemplate, podNet.ProviderName)] = vmName
×
596
                }
×
597
                if err := util.ValidateNetworkBroadcast(podNet.Subnet.Spec.CIDRBlock, ipStr); err != nil {
×
598
                        klog.Errorf("validate pod %s/%s failed: %v", namespace, name, err)
×
599
                        c.recorder.Eventf(pod, v1.EventTypeWarning, "ValidatePodNetworkFailed", err.Error())
×
600
                        return nil, err
×
601
                }
×
602

603
                if podNet.Type != providerTypeIPAM {
×
604
                        if (subnet.Spec.Vlan == "" || subnet.Spec.LogicalGateway || subnet.Spec.U2OInterconnection) && subnet.Spec.Vpc != "" {
×
605
                                patch[fmt.Sprintf(util.LogicalRouterAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.Vpc
×
606
                        }
×
607

608
                        if subnet.Spec.Vlan != "" {
×
609
                                vlan, err := c.vlansLister.Get(subnet.Spec.Vlan)
×
610
                                if err != nil {
×
611
                                        klog.Error(err)
×
612
                                        c.recorder.Eventf(pod, v1.EventTypeWarning, "GetVlanInfoFailed", err.Error())
×
613
                                        return nil, err
×
614
                                }
×
615
                                patch[fmt.Sprintf(util.VlanIDAnnotationTemplate, podNet.ProviderName)] = strconv.Itoa(vlan.Spec.ID)
×
616
                                patch[fmt.Sprintf(util.ProviderNetworkTemplate, podNet.ProviderName)] = vlan.Spec.Provider
×
617
                        }
618

619
                        portSecurity := false
×
620
                        if pod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)] == "true" {
×
621
                                portSecurity = true
×
622
                        }
×
623

624
                        vips := vipsMap[fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)]
×
625
                        for ip := range strings.SplitSeq(vips, ",") {
×
626
                                if ip != "" && net.ParseIP(ip) == nil {
×
627
                                        klog.Errorf("invalid vip address '%s' for pod %s", ip, name)
×
628
                                        vips = ""
×
629
                                        break
×
630
                                }
631
                        }
632

633
                        portName := ovs.PodNameToPortName(podName, namespace, podNet.ProviderName)
×
634
                        dhcpOptions := &ovs.DHCPOptionsUUIDs{
×
635
                                DHCPv4OptionsUUID: subnet.Status.DHCPv4OptionsUUID,
×
636
                                DHCPv6OptionsUUID: subnet.Status.DHCPv6OptionsUUID,
×
637
                        }
×
638

×
639
                        var oldSgList []string
×
640
                        if vmKey != "" {
×
641
                                existingLsp, err := c.OVNNbClient.GetLogicalSwitchPort(portName, true)
×
642
                                if err != nil {
×
643
                                        klog.Errorf("failed to get logical switch port %s: %v", portName, err)
×
644
                                        return nil, err
×
645
                                }
×
646
                                if existingLsp != nil {
×
647
                                        oldSgList, _ = c.getPortSg(existingLsp)
×
648
                                }
×
649
                        }
650

651
                        securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]
×
652
                        if err := c.OVNNbClient.CreateLogicalSwitchPort(subnet.Name, portName, ipStr, mac, podName, pod.Namespace,
×
653
                                portSecurity, securityGroupAnnotation, vips, podNet.Subnet.Spec.EnableDHCP, dhcpOptions, subnet.Spec.Vpc); err != nil {
×
654
                                c.recorder.Eventf(pod, v1.EventTypeWarning, "CreateOVNPortFailed", err.Error())
×
655
                                klog.Errorf("%v", err)
×
656
                                return nil, err
×
657
                        }
×
658

659
                        if pod.Annotations[fmt.Sprintf(util.Layer2ForwardAnnotationTemplate, podNet.ProviderName)] == "true" {
×
660
                                if err := c.OVNNbClient.EnablePortLayer2forward(portName); err != nil {
×
661
                                        c.recorder.Eventf(pod, v1.EventTypeWarning, "SetOVNPortL2ForwardFailed", err.Error())
×
662
                                        klog.Errorf("%v", err)
×
663
                                        return nil, err
×
664
                                }
×
665
                        }
666

667
                        if isVMPod {
×
668
                                if _, ok := pod.Labels["kubevirt.io/migrationJobUID"]; ok {
×
669
                                        if sourceNode, ok := pod.Labels["kubevirt.io/nodeName"]; ok && sourceNode != pod.Spec.NodeName {
×
670
                                                klog.Infof("VM pod %s/%s is migrating from %s to %s",
×
671
                                                        pod.Namespace, pod.Name, sourceNode, pod.Spec.NodeName)
×
672
                                                if err := c.OVNNbClient.SetLogicalSwitchPortMigrateOptions(portName, sourceNode, pod.Spec.NodeName); err != nil {
×
673
                                                        klog.Errorf("failed to set migrate options for VM pod lsp %s: %v", portName, err)
×
674
                                                        return nil, err
×
675
                                                }
×
676
                                        }
677
                                }
678
                        }
679

680
                        if securityGroupAnnotation != "" || oldSgList != nil {
×
681
                                securityGroups := strings.ReplaceAll(securityGroupAnnotation, " ", "")
×
682
                                newSgList := strings.Split(securityGroups, ",")
×
683
                                sgNames := util.UnionStringSlice(oldSgList, newSgList)
×
684
                                for _, sgName := range sgNames {
×
685
                                        if sgName != "" {
×
686
                                                c.syncSgPortsQueue.Add(sgName)
×
687
                                        }
×
688
                                }
689
                        }
690

691
                        if vips != "" {
×
692
                                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)
×
693
                        }
×
694
                }
695
                // CreatePort may fail, so put ip CR creation after CreatePort
696
                ipCRName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
×
697
                if err := c.createOrUpdateIPCR(ipCRName, podName, ipStr, mac, subnet.Name, pod.Namespace, pod.Spec.NodeName, podType); err != nil {
×
698
                        err = fmt.Errorf("failed to create ips CR %s.%s: %w", podName, pod.Namespace, err)
×
699
                        klog.Error(err)
×
700
                        return nil, err
×
701
                }
×
702
        }
703
        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(namespace), name, patch); err != nil {
×
704
                if k8serrors.IsNotFound(err) {
×
705
                        // Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.
×
706
                        // Then we need to recycle the resource again.
×
707
                        key := strings.Join([]string{namespace, name}, "/")
×
708
                        c.deletingPodObjMap.Store(key, pod)
×
709
                        c.deletePodQueue.AddRateLimited(key)
×
710
                        return nil, nil
×
711
                }
×
712
                klog.Errorf("failed to patch pod %s/%s: %v", namespace, name, err)
×
713
                return nil, err
×
714
        }
715

716
        if pod, err = c.config.KubeClient.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{}); err != nil {
×
717
                if k8serrors.IsNotFound(err) {
×
718
                        key := strings.Join([]string{namespace, name}, "/")
×
719
                        c.deletingPodObjMap.Store(key, pod)
×
720
                        c.deletePodQueue.AddRateLimited(key)
×
721
                        return nil, nil
×
722
                }
×
723
                klog.Errorf("failed to get pod %s/%s: %v", namespace, name, err)
×
724
                return nil, err
×
725
        }
726

727
        // Check if pod is a vpc nat gateway. Annotation set will have subnet provider name as prefix
728
        isVpcNatGw, vpcGwName := c.checkIsPodVpcNatGw(pod)
×
729
        if isVpcNatGw {
×
730
                klog.Infof("init vpc nat gateway pod %s/%s with name %s", namespace, name, vpcGwName)
×
731
                c.initVpcNatGatewayQueue.Add(vpcGwName)
×
732
        }
×
733

734
        return pod, nil
×
735
}
736

737
// do the same thing as update pod
738
func (c *Controller) reconcileRouteSubnets(pod *v1.Pod, needRoutePodNets []*kubeovnNet) error {
×
739
        // the lb-svc pod has dependencies on Running state, check it when pod state get updated
×
740
        if err := c.checkAndReInitLbSvcPod(pod); err != nil {
×
741
                klog.Errorf("failed to init iptable rules for load-balancer pod %s/%s: %v", pod.Namespace, pod.Name, err)
×
742
        }
×
743

744
        if len(needRoutePodNets) == 0 {
×
745
                return nil
×
746
        }
×
747

748
        namespace := pod.Namespace
×
749
        name := pod.Name
×
750
        podName := c.getNameByPod(pod)
×
751

×
752
        klog.Infof("sync pod %s/%s routed", namespace, name)
×
753

×
754
        node, err := c.nodesLister.Get(pod.Spec.NodeName)
×
755
        if err != nil {
×
756
                klog.Errorf("failed to get node %s: %v", pod.Spec.NodeName, err)
×
757
                return err
×
758
        }
×
759

760
        portGroups, err := c.OVNNbClient.ListPortGroups(map[string]string{"node": "", networkPolicyKey: ""})
×
761
        if err != nil {
×
762
                klog.Errorf("failed to list port groups: %v", err)
×
763
                return err
×
764
        }
×
765

766
        var nodePortGroups []string
×
767
        nodePortGroup := strings.ReplaceAll(node.Annotations[util.PortNameAnnotation], "-", ".")
×
768
        for _, pg := range portGroups {
×
769
                if pg.Name != nodePortGroup && pg.ExternalIDs["subnet"] == "" {
×
770
                        nodePortGroups = append(nodePortGroups, pg.Name)
×
771
                }
×
772
        }
773

774
        var podIP string
×
775
        var subnet *kubeovnv1.Subnet
×
776
        patch := util.KVPatch{}
×
777
        for _, podNet := range needRoutePodNets {
×
778
                // in case update handler overlap the annotation when cache is not in sync
×
779
                if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "" {
×
780
                        return fmt.Errorf("no address has been allocated to %s/%s", namespace, name)
×
781
                }
×
782

783
                podIP = pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]
×
784
                subnet = podNet.Subnet
×
785

×
786
                // Check if pod uses nodeSwitch subnet
×
787
                if subnet.Name == c.config.NodeSwitch {
×
788
                        return fmt.Errorf("NodeSwitch subnet %s is unavailable for pod", subnet.Name)
×
789
                }
×
790

791
                if portGroups, err = c.OVNNbClient.ListPortGroups(map[string]string{"subnet": subnet.Name, "node": "", networkPolicyKey: ""}); err != nil {
×
792
                        klog.Errorf("failed to list port groups: %v", err)
×
793
                        return err
×
794
                }
×
795

796
                pgName := getOverlaySubnetsPortGroupName(subnet.Name, pod.Spec.NodeName)
×
797
                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
×
798
                subnetPortGroups := make([]string, 0, len(portGroups))
×
799
                for _, pg := range portGroups {
×
800
                        if pg.Name != pgName {
×
801
                                subnetPortGroups = append(subnetPortGroups, pg.Name)
×
802
                        }
×
803
                }
804

805
                if (!c.config.EnableLb || (subnet.Spec.EnableLb == nil || !*subnet.Spec.EnableLb)) &&
×
806
                        subnet.Spec.Vpc == c.config.ClusterRouter &&
×
807
                        subnet.Spec.U2OInterconnection &&
×
808
                        subnet.Spec.Vlan != "" &&
×
809
                        !subnet.Spec.LogicalGateway {
×
810
                        // remove lsp from other port groups
×
811
                        // we need to do this because the pod, e.g. a sts/vm, can be rescheduled to another node
×
812
                        if err = c.OVNNbClient.RemovePortFromPortGroups(portName, subnetPortGroups...); err != nil {
×
813
                                klog.Errorf("failed to remove port %s from port groups %v: %v", portName, subnetPortGroups, err)
×
814
                                return err
×
815
                        }
×
816
                        // add lsp to the port group
817
                        if err := c.OVNNbClient.PortGroupAddPorts(pgName, portName); err != nil {
×
818
                                klog.Errorf("failed to add port to u2o port group %s: %v", pgName, err)
×
819
                                return err
×
820
                        }
×
821
                }
822

823
                if podIP != "" && (subnet.Spec.Vlan == "" || subnet.Spec.LogicalGateway) && subnet.Spec.Vpc == c.config.ClusterRouter {
×
824
                        node, err := c.nodesLister.Get(pod.Spec.NodeName)
×
825
                        if err != nil {
×
826
                                klog.Errorf("failed to get node %s: %v", pod.Spec.NodeName, err)
×
827
                                return err
×
828
                        }
×
829

830
                        // remove lsp from other port groups
831
                        // we need to do this because the pod, e.g. a sts/vm, can be rescheduled to another node
832
                        if err = c.OVNNbClient.RemovePortFromPortGroups(portName, nodePortGroups...); err != nil {
×
833
                                klog.Errorf("failed to remove port %s from port groups %v: %v", portName, nodePortGroups, err)
×
834
                                return err
×
835
                        }
×
836
                        // add lsp to the port group
837
                        if err = c.OVNNbClient.PortGroupAddPorts(nodePortGroup, portName); err != nil {
×
838
                                klog.Errorf("failed to add port %s to port group %s: %v", portName, nodePortGroup, err)
×
839
                                return err
×
840
                        }
×
841

842
                        if c.config.EnableEipSnat && (pod.Annotations[util.EipAnnotation] != "" || pod.Annotations[util.SnatAnnotation] != "") {
×
843
                                cm, err := c.configMapsLister.ConfigMaps(c.config.ExternalGatewayConfigNS).Get(util.ExternalGatewayConfig)
×
844
                                if err != nil {
×
845
                                        klog.Errorf("failed to get ex-gateway config, %v", err)
×
846
                                        return err
×
847
                                }
×
848
                                nextHop := cm.Data["external-gw-addr"]
×
849
                                if nextHop == "" {
×
850
                                        externalSubnet, err := c.subnetsLister.Get(c.config.ExternalGatewaySwitch)
×
851
                                        if err != nil {
×
852
                                                klog.Errorf("failed to get subnet %s, %v", c.config.ExternalGatewaySwitch, err)
×
853
                                                return err
×
854
                                        }
×
855
                                        nextHop = externalSubnet.Spec.Gateway
×
856
                                        if nextHop == "" {
×
857
                                                klog.Errorf("no available gateway address")
×
858
                                                return errors.New("no available gateway address")
×
859
                                        }
×
860
                                }
861
                                if strings.Contains(nextHop, "/") {
×
862
                                        nextHop = strings.Split(nextHop, "/")[0]
×
863
                                }
×
864

865
                                if err := c.addPolicyRouteToVpc(
×
866
                                        subnet.Spec.Vpc,
×
867
                                        &kubeovnv1.PolicyRoute{
×
868
                                                Priority:  util.NorthGatewayRoutePolicyPriority,
×
869
                                                Match:     "ip4.src == " + podIP,
×
870
                                                Action:    kubeovnv1.PolicyRouteActionReroute,
×
871
                                                NextHopIP: nextHop,
×
872
                                        },
×
873
                                        map[string]string{
×
874
                                                "vendor": util.CniTypeName,
×
875
                                                "subnet": subnet.Name,
×
876
                                        },
×
877
                                ); err != nil {
×
878
                                        klog.Errorf("failed to add policy route, %v", err)
×
879
                                        return err
×
880
                                }
×
881

882
                                // remove lsp from port group to make EIP/SNAT work
883
                                if err = c.OVNNbClient.PortGroupRemovePorts(pgName, portName); err != nil {
×
884
                                        klog.Error(err)
×
885
                                        return err
×
886
                                }
×
887
                        } else {
×
888
                                if subnet.Spec.GatewayType == kubeovnv1.GWDistributedType && pod.Annotations[util.NorthGatewayAnnotation] == "" {
×
889
                                        nodeTunlIPAddr, err := getNodeTunlIP(node)
×
890
                                        if err != nil {
×
891
                                                klog.Error(err)
×
892
                                                return err
×
893
                                        }
×
894

895
                                        var added bool
×
896
                                        for _, nodeAddr := range nodeTunlIPAddr {
×
897
                                                for podAddr := range strings.SplitSeq(podIP, ",") {
×
898
                                                        if util.CheckProtocol(nodeAddr.String()) != util.CheckProtocol(podAddr) {
×
899
                                                                continue
×
900
                                                        }
901

902
                                                        // remove lsp from other port groups
903
                                                        // we need to do this because the pod, e.g. a sts/vm, can be rescheduled to another node
904
                                                        if err = c.OVNNbClient.RemovePortFromPortGroups(portName, subnetPortGroups...); err != nil {
×
905
                                                                klog.Errorf("failed to remove port %s from port groups %v: %v", portName, subnetPortGroups, err)
×
906
                                                                return err
×
907
                                                        }
×
908
                                                        if err := c.OVNNbClient.PortGroupAddPorts(pgName, portName); err != nil {
×
909
                                                                klog.Errorf("failed to add port %s to port group %s: %v", portName, pgName, err)
×
910
                                                                return err
×
911
                                                        }
×
912

913
                                                        added = true
×
914
                                                        break
×
915
                                                }
916
                                                if added {
×
917
                                                        break
×
918
                                                }
919
                                        }
920
                                }
921

922
                                if pod.Annotations[util.NorthGatewayAnnotation] != "" && pod.Annotations[util.IPAddressAnnotation] != "" {
×
923
                                        for podAddr := range strings.SplitSeq(pod.Annotations[util.IPAddressAnnotation], ",") {
×
924
                                                if util.CheckProtocol(podAddr) != util.CheckProtocol(pod.Annotations[util.NorthGatewayAnnotation]) {
×
925
                                                        continue
×
926
                                                }
927
                                                ipSuffix := "ip4"
×
928
                                                if util.CheckProtocol(podAddr) == kubeovnv1.ProtocolIPv6 {
×
929
                                                        ipSuffix = "ip6"
×
930
                                                }
×
931

932
                                                if err := c.addPolicyRouteToVpc(
×
933
                                                        subnet.Spec.Vpc,
×
934
                                                        &kubeovnv1.PolicyRoute{
×
935
                                                                Priority:  util.NorthGatewayRoutePolicyPriority,
×
936
                                                                Match:     fmt.Sprintf("%s.src == %s", ipSuffix, podAddr),
×
937
                                                                Action:    kubeovnv1.PolicyRouteActionReroute,
×
938
                                                                NextHopIP: pod.Annotations[util.NorthGatewayAnnotation],
×
939
                                                        },
×
940
                                                        map[string]string{
×
941
                                                                "vendor": util.CniTypeName,
×
942
                                                                "subnet": subnet.Name,
×
943
                                                        },
×
944
                                                ); err != nil {
×
945
                                                        klog.Errorf("failed to add policy route, %v", err)
×
946
                                                        return err
×
947
                                                }
×
948
                                        }
949
                                } else if c.config.EnableEipSnat {
×
950
                                        if err = c.deleteStaticRouteFromVpc(
×
951
                                                c.config.ClusterRouter,
×
952
                                                subnet.Spec.RouteTable,
×
953
                                                podIP,
×
954
                                                "",
×
955
                                                kubeovnv1.PolicyDst,
×
956
                                        ); err != nil {
×
957
                                                klog.Error(err)
×
958
                                                return err
×
959
                                        }
×
960
                                }
961
                        }
962

963
                        if c.config.EnableEipSnat {
×
964
                                for ipStr := range strings.SplitSeq(podIP, ",") {
×
965
                                        if eip := pod.Annotations[util.EipAnnotation]; eip == "" {
×
966
                                                if err = c.OVNNbClient.DeleteNats(c.config.ClusterRouter, ovnnb.NATTypeDNATAndSNAT, ipStr); err != nil {
×
967
                                                        klog.Errorf("failed to delete nat rules: %v", err)
×
968
                                                }
×
969
                                        } else if util.CheckProtocol(eip) == util.CheckProtocol(ipStr) {
×
970
                                                if err = c.OVNNbClient.UpdateDnatAndSnat(c.config.ClusterRouter, eip, ipStr, fmt.Sprintf("%s.%s", podName, pod.Namespace), pod.Annotations[util.MacAddressAnnotation], c.ExternalGatewayType); err != nil {
×
971
                                                        klog.Errorf("failed to add nat rules, %v", err)
×
972
                                                        return err
×
973
                                                }
×
974
                                        }
975
                                        if eip := pod.Annotations[util.SnatAnnotation]; eip == "" {
×
976
                                                if err = c.OVNNbClient.DeleteNats(c.config.ClusterRouter, ovnnb.NATTypeSNAT, ipStr); err != nil {
×
977
                                                        klog.Errorf("failed to delete nat rules: %v", err)
×
978
                                                }
×
979
                                        } else if util.CheckProtocol(eip) == util.CheckProtocol(ipStr) {
×
980
                                                if err = c.OVNNbClient.UpdateSnat(c.config.ClusterRouter, eip, ipStr); err != nil {
×
981
                                                        klog.Errorf("failed to add nat rules, %v", err)
×
982
                                                        return err
×
983
                                                }
×
984
                                        }
985
                                }
986
                        }
987
                }
988

989
                if pod.Annotations[fmt.Sprintf(util.ActivationStrategyTemplate, podNet.ProviderName)] != "" {
×
990
                        if err := c.OVNNbClient.SetLogicalSwitchPortActivationStrategy(portName, pod.Spec.NodeName); err != nil {
×
991
                                klog.Errorf("failed to set activation strategy for lsp %s: %v", portName, err)
×
992
                                return err
×
993
                        }
×
994
                }
995

996
                patch[fmt.Sprintf(util.RoutedAnnotationTemplate, podNet.ProviderName)] = "true"
×
997
        }
998
        if err := util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(namespace), name, patch); err != nil {
×
999
                if k8serrors.IsNotFound(err) {
×
1000
                        // Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.
×
1001
                        // Then we need to recycle the resource again.
×
1002
                        key := strings.Join([]string{namespace, name}, "/")
×
1003
                        c.deletingPodObjMap.Store(key, pod)
×
1004
                        c.deletePodQueue.AddRateLimited(key)
×
1005
                        return nil
×
1006
                }
×
1007
                klog.Errorf("failed to patch pod %s/%s: %v", namespace, name, err)
×
1008
                return err
×
1009
        }
1010
        return nil
×
1011
}
1012

1013
func (c *Controller) handleDeletePod(key string) (err error) {
×
1014
        pod, ok := c.deletingPodObjMap.Load(key)
×
1015
        if !ok {
×
1016
                return nil
×
1017
        }
×
1018
        now := time.Now()
×
1019
        klog.Infof("handle delete pod %s", key)
×
1020
        podName := c.getNameByPod(pod)
×
1021
        c.podKeyMutex.LockKey(key)
×
1022
        defer func() {
×
1023
                _ = c.podKeyMutex.UnlockKey(key)
×
1024
                if err == nil {
×
1025
                        c.deletingPodObjMap.Delete(key)
×
1026
                }
×
1027
                last := time.Since(now)
×
1028
                klog.Infof("take %d ms to handle delete pod %s", last.Milliseconds(), key)
×
1029
        }()
1030

1031
        p, _ := c.podsLister.Pods(pod.Namespace).Get(pod.Name)
×
1032
        if p != nil && p.UID != pod.UID {
×
1033
                // Pod with same name exists, just return here
×
1034
                return nil
×
1035
        }
×
1036

1037
        if aaps := pod.Annotations[util.AAPsAnnotation]; aaps != "" {
×
1038
                for vipName := range strings.SplitSeq(aaps, ",") {
×
1039
                        if vip, err := c.virtualIpsLister.Get(vipName); err == nil {
×
1040
                                if vip.Spec.Namespace != pod.Namespace {
×
1041
                                        continue
×
1042
                                }
1043
                                klog.Infof("enqueue update virtual parents for %s", vipName)
×
1044
                                c.updateVirtualParentsQueue.Add(vipName)
×
1045
                        }
1046
                }
1047
        }
1048

1049
        podKey := fmt.Sprintf("%s/%s", pod.Namespace, podName)
×
1050

×
1051
        var keepIPCR bool
×
1052
        if ok, stsName, stsUID := isStatefulSetPod(pod); ok {
×
1053
                if !pod.DeletionTimestamp.IsZero() {
×
1054
                        klog.Infof("handle deletion of sts pod %s", podKey)
×
1055
                        toDel := isStatefulSetPodToDel(c.config.KubeClient, pod, stsName, stsUID)
×
1056
                        if !toDel {
×
1057
                                klog.Infof("try keep ip for sts pod %s", podKey)
×
1058
                                keepIPCR = true
×
1059
                        }
×
1060
                }
1061
                if keepIPCR {
×
1062
                        isDelete, err := appendCheckPodToDel(c, pod, stsName, util.StatefulSet)
×
1063
                        if err != nil {
×
1064
                                klog.Error(err)
×
1065
                                return err
×
1066
                        }
×
1067
                        if isDelete {
×
1068
                                klog.Infof("not keep ip for sts pod %s", podKey)
×
1069
                                keepIPCR = false
×
1070
                        }
×
1071
                }
1072
        }
1073
        isVMPod, vmName := isVMPod(pod)
×
1074
        if isVMPod && c.config.EnableKeepVMIP {
×
1075
                ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": podKey})
×
1076
                if err != nil {
×
1077
                        klog.Errorf("failed to list lsps of pod %s: %v", podKey, err)
×
1078
                        return err
×
1079
                }
×
1080
                for _, port := range ports {
×
1081
                        if err := c.OVNNbClient.CleanLogicalSwitchPortMigrateOptions(port.Name); err != nil {
×
1082
                                err = fmt.Errorf("failed to clean migrate options for vm lsp %s, %w", port.Name, err)
×
1083
                                klog.Error(err)
×
1084
                                return err
×
1085
                        }
×
1086
                }
1087
                if pod.DeletionTimestamp != nil {
×
1088
                        klog.Infof("handle deletion of vm pod %s", podKey)
×
1089
                        vmToBeDel := c.isVMToDel(pod, vmName)
×
1090
                        if !vmToBeDel {
×
1091
                                klog.Infof("try keep ip for vm pod %s", podKey)
×
1092
                                keepIPCR = true
×
1093
                        }
×
1094
                }
1095
                if keepIPCR {
×
1096
                        isDelete, err := appendCheckPodToDel(c, pod, vmName, util.VMInstance)
×
1097
                        if err != nil {
×
1098
                                klog.Error(err)
×
1099
                                return err
×
1100
                        }
×
1101
                        if isDelete {
×
1102
                                klog.Infof("not keep ip for vm pod %s", podKey)
×
1103
                                keepIPCR = false
×
1104
                        }
×
1105
                }
1106
        }
1107

1108
        podNets, err := c.getPodKubeovnNets(pod)
×
1109
        if err != nil {
×
1110
                klog.Errorf("failed to get kube-ovn nets of pod %s: %v", podKey, err)
×
1111
        }
×
1112
        ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": podKey})
×
1113
        if err != nil {
×
1114
                klog.Errorf("failed to list lsps of pod %s: %v", podKey, err)
×
1115
                return err
×
1116
        }
×
1117
        if keepIPCR {
×
1118
                // always remove lsp from port groups
×
1119
                for _, port := range ports {
×
1120
                        klog.Infof("remove lsp %s from all port groups", port.Name)
×
1121
                        if err = c.OVNNbClient.RemovePortFromPortGroups(port.Name); err != nil {
×
1122
                                klog.Errorf("failed to remove lsp %s from all port groups: %v", port.Name, err)
×
1123
                                return err
×
1124
                        }
×
1125
                }
1126
        } else {
×
1127
                if len(ports) != 0 {
×
1128
                        addresses := c.ipam.GetPodAddress(podKey)
×
1129
                        for _, address := range addresses {
×
1130
                                if strings.TrimSpace(address.IP) == "" {
×
1131
                                        continue
×
1132
                                }
1133
                                subnet, err := c.subnetsLister.Get(address.Subnet.Name)
×
1134
                                if k8serrors.IsNotFound(err) {
×
1135
                                        continue
×
1136
                                } else if err != nil {
×
1137
                                        klog.Error(err)
×
1138
                                        return err
×
1139
                                }
×
1140
                                vpc, err := c.vpcsLister.Get(subnet.Spec.Vpc)
×
1141
                                if k8serrors.IsNotFound(err) {
×
1142
                                        continue
×
1143
                                } else if err != nil {
×
1144
                                        klog.Error(err)
×
1145
                                        return err
×
1146
                                }
×
1147

1148
                                ipSuffix := "ip4"
×
1149
                                if util.CheckProtocol(address.IP) == kubeovnv1.ProtocolIPv6 {
×
1150
                                        ipSuffix = "ip6"
×
1151
                                }
×
1152
                                if err = c.deletePolicyRouteFromVpc(
×
1153
                                        vpc.Name,
×
1154
                                        util.NorthGatewayRoutePolicyPriority,
×
1155
                                        fmt.Sprintf("%s.src == %s", ipSuffix, address.IP),
×
1156
                                ); err != nil {
×
1157
                                        klog.Errorf("failed to delete static route, %v", err)
×
1158
                                        return err
×
1159
                                }
×
1160

1161
                                if c.config.EnableEipSnat {
×
1162
                                        if pod.Annotations[util.EipAnnotation] != "" {
×
1163
                                                if err = c.OVNNbClient.DeleteNat(c.config.ClusterRouter, ovnnb.NATTypeDNATAndSNAT, pod.Annotations[util.EipAnnotation], address.IP); err != nil {
×
1164
                                                        klog.Errorf("failed to delete nat rules: %v", err)
×
1165
                                                }
×
1166
                                        }
1167
                                        if pod.Annotations[util.SnatAnnotation] != "" {
×
1168
                                                if err = c.OVNNbClient.DeleteNat(c.config.ClusterRouter, ovnnb.NATTypeSNAT, "", address.IP); err != nil {
×
1169
                                                        klog.Errorf("failed to delete nat rules: %v", err)
×
1170
                                                }
×
1171
                                        }
1172
                                }
1173
                        }
1174
                }
1175
                for _, port := range ports {
×
1176
                        // when lsp is deleted, the port of pod is deleted from any port-group automatically.
×
1177
                        klog.Infof("delete logical switch port %s", port.Name)
×
1178
                        if err := c.OVNNbClient.DeleteLogicalSwitchPort(port.Name); err != nil {
×
1179
                                klog.Errorf("failed to delete lsp %s, %v", port.Name, err)
×
1180
                                return err
×
1181
                        }
×
1182
                }
1183
                klog.Infof("try release all ip address for deleting pod %s", podKey)
×
1184
                for _, podNet := range podNets {
×
1185
                        portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
×
1186
                        ipCR, err := c.ipsLister.Get(portName)
×
1187
                        if err != nil {
×
1188
                                if k8serrors.IsNotFound(err) {
×
1189
                                        continue
×
1190
                                }
1191
                                klog.Errorf("failed to get ip %s, %v", portName, err)
×
1192
                                return err
×
1193
                        }
1194
                        if ipCR.Labels[util.IPReservedLabel] != "true" {
×
1195
                                klog.Infof("delete ip CR %s", ipCR.Name)
×
1196
                                if err := c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), ipCR.Name, metav1.DeleteOptions{}); err != nil {
×
1197
                                        if !k8serrors.IsNotFound(err) {
×
1198
                                                klog.Errorf("failed to delete ip %s, %v", ipCR.Name, err)
×
1199
                                                return err
×
1200
                                        }
×
1201
                                }
1202
                                // release ipam address after delete ip CR
1203
                                c.ipam.ReleaseAddressByNic(podKey, portName, podNet.Subnet.Name)
×
1204
                        }
1205
                }
1206
                if pod.Annotations[util.VipAnnotation] != "" {
×
1207
                        if err = c.releaseVip(pod.Annotations[util.VipAnnotation]); err != nil {
×
1208
                                klog.Errorf("failed to clean label from vip %s, %v", pod.Annotations[util.VipAnnotation], err)
×
1209
                                return err
×
1210
                        }
×
1211
                }
1212
        }
1213
        for _, podNet := range podNets {
×
1214
                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)
×
1215
                securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]
×
1216
                if securityGroupAnnotation != "" {
×
1217
                        securityGroups := strings.ReplaceAll(securityGroupAnnotation, " ", "")
×
1218
                        for sgName := range strings.SplitSeq(securityGroups, ",") {
×
1219
                                if sgName != "" {
×
1220
                                        c.syncSgPortsQueue.Add(sgName)
×
1221
                                }
×
1222
                        }
1223
                }
1224
        }
1225
        return nil
×
1226
}
1227

1228
func (c *Controller) handleUpdatePodSecurity(key string) error {
×
1229
        now := time.Now()
×
1230
        klog.Infof("handle add/update pod security group %s", key)
×
1231

×
1232
        namespace, name, err := cache.SplitMetaNamespaceKey(key)
×
1233
        if err != nil {
×
1234
                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
×
1235
                return nil
×
1236
        }
×
1237

1238
        c.podKeyMutex.LockKey(key)
×
1239
        defer func() {
×
1240
                _ = c.podKeyMutex.UnlockKey(key)
×
1241
                last := time.Since(now)
×
1242
                klog.Infof("take %d ms to handle sg for pod %s", last.Milliseconds(), key)
×
1243
        }()
×
1244

1245
        pod, err := c.podsLister.Pods(namespace).Get(name)
×
1246
        if err != nil {
×
1247
                if k8serrors.IsNotFound(err) {
×
1248
                        return nil
×
1249
                }
×
1250
                klog.Error(err)
×
1251
                return err
×
1252
        }
1253
        podName := c.getNameByPod(pod)
×
1254

×
1255
        podNets, err := c.getPodKubeovnNets(pod)
×
1256
        if err != nil {
×
1257
                klog.Errorf("failed to pod nets %v", err)
×
1258
                return err
×
1259
        }
×
1260

1261
        vipsMap := c.getVirtualIPs(pod, podNets)
×
1262

×
1263
        // associated with security group
×
1264
        for _, podNet := range podNets {
×
1265
                portSecurity := false
×
1266
                if pod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)] == "true" {
×
1267
                        portSecurity = true
×
1268
                }
×
1269

1270
                mac := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]
×
1271
                ipStr := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]
×
1272
                vips := vipsMap[fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)]
×
1273
                portName := ovs.PodNameToPortName(podName, namespace, podNet.ProviderName)
×
1274
                if err = c.OVNNbClient.SetLogicalSwitchPortSecurity(portSecurity, portName, mac, ipStr, vips); err != nil {
×
1275
                        klog.Errorf("failed to set security for logical switch port %s: %v", portName, err)
×
1276
                        return err
×
1277
                }
×
1278

1279
                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)
×
1280
                securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]
×
1281
                var securityGroups string
×
1282
                if securityGroupAnnotation != "" {
×
1283
                        securityGroups = strings.ReplaceAll(securityGroupAnnotation, " ", "")
×
1284
                        for sgName := range strings.SplitSeq(securityGroups, ",") {
×
1285
                                if sgName != "" {
×
1286
                                        c.syncSgPortsQueue.Add(sgName)
×
1287
                                }
×
1288
                        }
1289
                }
1290
                if err = c.reconcilePortSg(portName, securityGroups); err != nil {
×
1291
                        klog.Errorf("reconcilePortSg failed. %v", err)
×
1292
                        return err
×
1293
                }
×
1294
        }
1295
        return nil
×
1296
}
1297

1298
func (c *Controller) syncKubeOvnNet(pod *v1.Pod, podNets []*kubeovnNet) (*v1.Pod, error) {
×
1299
        podName := c.getNameByPod(pod)
×
1300
        key := fmt.Sprintf("%s/%s", pod.Namespace, podName)
×
1301
        targetPortNameList := strset.NewWithSize(len(podNets))
×
1302
        portsNeedToDel := []string{}
×
1303
        annotationsNeedToDel := []string{}
×
1304
        annotationsNeedToAdd := make(map[string]string)
×
1305
        subnetUsedByPort := make(map[string]string)
×
1306

×
1307
        for _, podNet := range podNets {
×
1308
                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
×
1309
                targetPortNameList.Add(portName)
×
1310
                if podNet.IPRequest != "" {
×
1311
                        klog.Infof("pod %s/%s use custom IP %s for provider %s", pod.Namespace, pod.Name, podNet.IPRequest, podNet.ProviderName)
×
1312
                        annotationsNeedToAdd[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] = podNet.IPRequest
×
1313
                }
×
1314

1315
                if podNet.MacRequest != "" {
×
1316
                        klog.Infof("pod %s/%s use custom MAC %s for provider %s", pod.Namespace, pod.Name, podNet.MacRequest, podNet.ProviderName)
×
1317
                        annotationsNeedToAdd[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = podNet.MacRequest
×
1318
                }
×
1319
        }
1320

1321
        ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": key})
×
1322
        if err != nil {
×
1323
                klog.Errorf("failed to list lsps of pod '%s', %v", pod.Name, err)
×
1324
                return nil, err
×
1325
        }
×
1326

1327
        for _, port := range ports {
×
1328
                if !targetPortNameList.Has(port.Name) {
×
1329
                        portsNeedToDel = append(portsNeedToDel, port.Name)
×
1330
                        subnetUsedByPort[port.Name] = port.ExternalIDs["ls"]
×
1331
                        portNameSlice := strings.Split(port.Name, ".")
×
1332
                        providerName := strings.Join(portNameSlice[2:], ".")
×
1333
                        if providerName == util.OvnProvider {
×
1334
                                continue
×
1335
                        }
1336
                        annotationsNeedToDel = append(annotationsNeedToDel, providerName)
×
1337
                }
1338
        }
1339

1340
        if len(portsNeedToDel) == 0 && len(annotationsNeedToAdd) == 0 {
×
1341
                return pod, nil
×
1342
        }
×
1343

1344
        for _, portNeedDel := range portsNeedToDel {
×
1345
                klog.Infof("release port %s for pod %s", portNeedDel, podName)
×
1346
                if subnet, ok := c.ipam.Subnets[subnetUsedByPort[portNeedDel]]; ok {
×
1347
                        subnet.ReleaseAddressWithNicName(podName, portNeedDel)
×
1348
                }
×
1349
                if err := c.OVNNbClient.DeleteLogicalSwitchPort(portNeedDel); err != nil {
×
1350
                        klog.Errorf("failed to delete lsp %s, %v", portNeedDel, err)
×
1351
                        return nil, err
×
1352
                }
×
1353
                if err := c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), portNeedDel, metav1.DeleteOptions{}); err != nil {
×
1354
                        if !k8serrors.IsNotFound(err) {
×
1355
                                klog.Errorf("failed to delete ip %s, %v", portNeedDel, err)
×
1356
                                return nil, err
×
1357
                        }
×
1358
                }
1359
        }
1360

1361
        patch := util.KVPatch{}
×
1362
        for _, providerName := range annotationsNeedToDel {
×
1363
                for key := range pod.Annotations {
×
1364
                        if strings.HasPrefix(key, providerName) {
×
1365
                                patch[key] = nil
×
1366
                        }
×
1367
                }
1368
        }
1369

1370
        for key, value := range annotationsNeedToAdd {
×
1371
                patch[key] = value
×
1372
        }
×
1373

1374
        if len(patch) == 0 {
×
1375
                return pod, nil
×
1376
        }
×
1377

1378
        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(pod.Namespace), pod.Name, patch); err != nil {
×
1379
                if k8serrors.IsNotFound(err) {
×
1380
                        return nil, nil
×
1381
                }
×
1382
                klog.Errorf("failed to clean annotations for pod %s/%s: %v", pod.Namespace, pod.Name, err)
×
1383
                return nil, err
×
1384
        }
1385

1386
        if pod, err = c.config.KubeClient.CoreV1().Pods(pod.Namespace).Get(context.TODO(), pod.Name, metav1.GetOptions{}); err != nil {
×
1387
                if k8serrors.IsNotFound(err) {
×
1388
                        return nil, nil
×
1389
                }
×
1390
                klog.Errorf("failed to get pod %s/%s: %v", pod.Namespace, pod.Name, err)
×
1391
                return nil, err
×
1392
        }
1393

1394
        return pod, nil
×
1395
}
1396

1397
func isStatefulSetPod(pod *v1.Pod) (bool, string, types.UID) {
×
1398
        for _, owner := range pod.OwnerReferences {
×
1399
                if owner.Kind == util.StatefulSet && strings.HasPrefix(owner.APIVersion, "apps/") {
×
1400
                        if strings.HasPrefix(pod.Name, owner.Name) {
×
1401
                                return true, owner.Name, owner.UID
×
1402
                        }
×
1403
                }
1404
        }
1405
        return false, "", ""
×
1406
}
1407

1408
func isStatefulSetPodToDel(c kubernetes.Interface, pod *v1.Pod, statefulSetName string, statefulSetUID types.UID) bool {
×
1409
        // only delete statefulset pod lsp when statefulset deleted or down scaled
×
1410
        sts, err := c.AppsV1().StatefulSets(pod.Namespace).Get(context.Background(), statefulSetName, metav1.GetOptions{})
×
1411
        if err != nil {
×
1412
                // statefulset is deleted
×
1413
                if k8serrors.IsNotFound(err) {
×
1414
                        klog.Infof("statefulset %s/%s has been deleted", pod.Namespace, statefulSetName)
×
1415
                        return true
×
1416
                }
×
1417
                klog.Errorf("failed to get statefulset %s/%s: %v", pod.Namespace, statefulSetName, err)
×
1418
                return false
×
1419
        }
1420

1421
        // statefulset is being deleted, or it's a newly created one
1422
        if !sts.DeletionTimestamp.IsZero() {
×
1423
                klog.Infof("statefulset %s/%s is being deleted", pod.Namespace, statefulSetName)
×
1424
                return true
×
1425
        }
×
1426
        if sts.UID != statefulSetUID {
×
1427
                klog.Infof("statefulset %s/%s is a newly created one", pod.Namespace, statefulSetName)
×
1428
                return true
×
1429
        }
×
1430

1431
        // down scale statefulset
1432
        tempStrs := strings.Split(pod.Name, "-")
×
1433
        numStr := tempStrs[len(tempStrs)-1]
×
1434
        index, err := strconv.ParseInt(numStr, 10, 0)
×
1435
        if err != nil {
×
1436
                klog.Errorf("failed to parse %s to int", numStr)
×
1437
                return false
×
1438
        }
×
1439
        // down scaled
1440
        var startOrdinal int64
×
1441
        if sts.Spec.Ordinals != nil {
×
1442
                startOrdinal = int64(sts.Spec.Ordinals.Start)
×
1443
        }
×
1444
        if index >= startOrdinal+int64(*sts.Spec.Replicas) {
×
1445
                klog.Infof("statefulset %s/%s is down scaled", pod.Namespace, statefulSetName)
×
1446
                return true
×
1447
        }
×
1448
        return false
×
1449
}
1450

1451
// only gc statefulset pod lsp when:
1452
// 1. the statefulset has been deleted or is being deleted
1453
// 2. the statefulset has been deleted and recreated
1454
// 3. the statefulset is down scaled and the pod is not alive
1455
func isStatefulSetPodToGC(c kubernetes.Interface, pod *v1.Pod, statefulSetName string, statefulSetUID types.UID) bool {
×
1456
        sts, err := c.AppsV1().StatefulSets(pod.Namespace).Get(context.Background(), statefulSetName, metav1.GetOptions{})
×
1457
        if err != nil {
×
1458
                // the statefulset has been deleted
×
1459
                if k8serrors.IsNotFound(err) {
×
1460
                        klog.Infof("statefulset %s/%s has been deleted", pod.Namespace, statefulSetName)
×
1461
                        return true
×
1462
                }
×
1463
                klog.Errorf("failed to get statefulset %s/%s: %v", pod.Namespace, statefulSetName, err)
×
1464
                return false
×
1465
        }
1466

1467
        // statefulset is being deleted
1468
        if !sts.DeletionTimestamp.IsZero() {
×
1469
                klog.Infof("statefulset %s/%s is being deleted", pod.Namespace, statefulSetName)
×
1470
                return true
×
1471
        }
×
1472
        // the statefulset has been deleted and recreated
1473
        if sts.UID != statefulSetUID {
×
1474
                klog.Infof("statefulset %s/%s is a newly created one", pod.Namespace, statefulSetName)
×
1475
                return true
×
1476
        }
×
1477

1478
        // the statefulset is down scaled and the pod is not alive
1479

1480
        tempStrs := strings.Split(pod.Name, "-")
×
1481
        numStr := tempStrs[len(tempStrs)-1]
×
1482
        index, err := strconv.ParseInt(numStr, 10, 0)
×
1483
        if err != nil {
×
1484
                klog.Errorf("failed to parse %s to int", numStr)
×
1485
                return false
×
1486
        }
×
1487
        // down scaled
1488
        var startOrdinal int64
×
1489
        if sts.Spec.Ordinals != nil {
×
1490
                startOrdinal = int64(sts.Spec.Ordinals.Start)
×
1491
        }
×
1492
        if index >= startOrdinal+int64(*sts.Spec.Replicas) {
×
1493
                klog.Infof("statefulset %s/%s is down scaled", pod.Namespace, statefulSetName)
×
1494
                if !isPodAlive(pod) {
×
1495
                        // we must check whether the pod is alive because we have to consider the following case:
×
1496
                        // 1. the statefulset is down scaled to zero
×
1497
                        // 2. the lsp gc is triggered
×
1498
                        // 3. gc interval, e.g. 90s, is passed and the second gc is triggered
×
1499
                        // 4. the sts is up scaled to the original replicas
×
1500
                        // 5. the pod is still running and it will not be recreated
×
1501
                        return true
×
1502
                }
×
1503
        }
1504

1505
        return false
×
1506
}
1507

1508
func getNodeTunlIP(node *v1.Node) ([]net.IP, error) {
×
1509
        var nodeTunlIPAddr []net.IP
×
1510
        nodeTunlIP := node.Annotations[util.IPAddressAnnotation]
×
1511
        if nodeTunlIP == "" {
×
1512
                return nil, errors.New("node has no tunnel ip annotation")
×
1513
        }
×
1514

1515
        for ip := range strings.SplitSeq(nodeTunlIP, ",") {
×
1516
                nodeTunlIPAddr = append(nodeTunlIPAddr, net.ParseIP(ip))
×
1517
        }
×
1518
        return nodeTunlIPAddr, nil
×
1519
}
1520

1521
func getNextHopByTunnelIP(gw []net.IP) string {
×
1522
        // validation check by caller
×
1523
        nextHop := gw[0].String()
×
1524
        if len(gw) == 2 {
×
1525
                nextHop = gw[0].String() + "," + gw[1].String()
×
1526
        }
×
1527
        return nextHop
×
1528
}
1529

1530
func needAllocateSubnets(pod *v1.Pod, nets []*kubeovnNet) []*kubeovnNet {
×
1531
        // check if allocate from subnet is need.
×
1532
        // allocate subnet when change subnet to hotplug nic
×
1533
        // allocate subnet when migrate vm
×
1534
        if !isPodAlive(pod) {
×
1535
                return nil
×
1536
        }
×
1537

1538
        if pod.Annotations == nil {
×
1539
                return nets
×
1540
        }
×
1541

1542
        migrate := false
×
1543
        if job, ok := pod.Annotations[util.MigrationJobAnnotation]; ok {
×
1544
                klog.Infof("pod %s/%s is in the migration job %s", pod.Namespace, pod.Name, job)
×
1545
                migrate = true
×
1546
        }
×
1547

1548
        result := make([]*kubeovnNet, 0, len(nets))
×
1549
        for _, n := range nets {
×
1550
                if migrate || pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, n.ProviderName)] != "true" {
×
1551
                        result = append(result, n)
×
1552
                }
×
1553
        }
1554
        return result
×
1555
}
1556

1557
func needRestartNatGatewayPod(pod *v1.Pod) bool {
×
1558
        for _, psc := range pod.Status.ContainerStatuses {
×
1559
                if psc.Name != "vpc-nat-gw" {
×
1560
                        continue
×
1561
                }
1562
                if psc.RestartCount > 0 {
×
1563
                        return true
×
1564
                }
×
1565
        }
1566
        return false
×
1567
}
1568

1569
func (c *Controller) podNeedSync(pod *v1.Pod) (bool, error) {
×
1570
        // 1. check annotations
×
1571
        if pod.Annotations == nil {
×
1572
                return true, nil
×
1573
        }
×
1574
        // 2. check annotation ovn subnet
1575
        if pod.Annotations[util.RoutedAnnotation] != "true" {
×
1576
                return true, nil
×
1577
        }
×
1578
        // 3. check multus subnet
1579
        attachmentNets, err := c.getPodAttachmentNet(pod)
×
1580
        if err != nil {
×
1581
                klog.Error(err)
×
1582
                return false, err
×
1583
        }
×
1584

1585
        podName := c.getNameByPod(pod)
×
1586
        for _, n := range attachmentNets {
×
1587
                if pod.Annotations[fmt.Sprintf(util.RoutedAnnotationTemplate, n.ProviderName)] != "true" {
×
1588
                        return true, nil
×
1589
                }
×
1590
                ipName := ovs.PodNameToPortName(podName, pod.Namespace, n.ProviderName)
×
1591
                if _, err = c.ipsLister.Get(ipName); err != nil {
×
1592
                        if !k8serrors.IsNotFound(err) {
×
1593
                                err = fmt.Errorf("failed to get ip %s: %w", ipName, err)
×
1594
                                klog.Error(err)
×
1595
                                return false, err
×
1596
                        }
×
1597
                        klog.Infof("ip %s not found", ipName)
×
1598
                        // need to sync to create ip
×
1599
                        return true, nil
×
1600
                }
1601
        }
1602
        return false, nil
×
1603
}
1604

1605
func needRouteSubnets(pod *v1.Pod, nets []*kubeovnNet) []*kubeovnNet {
×
1606
        if !isPodAlive(pod) {
×
1607
                return nil
×
1608
        }
×
1609

1610
        if pod.Annotations == nil {
×
1611
                return nets
×
1612
        }
×
1613

1614
        result := make([]*kubeovnNet, 0, len(nets))
×
1615
        for _, n := range nets {
×
1616
                if !isOvnSubnet(n.Subnet) {
×
1617
                        continue
×
1618
                }
1619

1620
                if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, n.ProviderName)] == "true" && pod.Spec.NodeName != "" {
×
1621
                        if pod.Annotations[fmt.Sprintf(util.RoutedAnnotationTemplate, n.ProviderName)] != "true" {
×
1622
                                result = append(result, n)
×
1623
                        }
×
1624
                }
1625
        }
1626
        return result
×
1627
}
1628

1629
func (c *Controller) getPodDefaultSubnet(pod *v1.Pod) (*kubeovnv1.Subnet, error) {
1✔
1630
        // ignore to clean its ip crd in existing subnets
1✔
1631
        ignoreSubnetNotExist := !pod.DeletionTimestamp.IsZero()
1✔
1632

1✔
1633
        // check pod annotations
1✔
1634
        if lsName := pod.Annotations[util.LogicalSwitchAnnotation]; lsName != "" {
2✔
1635
                // annotations only has one default subnet
1✔
1636
                subnet, err := c.subnetsLister.Get(lsName)
1✔
1637
                if err != nil {
1✔
1638
                        klog.Errorf("failed to get subnet %s: %v", lsName, err)
×
1639
                        if k8serrors.IsNotFound(err) {
×
1640
                                if ignoreSubnetNotExist {
×
1641
                                        klog.Errorf("deletting pod %s/%s default subnet %s already not exist, gc will clean its ip cr", pod.Namespace, pod.Name, lsName)
×
1642
                                        return nil, nil
×
1643
                                }
×
1644
                        }
1645
                        return nil, err
×
1646
                }
1647
                return subnet, nil
1✔
1648
        }
1649

1650
        ns, err := c.namespacesLister.Get(pod.Namespace)
1✔
1651
        if err != nil {
1✔
1652
                klog.Errorf("failed to get namespace %s: %v", pod.Namespace, err)
×
1653
                return nil, err
×
1654
        }
×
1655
        if len(ns.Annotations) == 0 {
1✔
1656
                err = fmt.Errorf("namespace %s network annotations is empty", ns.Name)
×
1657
                klog.Error(err)
×
1658
                return nil, err
×
1659
        }
×
1660

1661
        subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]
1✔
1662
        for subnetName := range strings.SplitSeq(subnetNames, ",") {
2✔
1663
                if subnetName == "" {
1✔
1664
                        err = fmt.Errorf("namespace %s default logical switch is not found", ns.Name)
×
1665
                        klog.Error(err)
×
1666
                        return nil, err
×
1667
                }
×
1668
                subnet, err := c.subnetsLister.Get(subnetName)
1✔
1669
                if err != nil {
1✔
1670
                        klog.Errorf("failed to get subnet %s: %v", subnetName, err)
×
1671
                        if k8serrors.IsNotFound(err) {
×
1672
                                if ignoreSubnetNotExist {
×
1673
                                        klog.Errorf("deletting pod %s/%s namespace subnet %s already not exist, gc will clean its ip cr", pod.Namespace, pod.Name, subnetName)
×
1674
                                        // ip name is unique, it is ok if any subnet release it
×
1675
                                        // gc will handle their ip cr, if all subnets are not exist
×
1676
                                        continue
×
1677
                                }
1678
                        }
1679
                        return nil, err
×
1680
                }
1681

1682
                switch subnet.Spec.Protocol {
1✔
1683
                case kubeovnv1.ProtocolDual:
×
NEW
1684
                        if subnet.Status.V6AvailableIPs.EqualInt64(0) && !c.podCanUseExcludeIPs(pod, subnet) {
×
1685
                                klog.Infof("there's no available ipv6 address in subnet %s, try next one", subnet.Name)
×
1686
                                continue
×
1687
                        }
1688
                        fallthrough
×
1689
                case kubeovnv1.ProtocolIPv4:
×
NEW
1690
                        if subnet.Status.V4AvailableIPs.EqualInt64(0) && !c.podCanUseExcludeIPs(pod, subnet) {
×
1691
                                klog.Infof("there's no available ipv4 address in subnet %s, try next one", subnet.Name)
×
1692
                                continue
×
1693
                        }
1694
                case kubeovnv1.ProtocolIPv6:
×
NEW
1695
                        if subnet.Status.V6AvailableIPs.EqualInt64(0) && !c.podCanUseExcludeIPs(pod, subnet) {
×
1696
                                klog.Infof("there's no available ipv6 address in subnet %s, try next one", subnet.Name)
×
1697
                                continue
×
1698
                        }
1699
                }
1700
                return subnet, nil
1✔
1701
        }
1702
        return nil, ipam.ErrNoAvailable
×
1703
}
1704

1705
func (c *Controller) podCanUseExcludeIPs(pod *v1.Pod, subnet *kubeovnv1.Subnet) bool {
×
1706
        if ipAddr := pod.Annotations[util.IPAddressAnnotation]; ipAddr != "" {
×
1707
                return c.checkIPsInExcludeList(ipAddr, subnet.Spec.ExcludeIps, subnet.Spec.CIDRBlock)
×
1708
        }
×
1709
        if ipPool := pod.Annotations[util.IPPoolAnnotation]; ipPool != "" {
×
1710
                return c.checkIPsInExcludeList(ipPool, subnet.Spec.ExcludeIps, subnet.Spec.CIDRBlock)
×
1711
        }
×
1712

1713
        return false
×
1714
}
1715

1716
func (c *Controller) checkIPsInExcludeList(ips string, excludeIPs []string, cidr string) bool {
×
1717
        expandedExcludeIPs := util.ExpandExcludeIPs(excludeIPs, cidr)
×
1718

×
1719
        for ipAddr := range strings.SplitSeq(strings.TrimSpace(ips), ",") {
×
1720
                ipAddr = strings.TrimSpace(ipAddr)
×
1721
                if ipAddr == "" {
×
1722
                        continue
×
1723
                }
1724

1725
                for _, excludeIP := range expandedExcludeIPs {
×
1726
                        if util.ContainsIPs(excludeIP, ipAddr) {
×
1727
                                klog.V(3).Infof("IP %s is found in exclude IP %s, allowing allocation", ipAddr, excludeIP)
×
1728
                                return true
×
1729
                        }
×
1730
                }
1731
        }
1732
        return false
×
1733
}
1734

1735
type providerType int
1736

1737
const (
1738
        providerTypeIPAM providerType = iota
1739
        providerTypeOriginal
1740
)
1741

1742
type kubeovnNet struct {
1743
        Type               providerType
1744
        ProviderName       string
1745
        Subnet             *kubeovnv1.Subnet
1746
        IsDefault          bool
1747
        AllowLiveMigration bool
1748
        IPRequest          string
1749
        MacRequest         string
1750
}
1751

1752
func (c *Controller) getPodAttachmentNet(pod *v1.Pod) ([]*kubeovnNet, error) {
1✔
1753
        var multusNets []*nadv1.NetworkSelectionElement
1✔
1754
        defaultAttachNetworks := pod.Annotations[util.DefaultNetworkAnnotation]
1✔
1755
        if defaultAttachNetworks != "" {
1✔
1756
                attachments, err := nadutils.ParseNetworkAnnotation(defaultAttachNetworks, pod.Namespace)
×
1757
                if err != nil {
×
1758
                        klog.Errorf("failed to parse default attach net for pod '%s', %v", pod.Name, err)
×
1759
                        return nil, err
×
1760
                }
×
1761
                multusNets = attachments
×
1762
        }
1763

1764
        attachNetworks := pod.Annotations[nadv1.NetworkAttachmentAnnot]
1✔
1765
        if attachNetworks != "" {
2✔
1766
                attachments, err := nadutils.ParseNetworkAnnotation(attachNetworks, pod.Namespace)
1✔
1767
                if err != nil {
1✔
1768
                        klog.Errorf("failed to parse attach net for pod '%s', %v", pod.Name, err)
×
1769
                        return nil, err
×
1770
                }
×
1771
                multusNets = append(multusNets, attachments...)
1✔
1772
        }
1773
        subnets, err := c.subnetsLister.List(labels.Everything())
1✔
1774
        if err != nil {
1✔
1775
                klog.Errorf("failed to list subnets: %v", err)
×
1776
                return nil, err
×
1777
        }
×
1778

1779
        // ignore to return all existing subnets to clean its ip crd
1780
        ignoreSubnetNotExist := !pod.DeletionTimestamp.IsZero()
1✔
1781

1✔
1782
        result := make([]*kubeovnNet, 0, len(multusNets))
1✔
1783
        for _, attach := range multusNets {
2✔
1784
                network, err := c.netAttachLister.NetworkAttachmentDefinitions(attach.Namespace).Get(attach.Name)
1✔
1785
                if err != nil {
1✔
1786
                        klog.Errorf("failed to get net-attach-def %s, %v", attach.Name, err)
×
1787
                        return nil, err
×
1788
                }
×
1789

1790
                if network.Spec.Config == "" {
1✔
1791
                        continue
×
1792
                }
1793

1794
                netCfg, err := loadNetConf([]byte(network.Spec.Config))
1✔
1795
                if err != nil {
1✔
1796
                        klog.Errorf("failed to load config of net-attach-def %s, %v", attach.Name, err)
×
1797
                        return nil, err
×
1798
                }
×
1799

1800
                // allocate kubeovn network
1801
                var providerName string
1✔
1802
                if util.IsOvnNetwork(netCfg) {
2✔
1803
                        allowLiveMigration := false
1✔
1804
                        isDefault := util.IsDefaultNet(pod.Annotations[util.DefaultNetworkAnnotation], attach)
1✔
1805

1✔
1806
                        providerName = fmt.Sprintf("%s.%s.%s", attach.Name, attach.Namespace, util.OvnProvider)
1✔
1807
                        if pod.Annotations[util.MigrationJobAnnotation] != "" {
1✔
1808
                                allowLiveMigration = true
×
1809
                        }
×
1810

1811
                        subnetName := pod.Annotations[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, providerName)]
1✔
1812
                        if subnetName == "" {
1✔
1813
                                for _, subnet := range subnets {
×
1814
                                        if subnet.Spec.Provider == providerName {
×
1815
                                                subnetName = subnet.Name
×
1816
                                                break
×
1817
                                        }
1818
                                }
1819
                        }
1820
                        var subnet *kubeovnv1.Subnet
1✔
1821
                        if subnetName == "" {
1✔
1822
                                // attachment network not specify subnet, use pod default subnet or namespace subnet
×
1823
                                subnet, err = c.getPodDefaultSubnet(pod)
×
1824
                                if err != nil {
×
1825
                                        klog.Errorf("failed to pod default subnet, %v", err)
×
1826
                                        if k8serrors.IsNotFound(err) {
×
1827
                                                if ignoreSubnetNotExist {
×
1828
                                                        klog.Errorf("deletting pod %s/%s attach subnet %s already not exist, gc will clean its ip cr", pod.Namespace, pod.Name, subnetName)
×
1829
                                                        continue
×
1830
                                                }
1831
                                        }
1832
                                        return nil, err
×
1833
                                }
1834
                                // default subnet may change after pod restart
1835
                                klog.Infof("pod %s/%s attachment network %s use default subnet %s", pod.Namespace, pod.Name, attach.Name, subnet.Name)
×
1836
                        } else {
1✔
1837
                                subnet, err = c.subnetsLister.Get(subnetName)
1✔
1838
                                if err != nil {
1✔
1839
                                        klog.Errorf("failed to get subnet %s, %v", subnetName, err)
×
1840
                                        if k8serrors.IsNotFound(err) {
×
1841
                                                if ignoreSubnetNotExist {
×
1842
                                                        klog.Errorf("deletting pod %s/%s attach subnet %s already not exist, gc will clean its ip cr", pod.Namespace, pod.Name, subnetName)
×
1843
                                                        // just continue to next attach subnet
×
1844
                                                        // ip name is unique, so it is ok if the other subnet release it
×
1845
                                                        continue
×
1846
                                                }
1847
                                        }
1848
                                        return nil, err
×
1849
                                }
1850
                        }
1851

1852
                        ret := &kubeovnNet{
1✔
1853
                                Type:               providerTypeOriginal,
1✔
1854
                                ProviderName:       providerName,
1✔
1855
                                Subnet:             subnet,
1✔
1856
                                IsDefault:          isDefault,
1✔
1857
                                AllowLiveMigration: allowLiveMigration,
1✔
1858
                                MacRequest:         attach.MacRequest,
1✔
1859
                                IPRequest:          strings.Join(attach.IPRequest, ","),
1✔
1860
                        }
1✔
1861
                        result = append(result, ret)
1✔
1862
                } else {
×
1863
                        providerName = fmt.Sprintf("%s.%s", attach.Name, attach.Namespace)
×
1864
                        for _, subnet := range subnets {
×
1865
                                if subnet.Spec.Provider == providerName {
×
1866
                                        result = append(result, &kubeovnNet{
×
1867
                                                Type:         providerTypeIPAM,
×
1868
                                                ProviderName: providerName,
×
1869
                                                Subnet:       subnet,
×
1870
                                                MacRequest:   attach.MacRequest,
×
1871
                                                IPRequest:    strings.Join(attach.IPRequest, ","),
×
1872
                                        })
×
1873
                                        break
×
1874
                                }
1875
                        }
1876
                }
1877
        }
1878
        return result, nil
1✔
1879
}
1880

1881
func (c *Controller) validatePodIP(podName, subnetName, ipv4, ipv6 string) (bool, bool, error) {
×
1882
        subnet, err := c.subnetsLister.Get(subnetName)
×
1883
        if err != nil {
×
1884
                klog.Errorf("failed to get subnet %s: %v", subnetName, err)
×
1885
                return false, false, err
×
1886
        }
×
1887

1888
        if subnet.Spec.Vlan == "" && subnet.Spec.Vpc == c.config.ClusterRouter {
×
1889
                nodes, err := c.nodesLister.List(labels.Everything())
×
1890
                if err != nil {
×
1891
                        klog.Errorf("failed to list nodes: %v", err)
×
1892
                        return false, false, err
×
1893
                }
×
1894

1895
                for _, node := range nodes {
×
1896
                        nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
×
1897
                        if ipv4 != "" && ipv4 == nodeIPv4 {
×
1898
                                klog.Errorf("IP address (%s) assigned to pod %s is the same with internal IP address of node %s, reallocating...", ipv4, podName, node.Name)
×
1899
                                return false, true, nil
×
1900
                        }
×
1901
                        if ipv6 != "" && ipv6 == nodeIPv6 {
×
1902
                                klog.Errorf("IP address (%s) assigned to pod %s is the same with internal IP address of node %s, reallocating...", ipv6, podName, node.Name)
×
1903
                                return true, false, nil
×
1904
                        }
×
1905
                }
1906
        }
1907

1908
        return true, true, nil
×
1909
}
1910

1911
func (c *Controller) acquireAddress(pod *v1.Pod, podNet *kubeovnNet) (string, string, string, *kubeovnv1.Subnet, error) {
×
1912
        podName := c.getNameByPod(pod)
×
1913
        key := fmt.Sprintf("%s/%s", pod.Namespace, podName)
×
1914
        portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
×
1915

×
1916
        var checkVMPod bool
×
1917
        isStsPod, _, _ := isStatefulSetPod(pod)
×
1918
        // if pod has static vip
×
1919
        vipName := pod.Annotations[util.VipAnnotation]
×
1920
        if vipName != "" {
×
1921
                vip, err := c.virtualIpsLister.Get(vipName)
×
1922
                if err != nil {
×
1923
                        klog.Errorf("failed to get static vip '%s', %v", vipName, err)
×
1924
                        return "", "", "", podNet.Subnet, err
×
1925
                }
×
1926
                if c.config.EnableKeepVMIP {
×
1927
                        checkVMPod, _ = isVMPod(pod)
×
1928
                }
×
1929
                if err = c.podReuseVip(vipName, portName, isStsPod || checkVMPod); err != nil {
×
1930
                        return "", "", "", podNet.Subnet, err
×
1931
                }
×
1932
                return vip.Status.V4ip, vip.Status.V6ip, vip.Status.Mac, podNet.Subnet, nil
×
1933
        }
1934

1935
        var macPointer *string
×
1936
        if isOvnSubnet(podNet.Subnet) {
×
1937
                annoMAC := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]
×
1938
                if annoMAC != "" {
×
1939
                        if _, err := net.ParseMAC(annoMAC); err != nil {
×
1940
                                return "", "", "", podNet.Subnet, err
×
1941
                        }
×
1942
                        macPointer = &annoMAC
×
1943
                }
1944
        } else {
×
1945
                macPointer = ptr.To("")
×
1946
        }
×
1947

1948
        var err error
×
1949
        var nsNets []*kubeovnNet
×
1950
        ippoolStr := pod.Annotations[fmt.Sprintf(util.IPPoolAnnotationTemplate, podNet.ProviderName)]
×
1951
        subnetStr := pod.Annotations[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName)]
×
1952
        if ippoolStr == "" && podNet.IsDefault {
×
1953
                // no ippool specified by pod annotation, use namespace ippools or ippools in the subnet specified by pod annotation
×
1954
                ns, err := c.namespacesLister.Get(pod.Namespace)
×
1955
                if err != nil {
×
1956
                        klog.Errorf("failed to get namespace %s: %v", pod.Namespace, err)
×
1957
                        return "", "", "", podNet.Subnet, err
×
1958
                }
×
1959
                if nsNets, err = c.getNsAvailableSubnets(pod, podNet); err != nil {
×
1960
                        klog.Errorf("failed to get available subnets for pod %s/%s, %v", pod.Namespace, pod.Name, err)
×
1961
                        return "", "", "", podNet.Subnet, err
×
1962
                }
×
1963
                subnetNames := make([]string, 0, len(nsNets))
×
1964
                for _, net := range nsNets {
×
1965
                        if net.Subnet.Name == subnetStr {
×
1966
                                // allocate from ippools in the subnet specified by pod annotation
×
1967
                                podNet.Subnet = net.Subnet
×
1968
                                subnetNames = []string{net.Subnet.Name}
×
1969
                                break
×
1970
                        }
1971
                        subnetNames = append(subnetNames, net.Subnet.Name)
×
1972
                }
1973

1974
                if subnetStr == "" || slices.Contains(subnetNames, subnetStr) {
×
1975
                        // no subnet specified by pod annotation or specified subnet is in namespace subnets
×
1976
                        if ipPoolList, ok := ns.Annotations[util.IPPoolAnnotation]; ok {
×
1977
                                for ipPoolName := range strings.SplitSeq(ipPoolList, ",") {
×
1978
                                        ippool, err := c.ippoolLister.Get(ipPoolName)
×
1979
                                        if err != nil {
×
1980
                                                klog.Errorf("failed to get ippool %s: %v", ipPoolName, err)
×
1981
                                                return "", "", "", podNet.Subnet, err
×
1982
                                        }
×
1983

1984
                                        switch podNet.Subnet.Spec.Protocol {
×
1985
                                        case kubeovnv1.ProtocolDual:
×
1986
                                                if ippool.Status.V4AvailableIPs.Int64() == 0 || ippool.Status.V6AvailableIPs.Int64() == 0 {
×
1987
                                                        continue
×
1988
                                                }
1989
                                        case kubeovnv1.ProtocolIPv4:
×
1990
                                                if ippool.Status.V4AvailableIPs.Int64() == 0 {
×
1991
                                                        continue
×
1992
                                                }
1993

1994
                                        default:
×
1995
                                                if ippool.Status.V6AvailableIPs.Int64() == 0 {
×
1996
                                                        continue
×
1997
                                                }
1998
                                        }
1999

2000
                                        for _, net := range nsNets {
×
2001
                                                if net.Subnet.Name == ippool.Spec.Subnet && slices.Contains(subnetNames, net.Subnet.Name) {
×
2002
                                                        ippoolStr = ippool.Name
×
2003
                                                        podNet.Subnet = net.Subnet
×
2004
                                                        break
×
2005
                                                }
2006
                                        }
2007
                                        if ippoolStr != "" {
×
2008
                                                break
×
2009
                                        }
2010
                                }
2011
                                if ippoolStr == "" {
×
2012
                                        klog.Infof("no available ippool in subnet(s) %s for pod %s/%s", strings.Join(subnetNames, ","), pod.Namespace, pod.Name)
×
2013
                                        return "", "", "", podNet.Subnet, ipam.ErrNoAvailable
×
2014
                                }
×
2015
                        }
2016
                }
2017
        }
2018

2019
        // Random allocate
2020
        if pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] == "" &&
×
2021
                ippoolStr == "" {
×
2022
                var skippedAddrs []string
×
2023
                for {
×
2024
                        ipv4, ipv6, mac, err := c.ipam.GetRandomAddress(key, portName, macPointer, podNet.Subnet.Name, "", skippedAddrs, !podNet.AllowLiveMigration)
×
2025
                        if err != nil {
×
2026
                                klog.Error(err)
×
2027
                                return "", "", "", podNet.Subnet, err
×
2028
                        }
×
2029
                        ipv4OK, ipv6OK, err := c.validatePodIP(pod.Name, podNet.Subnet.Name, ipv4, ipv6)
×
2030
                        if err != nil {
×
2031
                                klog.Error(err)
×
2032
                                return "", "", "", podNet.Subnet, err
×
2033
                        }
×
2034
                        if ipv4OK && ipv6OK {
×
2035
                                return ipv4, ipv6, mac, podNet.Subnet, nil
×
2036
                        }
×
2037

2038
                        if !ipv4OK {
×
2039
                                skippedAddrs = append(skippedAddrs, ipv4)
×
2040
                        }
×
2041
                        if !ipv6OK {
×
2042
                                skippedAddrs = append(skippedAddrs, ipv6)
×
2043
                        }
×
2044
                }
2045
        }
2046

2047
        // The static ip can be assigned from any subnet after ns supports multi subnets
2048
        if nsNets == nil {
×
2049
                if nsNets, err = c.getNsAvailableSubnets(pod, podNet); err != nil {
×
2050
                        klog.Errorf("failed to get available subnets for pod %s/%s, %v", pod.Namespace, pod.Name, err)
×
2051
                        return "", "", "", podNet.Subnet, err
×
2052
                }
×
2053
        }
2054

2055
        var v4IP, v6IP, mac string
×
2056

×
2057
        // Static allocate
×
2058
        if ipStr := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]; ipStr != "" {
×
2059
                for _, net := range nsNets {
×
2060
                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipStr, macPointer, net.Subnet.Name, net.AllowLiveMigration)
×
2061
                        if err == nil {
×
2062
                                return v4IP, v6IP, mac, net.Subnet, nil
×
2063
                        }
×
2064
                }
2065
                return v4IP, v6IP, mac, podNet.Subnet, err
×
2066
        }
2067

2068
        // IPPool allocate
2069
        if ippoolStr != "" {
×
2070
                var ipPool []string
×
2071
                if strings.ContainsRune(ippoolStr, ';') {
×
2072
                        ipPool = strings.Split(ippoolStr, ";")
×
2073
                } else {
×
2074
                        ipPool = strings.Split(ippoolStr, ",")
×
2075
                        if len(ipPool) == 2 && util.CheckProtocol(ipPool[0]) != util.CheckProtocol(ipPool[1]) {
×
2076
                                ipPool = []string{ippoolStr}
×
2077
                        }
×
2078
                }
2079
                for i, ip := range ipPool {
×
2080
                        ipPool[i] = strings.TrimSpace(ip)
×
2081
                }
×
2082

2083
                if len(ipPool) == 1 && (!strings.ContainsRune(ipPool[0], ',') && net.ParseIP(ipPool[0]) == nil) {
×
2084
                        var skippedAddrs []string
×
2085
                        pool, err := c.ippoolLister.Get(ipPool[0])
×
2086
                        if err != nil {
×
2087
                                klog.Errorf("failed to get ippool %s: %v", ipPool[0], err)
×
2088
                                return "", "", "", podNet.Subnet, err
×
2089
                        }
×
2090
                        for {
×
2091
                                ipv4, ipv6, mac, err := c.ipam.GetRandomAddress(key, portName, macPointer, pool.Spec.Subnet, ipPool[0], skippedAddrs, !podNet.AllowLiveMigration)
×
2092
                                if err != nil {
×
2093
                                        klog.Error(err)
×
2094
                                        return "", "", "", podNet.Subnet, err
×
2095
                                }
×
2096
                                ipv4OK, ipv6OK, err := c.validatePodIP(pod.Name, podNet.Subnet.Name, ipv4, ipv6)
×
2097
                                if err != nil {
×
2098
                                        klog.Error(err)
×
2099
                                        return "", "", "", podNet.Subnet, err
×
2100
                                }
×
2101
                                if ipv4OK && ipv6OK {
×
2102
                                        return ipv4, ipv6, mac, podNet.Subnet, nil
×
2103
                                }
×
2104

2105
                                if !ipv4OK {
×
2106
                                        skippedAddrs = append(skippedAddrs, ipv4)
×
2107
                                }
×
2108
                                if !ipv6OK {
×
2109
                                        skippedAddrs = append(skippedAddrs, ipv6)
×
2110
                                }
×
2111
                        }
2112
                }
2113

2114
                if !isStsPod {
×
2115
                        for _, net := range nsNets {
×
2116
                                for _, staticIP := range ipPool {
×
2117
                                        var checkIP string
×
2118
                                        ipProtocol := util.CheckProtocol(staticIP)
×
2119
                                        if ipProtocol == kubeovnv1.ProtocolDual {
×
2120
                                                checkIP = strings.Split(staticIP, ",")[0]
×
2121
                                        } else {
×
2122
                                                checkIP = staticIP
×
2123
                                        }
×
2124

2125
                                        if assignedPod, ok := c.ipam.IsIPAssignedToOtherPod(checkIP, net.Subnet.Name, key); ok {
×
2126
                                                klog.Errorf("static address %s for %s has been assigned to %s", staticIP, key, assignedPod)
×
2127
                                                continue
×
2128
                                        }
2129

2130
                                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, staticIP, macPointer, net.Subnet.Name, net.AllowLiveMigration)
×
2131
                                        if err == nil {
×
2132
                                                return v4IP, v6IP, mac, net.Subnet, nil
×
2133
                                        }
×
2134
                                }
2135
                        }
2136
                        klog.Errorf("acquire address from ippool %s for %s failed, %v", ippoolStr, key, err)
×
2137
                } else {
×
2138
                        tempStrs := strings.Split(pod.Name, "-")
×
2139
                        numStr := tempStrs[len(tempStrs)-1]
×
2140
                        index, _ := strconv.Atoi(numStr)
×
2141

×
2142
                        if index < len(ipPool) {
×
2143
                                for _, net := range nsNets {
×
2144
                                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipPool[index], macPointer, net.Subnet.Name, net.AllowLiveMigration)
×
2145
                                        if err == nil {
×
2146
                                                return v4IP, v6IP, mac, net.Subnet, nil
×
2147
                                        }
×
2148
                                }
2149
                                klog.Errorf("acquire address %s for %s failed, %v", ipPool[index], key, err)
×
2150
                        }
2151
                }
2152
        }
2153
        klog.Errorf("allocate address for %s failed, return NoAvailableAddress", key)
×
2154
        return "", "", "", podNet.Subnet, ipam.ErrNoAvailable
×
2155
}
2156

2157
func (c *Controller) acquireStaticAddress(key, nicName, ip string, mac *string, subnet string, liveMigration bool) (string, string, string, error) {
×
2158
        var v4IP, v6IP, macStr string
×
2159
        var err error
×
2160
        for ipStr := range strings.SplitSeq(ip, ",") {
×
2161
                if net.ParseIP(ipStr) == nil {
×
2162
                        return "", "", "", fmt.Errorf("failed to parse IP %s", ipStr)
×
2163
                }
×
2164
        }
2165

2166
        if v4IP, v6IP, macStr, err = c.ipam.GetStaticAddress(key, nicName, ip, mac, subnet, !liveMigration); err != nil {
×
2167
                klog.Errorf("failed to get static ip %v, mac %v, subnet %v, err %v", ip, mac, subnet, err)
×
2168
                return "", "", "", err
×
2169
        }
×
2170
        return v4IP, v6IP, macStr, nil
×
2171
}
2172

2173
func appendCheckPodToDel(c *Controller, pod *v1.Pod, ownerRefName, ownerRefKind string) (bool, error) {
×
2174
        // subnet for ns has been changed, and statefulset pod's ip is not in the range of subnet's cidr anymore
×
2175
        podNs, err := c.namespacesLister.Get(pod.Namespace)
×
2176
        if err != nil {
×
2177
                klog.Errorf("failed to get namespace %s, %v", pod.Namespace, err)
×
2178
                return false, err
×
2179
        }
×
2180

2181
        // check if subnet exist in OwnerReference
2182
        var ownerRefSubnetExist bool
×
2183
        var ownerRefSubnet string
×
2184
        switch ownerRefKind {
×
2185
        case util.StatefulSet:
×
2186
                ss, err := c.config.KubeClient.AppsV1().StatefulSets(pod.Namespace).Get(context.Background(), ownerRefName, metav1.GetOptions{})
×
2187
                if err != nil {
×
2188
                        if k8serrors.IsNotFound(err) {
×
2189
                                klog.Infof("Statefulset %s is not found", ownerRefName)
×
2190
                                return true, nil
×
2191
                        }
×
2192
                        klog.Errorf("failed to get StatefulSet %s, %v", ownerRefName, err)
×
2193
                }
2194
                if ss.Spec.Template.Annotations[util.LogicalSwitchAnnotation] != "" {
×
2195
                        ownerRefSubnetExist = true
×
2196
                        ownerRefSubnet = ss.Spec.Template.Annotations[util.LogicalSwitchAnnotation]
×
2197
                }
×
2198

2199
        case util.VMInstance:
×
2200
                vm, err := c.config.KubevirtClient.VirtualMachine(pod.Namespace).Get(context.Background(), ownerRefName, metav1.GetOptions{})
×
2201
                if err != nil {
×
2202
                        if k8serrors.IsNotFound(err) {
×
2203
                                klog.Infof("VirtualMachine %s is not found", ownerRefName)
×
2204
                                return true, nil
×
2205
                        }
×
2206
                        klog.Errorf("failed to get VirtualMachine %s, %v", ownerRefName, err)
×
2207
                }
2208
                if vm != nil &&
×
2209
                        vm.Spec.Template != nil &&
×
2210
                        vm.Spec.Template.ObjectMeta.Annotations != nil &&
×
2211
                        vm.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation] != "" {
×
2212
                        ownerRefSubnetExist = true
×
2213
                        ownerRefSubnet = vm.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation]
×
2214
                }
×
2215
        }
2216
        podSwitch := strings.TrimSpace(pod.Annotations[util.LogicalSwitchAnnotation])
×
2217
        if !ownerRefSubnetExist {
×
2218
                nsSubnetNames := podNs.Annotations[util.LogicalSwitchAnnotation]
×
2219
                // check if pod use the subnet of its ns
×
2220
                if nsSubnetNames != "" && podSwitch != "" && !slices.Contains(strings.Split(nsSubnetNames, ","), podSwitch) {
×
2221
                        klog.Infof("ns %s annotation subnet is %s, which is inconstant with subnet for pod %s, delete pod", pod.Namespace, nsSubnetNames, pod.Name)
×
2222
                        return true, nil
×
2223
                }
×
2224
        }
2225

2226
        // subnet cidr has been changed, and statefulset pod's ip is not in the range of subnet's cidr anymore
2227
        podSubnet, err := c.subnetsLister.Get(podSwitch)
×
2228
        if err != nil {
×
2229
                klog.Errorf("failed to get subnet %s, %v, not auto clean ip", podSwitch, err)
×
2230
                return false, err
×
2231
        }
×
2232
        if podSubnet == nil {
×
2233
                // TODO: remove: CRD get interface will retrun a nil subnet ?
×
2234
                klog.Errorf("pod %s/%s subnet %s is nil, not auto clean ip", pod.Namespace, pod.Name, podSwitch)
×
2235
                return false, nil
×
2236
        }
×
2237
        podIP := pod.Annotations[util.IPAddressAnnotation]
×
2238
        if podIP == "" {
×
2239
                // delete pod just after it created < 1ms
×
2240
                klog.Infof("pod %s/%s annotaions has no ip address, not auto clean ip", pod.Namespace, pod.Name)
×
2241
                return false, nil
×
2242
        }
×
2243
        podSubnetCidr := podSubnet.Spec.CIDRBlock
×
2244
        if podSubnetCidr == "" {
×
2245
                // subnet spec cidr changed by user
×
2246
                klog.Errorf("invalid pod subnet %s empty cidr %s, not auto clean ip", podSwitch, podSubnetCidr)
×
2247
                return false, nil
×
2248
        }
×
2249
        if !util.CIDRContainIP(podSubnetCidr, podIP) {
×
2250
                klog.Infof("pod's ip %s is not in the range of subnet %s, delete pod", pod.Annotations[util.IPAddressAnnotation], podSubnet.Name)
×
2251
                return true, nil
×
2252
        }
×
2253
        // subnet of ownerReference(sts/vm) has been changed, it needs to handle delete pod and create port on the new logical switch
2254
        if ownerRefSubnet != "" && podSubnet.Name != ownerRefSubnet {
×
2255
                klog.Infof("Subnet of owner %s has been changed from %s to %s, delete pod %s/%s", ownerRefName, podSubnet.Name, ownerRefSubnet, pod.Namespace, pod.Name)
×
2256
                return true, nil
×
2257
        }
×
2258

2259
        return false, nil
×
2260
}
2261

2262
func isVMPod(pod *v1.Pod) (bool, string) {
×
2263
        for _, owner := range pod.OwnerReferences {
×
2264
                // The name of vmi is consistent with vm's name.
×
2265
                if owner.Kind == util.VMInstance && strings.HasPrefix(owner.APIVersion, "kubevirt.io") {
×
2266
                        return true, owner.Name
×
2267
                }
×
2268
        }
2269
        return false, ""
×
2270
}
2271

2272
func isOwnsByTheVM(vmi metav1.Object) (bool, string) {
×
2273
        for _, owner := range vmi.GetOwnerReferences() {
×
2274
                if owner.Kind == util.VM && strings.HasPrefix(owner.APIVersion, "kubevirt.io") {
×
2275
                        return true, owner.Name
×
2276
                }
×
2277
        }
2278
        return false, ""
×
2279
}
2280

2281
func (c *Controller) isVMToDel(pod *v1.Pod, vmiName string) bool {
×
2282
        var (
×
2283
                vmiAlive bool
×
2284
                vmName   string
×
2285
        )
×
2286
        // The vmi is also deleted when pod is deleted, only left vm exists.
×
2287
        vmi, err := c.config.KubevirtClient.VirtualMachineInstance(pod.Namespace).Get(context.Background(), vmiName, metav1.GetOptions{})
×
2288
        if err != nil {
×
2289
                if k8serrors.IsNotFound(err) {
×
2290
                        vmiAlive = false
×
2291
                        // The name of vmi is consistent with vm's name.
×
2292
                        vmName = vmiName
×
2293
                        klog.ErrorS(err, "failed to get vmi, will try to get the vm directly", "name", vmiName)
×
2294
                } else {
×
2295
                        klog.ErrorS(err, "failed to get vmi", "name", vmiName)
×
2296
                        return false
×
2297
                }
×
2298
        } else {
×
2299
                var ownsByVM bool
×
2300
                ownsByVM, vmName = isOwnsByTheVM(vmi)
×
2301
                if !ownsByVM && !vmi.DeletionTimestamp.IsZero() {
×
2302
                        klog.Infof("ephemeral vmi %s is deleting", vmiName)
×
2303
                        return true
×
2304
                }
×
2305
                vmiAlive = vmi.DeletionTimestamp.IsZero()
×
2306
        }
2307

2308
        if vmiAlive {
×
2309
                return false
×
2310
        }
×
2311

2312
        vm, err := c.config.KubevirtClient.VirtualMachine(pod.Namespace).Get(context.Background(), vmName, metav1.GetOptions{})
×
2313
        if err != nil {
×
2314
                // the vm has gone
×
2315
                if k8serrors.IsNotFound(err) {
×
2316
                        klog.ErrorS(err, "failed to get vm", "name", vmName)
×
2317
                        return true
×
2318
                }
×
2319
                klog.ErrorS(err, "failed to get vm", "name", vmName)
×
2320
                return false
×
2321
        }
2322

2323
        if !vm.DeletionTimestamp.IsZero() {
×
2324
                klog.Infof("vm %s is deleting", vmName)
×
2325
                return true
×
2326
        }
×
2327
        return false
×
2328
}
2329

2330
func (c *Controller) getNameByPod(pod *v1.Pod) string {
×
2331
        if c.config.EnableKeepVMIP {
×
2332
                if isVMPod, vmName := isVMPod(pod); isVMPod {
×
2333
                        return vmName
×
2334
                }
×
2335
        }
2336
        return pod.Name
×
2337
}
2338

2339
// When subnet's v4availableIPs is 0 but still there's available ip in exclude-ips, the static ip in exclude-ips can be allocated normal.
2340
func (c *Controller) getNsAvailableSubnets(pod *v1.Pod, podNet *kubeovnNet) ([]*kubeovnNet, error) {
×
2341
        // keep the annotation subnet of the pod in first position
×
2342
        result := []*kubeovnNet{podNet}
×
2343

×
2344
        ns, err := c.namespacesLister.Get(pod.Namespace)
×
2345
        if err != nil {
×
2346
                klog.Errorf("failed to get namespace %s, %v", pod.Namespace, err)
×
2347
                return nil, err
×
2348
        }
×
2349
        if ns.Annotations == nil {
×
2350
                return []*kubeovnNet{}, nil
×
2351
        }
×
2352

2353
        subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]
×
2354
        for subnetName := range strings.SplitSeq(subnetNames, ",") {
×
2355
                if subnetName == "" || subnetName == podNet.Subnet.Name {
×
2356
                        continue
×
2357
                }
2358
                subnet, err := c.subnetsLister.Get(subnetName)
×
2359
                if err != nil {
×
2360
                        klog.Errorf("failed to get subnet %v", err)
×
2361
                        return nil, err
×
2362
                }
×
2363

2364
                result = append(result, &kubeovnNet{
×
2365
                        Type:         providerTypeOriginal,
×
2366
                        ProviderName: subnet.Spec.Provider,
×
2367
                        Subnet:       subnet,
×
2368
                })
×
2369
        }
2370

2371
        return result, nil
×
2372
}
2373

2374
func getPodType(pod *v1.Pod) string {
×
2375
        if ok, _, _ := isStatefulSetPod(pod); ok {
×
2376
                return util.StatefulSet
×
2377
        }
×
2378

2379
        if isVMPod, _ := isVMPod(pod); isVMPod {
×
2380
                return util.VM
×
2381
        }
×
2382
        return ""
×
2383
}
2384

2385
func (c *Controller) getVirtualIPs(pod *v1.Pod, podNets []*kubeovnNet) map[string]string {
×
2386
        vipsListMap := make(map[string][]string)
×
2387
        var vipNamesList []string
×
2388
        for vipName := range strings.SplitSeq(strings.TrimSpace(pod.Annotations[util.AAPsAnnotation]), ",") {
×
2389
                if vipName = strings.TrimSpace(vipName); vipName == "" {
×
2390
                        continue
×
2391
                }
2392
                if !slices.Contains(vipNamesList, vipName) {
×
2393
                        vipNamesList = append(vipNamesList, vipName)
×
2394
                } else {
×
2395
                        continue
×
2396
                }
2397
                vip, err := c.virtualIpsLister.Get(vipName)
×
2398
                if err != nil {
×
2399
                        klog.Errorf("failed to get vip %s, %v", vipName, err)
×
2400
                        continue
×
2401
                }
2402
                if vip.Spec.Namespace != pod.Namespace || (vip.Status.V4ip == "" && vip.Status.V6ip == "") {
×
2403
                        continue
×
2404
                }
2405
                for _, podNet := range podNets {
×
2406
                        if podNet.Subnet.Name == vip.Spec.Subnet {
×
2407
                                key := fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)
×
2408
                                vipsList := vipsListMap[key]
×
2409
                                if vipsList == nil {
×
2410
                                        vipsList = []string{}
×
2411
                                }
×
2412
                                // ipam will ensure the uniqueness of VIP
2413
                                if util.IsValidIP(vip.Status.V4ip) {
×
2414
                                        vipsList = append(vipsList, vip.Status.V4ip)
×
2415
                                }
×
2416
                                if util.IsValidIP(vip.Status.V6ip) {
×
2417
                                        vipsList = append(vipsList, vip.Status.V6ip)
×
2418
                                }
×
2419

2420
                                vipsListMap[key] = vipsList
×
2421
                        }
2422
                }
2423
        }
2424

2425
        for _, podNet := range podNets {
×
2426
                vipStr := pod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]
×
2427
                if vipStr == "" {
×
2428
                        continue
×
2429
                }
2430
                key := fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)
×
2431
                vipsList := vipsListMap[key]
×
2432
                if vipsList == nil {
×
2433
                        vipsList = []string{}
×
2434
                }
×
2435

2436
                for vip := range strings.SplitSeq(vipStr, ",") {
×
2437
                        if util.IsValidIP(vip) && !slices.Contains(vipsList, vip) {
×
2438
                                vipsList = append(vipsList, vip)
×
2439
                        }
×
2440
                }
2441

2442
                vipsListMap[key] = vipsList
×
2443
        }
2444

2445
        vipsMap := make(map[string]string)
×
2446
        for key, vipsList := range vipsListMap {
×
2447
                vipsMap[key] = strings.Join(vipsList, ",")
×
2448
        }
×
2449
        return vipsMap
×
2450
}
2451

2452
func setPodRoutesAnnotation(annotations map[string]string, provider string, routes []request.Route) error {
×
2453
        key := fmt.Sprintf(util.RoutesAnnotationTemplate, provider)
×
2454
        if len(routes) == 0 {
×
2455
                delete(annotations, key)
×
2456
                return nil
×
2457
        }
×
2458

2459
        buf, err := json.Marshal(routes)
×
2460
        if err != nil {
×
2461
                err = fmt.Errorf("failed to marshal routes %+v: %w", routes, err)
×
2462
                klog.Error(err)
×
2463
                return err
×
2464
        }
×
2465
        annotations[key] = string(buf)
×
2466

×
2467
        return nil
×
2468
}
2469

2470
// Check if pod is a VPC NAT gateway using pod annotations
2471
func (c *Controller) checkIsPodVpcNatGw(pod *v1.Pod) (bool, string) {
1✔
2472
        if pod == nil {
2✔
2473
                return false, ""
1✔
2474
        }
1✔
2475
        if pod.Annotations == nil {
2✔
2476
                return false, ""
1✔
2477
        }
1✔
2478
        // default provider
2479
        providerName := util.OvnProvider
1✔
2480
        // In non-primary CNI mode, we get the providers from the pod annotations
1✔
2481
        // We get the vpc nat gw name using the provider name
1✔
2482
        if c.config.EnableNonPrimaryCNI {
2✔
2483
                // get providers
1✔
2484
                providers, err := c.getPodProviders(pod)
1✔
2485
                if err != nil {
1✔
2486
                        klog.Errorf("failed to get pod %s/%s providers, %v", pod.Namespace, pod.Name, err)
×
2487
                        return false, ""
×
2488
                }
×
2489
                if len(providers) > 0 {
2✔
2490
                        // use the first provider
1✔
2491
                        providerName = providers[0]
1✔
2492
                }
1✔
2493
        }
2494
        vpcGwName, isVpcNatGw := pod.Annotations[fmt.Sprintf(util.VpcNatGatewayAnnotationTemplate, providerName)]
1✔
2495
        if isVpcNatGw {
2✔
2496
                if vpcGwName == "" {
2✔
2497
                        klog.Errorf("pod %s is vpc nat gateway but name is empty", pod.Name)
1✔
2498
                        return false, ""
1✔
2499
                }
1✔
2500
                klog.Infof("pod %s is vpc nat gateway %s", pod.Name, vpcGwName)
1✔
2501
        }
2502
        return isVpcNatGw, vpcGwName
1✔
2503
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc