kubeovn/kube-ovn | Build 12545917939 | pkg/controller/pod.go | Coveralls

46

func NewNamedPort() *NamedPort {

×

47

        return &NamedPort{

×

48

                mutex:        sync.RWMutex{},

×

49

                namedPortMap: map[string]map[string]*util.NamedPortInfo{},

×

50

×

51

×

53

func (n *NamedPort) AddNamedPortByPod(pod *v1.Pod) {

×

54

        n.mutex.Lock()

×

55

        defer n.mutex.Unlock()

×

56

        ns := pod.Namespace

×

57

        podName := pod.Name

×

58

×

59

        if pod.Spec.Containers == nil {

×

60

                return

×

61

×

63

        for _, container := range pod.Spec.Containers {

×

64

                if container.Ports == nil {

×

65

                        continue

×

68

                for _, port := range container.Ports {

×

69

                        if port.Name == "" || port.ContainerPort == 0 {

×

70

                                continue

×

73

                        if _, ok := n.namedPortMap[ns]; ok {

×

74

                                if _, ok := n.namedPortMap[ns][port.Name]; ok {

×

75

                                        if n.namedPortMap[ns][port.Name].PortID == port.ContainerPort {

×

76

                                                n.namedPortMap[ns][port.Name].Pods.Add(podName)

×

77

                                        } else {

×

78

                                                klog.Warningf("named port %s has already be defined with portID %d",

×

79

                                                        port.Name, n.namedPortMap[ns][port.Name].PortID)

×

80

×

81

                                        continue

×

83

                        } else {

×

84

                                n.namedPortMap[ns] = make(map[string]*util.NamedPortInfo)

×

85

×

86

                        n.namedPortMap[ns][port.Name] = &util.NamedPortInfo{

×

87

                                PortID: port.ContainerPort,

×

88

                                Pods:   strset.New(podName),

×

89

×

94

func (n *NamedPort) DeleteNamedPortByPod(pod *v1.Pod) {

×

95

        n.mutex.Lock()

×

96

        defer n.mutex.Unlock()

×

97

×

98

        ns := pod.Namespace

×

99

        podName := pod.Name

×

100

×

101

        if pod.Spec.Containers == nil {

×

102

                return

×

103

×

105

        for _, container := range pod.Spec.Containers {

×

106

                if container.Ports == nil {

×

107

                        continue

×

110

                for _, port := range container.Ports {

×

111

                        if port.Name == "" {

×

112

                                continue

×

115

                        if _, ok := n.namedPortMap[ns]; !ok {

×

116

                                continue

×

119

                        if _, ok := n.namedPortMap[ns][port.Name]; !ok {

×

120

                                continue

×

123

                        if !n.namedPortMap[ns][port.Name].Pods.Has(podName) {

×

124

                                continue

×

127

                        n.namedPortMap[ns][port.Name].Pods.Remove(podName)

×

128

                        if n.namedPortMap[ns][port.Name].Pods.Size() == 0 {

×

129

                                delete(n.namedPortMap[ns], port.Name)

×

130

                                if len(n.namedPortMap[ns]) == 0 {

×

131

                                        delete(n.namedPortMap, ns)

×

132

×

138

func (n *NamedPort) GetNamedPortByNs(namespace string) map[string]*util.NamedPortInfo {

×

139

        n.mutex.RLock()

×

140

        defer n.mutex.RUnlock()

×

141

×

142

        if result, ok := n.namedPortMap[namespace]; ok {

×

143

                for portName, info := range result {

×

144

                        klog.Infof("namespace %s's namedPort portname is %s with info %v", namespace, portName, info)

×

145

×

146

                return result

×

148

        return nil

×

151

func isPodAlive(p *v1.Pod) bool {

×

152

        if !p.DeletionTimestamp.IsZero() && p.DeletionGracePeriodSeconds != nil {

×

153

                now := time.Now()

×

154

                deletionTime := p.DeletionTimestamp.Time

×

155

                gracePeriod := time.Duration(*p.DeletionGracePeriodSeconds) * time.Second

×

156

                if now.After(deletionTime.Add(gracePeriod)) {

×

157

                        return false

×

158

×

160

        return isPodStatusPhaseAlive(p)

×

163

func isPodStatusPhaseAlive(p *v1.Pod) bool {

×

164

        if p.Status.Phase == v1.PodSucceeded && p.Spec.RestartPolicy != v1.RestartPolicyAlways {

×

165

                return false

×

166

×

168

        if p.Status.Phase == v1.PodFailed && p.Spec.RestartPolicy == v1.RestartPolicyNever {

×

169

                return false

×

170

×

172

        if p.Status.Phase == v1.PodFailed && p.Status.Reason == "Evicted" {

×

173

                return false

×

174

×

175

        return true

×

178

func (c *Controller) enqueueAddPod(obj interface{}) {

×

179

        p := obj.(*v1.Pod)

×

180

        if p.Spec.HostNetwork {

×

181

                return

×

182

×

185

        if c.config.EnableNP {

×

186

                c.namedPort.AddNamedPortByPod(p)

×

187

                if p.Status.PodIP != "" {

×

188

                        for _, np := range c.podMatchNetworkPolicies(p) {

×

189

                                klog.V(3).Infof("enqueue update network policy %s", np)

×

190

                                c.updateNpQueue.Add(np)

×

191

×

195

        key := cache.MetaObjectToName(p).String()

×

196

        if !isPodAlive(p) {

×

197

                isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(p)

×

198

                isVMPod, vmName := isVMPod(p)

×

199

                if isStateful || (isVMPod && c.config.EnableKeepVMIP) {

×

200

                        if isStateful && isStatefulSetPodToDel(c.config.KubeClient, p, statefulSetName, statefulSetUID) {

×

201

                                klog.V(3).Infof("enqueue delete pod %s", key)

×

202

                                c.deletingPodObjMap.Store(key, p)

×

203

                                c.deletePodQueue.Add(key)

×

204

×

205

                        if isVMPod && c.isVMToDel(p, vmName) {

×

206

                                klog.V(3).Infof("enqueue delete pod %s", key)

×

207

                                c.deletingPodObjMap.Store(key, p)

×

208

                                c.deletePodQueue.Add(key)

×

209

×

210

                } else {

×

211

                        klog.V(3).Infof("enqueue delete pod %s", key)

×

212

                        c.deletingPodObjMap.Store(key, p)

×

213

                        c.deletePodQueue.Add(key)

×

214

×

215

                return

×

218

        need, err := c.podNeedSync(p)

×

219

        if err != nil {

×

220

                klog.Errorf("invalid pod net: %v", err)

×

221

                return

×

222

×

223

        if need {

×

224

                klog.Infof("enqueue add pod %s", key)

×

225

                c.addOrUpdatePodQueue.Add(key)

×

226

×

229

func (c *Controller) enqueueDeletePod(obj interface{}) {

×

230

        p := obj.(*v1.Pod)

×

231

        if p.Spec.HostNetwork {

×

232

                return

×

233

×

235

        if c.config.EnableNP {

×

236

                c.namedPort.DeleteNamedPortByPod(p)

×

237

                for _, np := range c.podMatchNetworkPolicies(p) {

×

238

                        c.updateNpQueue.Add(np)

×

239

×

242

        if c.config.EnableANP {

×

243

                podNs, _ := c.namespacesLister.Get(obj.(*v1.Pod).Namespace)

×

244

                c.updateAnpsByLabelsMatch(podNs.Labels, obj.(*v1.Pod).Labels)

×

245

×

247

        key := cache.MetaObjectToName(p).String()

×

248

        klog.Infof("enqueue delete pod %s", key)

×

249

        c.deletingPodObjMap.Store(key, p)

×

250

        c.deletePodQueue.Add(key)

×

253

func (c *Controller) enqueueUpdatePod(oldObj, newObj interface{}) {

×

254

        oldPod := oldObj.(*v1.Pod)

×

255

        newPod := newObj.(*v1.Pod)

×

256

×

257

        if oldPod.Annotations[util.AAPsAnnotation] != "" || newPod.Annotations[util.AAPsAnnotation] != "" {

×

258

                oldAAPs := strings.Split(oldPod.Annotations[util.AAPsAnnotation], ",")

×

259

                newAAPs := strings.Split(newPod.Annotations[util.AAPsAnnotation], ",")

×

260

                var vipNames []string

×

261

                for _, vipName := range oldAAPs {

×

262

                        vipNames = append(vipNames, strings.TrimSpace(vipName))

×

263

×

264

                for _, vipName := range newAAPs {

×

265

                        vipName = strings.TrimSpace(vipName)

×

266

                        if !slices.Contains(vipNames, vipName) {

×

267

                                vipNames = append(vipNames, vipName)

×

268

×

270

                for _, vipName := range vipNames {

×

271

                        if vip, err := c.virtualIpsLister.Get(vipName); err == nil {

×

272

                                if vip.Spec.Namespace != newPod.Namespace {

×

273

                                        continue

×

275

                                klog.Infof("enqueue update virtual parents for %s", vipName)

×

276

                                c.updateVirtualParentsQueue.Add(vipName)

×

281

        if oldPod.ResourceVersion == newPod.ResourceVersion {

×

282

                return

×

283

×

284

        if newPod.Spec.HostNetwork {

×

285

                return

×

286

×

288

        podNets, err := c.getPodKubeovnNets(newPod)

×

289

        if err != nil {

×

290

                klog.Errorf("failed to get newPod nets %v", err)

×

291

                return

×

292

×

294

        key := cache.MetaObjectToName(newPod).String()

×

295

        if c.config.EnableNP {

×

296

                c.namedPort.AddNamedPortByPod(newPod)

×

297

                newNp := c.podMatchNetworkPolicies(newPod)

×

298

                if !reflect.DeepEqual(oldPod.Labels, newPod.Labels) {

×

299

                        oldNp := c.podMatchNetworkPolicies(oldPod)

×

300

                        for _, np := range util.DiffStringSlice(oldNp, newNp) {

×

301

                                c.updateNpQueue.Add(np)

×

302

×

305

                for _, podNet := range podNets {

×

306

                        oldAllocated := oldPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]

×

307

                        newAllocated := newPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]

×

308

                        if oldAllocated != newAllocated {

×

309

                                for _, np := range newNp {

×

310

                                        klog.V(3).Infof("enqueue update network policy %s for pod %s", np, key)

×

311

                                        c.updateNpQueue.Add(np)

×

312

×

313

                                break

×

318

        if c.config.EnableANP {

×

319

                podNs, _ := c.namespacesLister.Get(newPod.Namespace)

×

320

                if !reflect.DeepEqual(oldPod.Labels, newPod.Labels) {

×

321

                        c.updateAnpsByLabelsMatch(podNs.Labels, newPod.Labels)

×

322

×

324

                for _, podNet := range podNets {

×

325

                        oldAllocated := oldPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]

×

326

                        newAllocated := newPod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)]

×

327

                        if oldAllocated != newAllocated {

×

328

                                c.updateAnpsByLabelsMatch(podNs.Labels, newPod.Labels)

×

329

                                break

×

334

        isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(newPod)

×

335

        isVMPod, vmName := isVMPod(newPod)

×

336

        if !isPodStatusPhaseAlive(newPod) && !isStateful && !isVMPod {

×

337

                klog.V(3).Infof("enqueue delete pod %s", key)

×

338

                c.deletingPodObjMap.Store(key, newPod)

×

339

                c.deletePodQueue.Add(key)

×

340

                return

×

341

×

344

        var delay time.Duration

×

345

        if newPod.Spec.TerminationGracePeriodSeconds != nil {

×

346

                if !newPod.DeletionTimestamp.IsZero() {

×

347

                        delay = time.Until(newPod.DeletionTimestamp.Add(time.Duration(*newPod.Spec.TerminationGracePeriodSeconds) * time.Second))

×

348

                } else {

×

349

                        delay = time.Duration(*newPod.Spec.TerminationGracePeriodSeconds) * time.Second

×

350

×

353

        if !newPod.DeletionTimestamp.IsZero() && !isStateful && !isVMPod {

×

354

                go func() {

×

355

                        // In case node get lost and pod can not be deleted,

×

356

                        // the ip address will not be recycled

×

357

                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)

×

358

                        c.deletingPodObjMap.Store(key, newPod)

×

359

                        c.deletePodQueue.AddAfter(key, delay)

×

360

}()

×

361

                return

×

365

        if isStateful && isStatefulSetPodToDel(c.config.KubeClient, newPod, statefulSetName, statefulSetUID) {

×

366

                go func() {

×

367

                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)

×

368

                        c.deletingPodObjMap.Store(key, newPod)

×

369

                        c.deletePodQueue.AddAfter(key, delay)

×

370

}()

×

371

                return

×

373

        if isVMPod && c.isVMToDel(newPod, vmName) {

×

374

                go func() {

×

375

                        klog.V(3).Infof("enqueue delete pod %s after %v", key, delay)

×

376

                        c.deletingPodObjMap.Store(key, newPod)

×

377

                        c.deletePodQueue.AddAfter(key, delay)

×

378

}()

×

379

                return

×

381

        klog.Infof("enqueue update pod %s", key)

×

382

        c.addOrUpdatePodQueue.Add(key)

×

383

×

384

        // security policy changed

×

385

        for _, podNet := range podNets {

×

386

                oldSecurity := oldPod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)]

×

387

                newSecurity := newPod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)]

×

388

                oldSg := oldPod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]

×

389

                newSg := newPod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]

×

390

                oldVips := oldPod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]

×

391

                newVips := newPod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]

×

392

                oldAAPs := oldPod.Annotations[util.AAPsAnnotation]

×

393

                newAAPs := newPod.Annotations[util.AAPsAnnotation]

×

394

                if oldSecurity != newSecurity || oldSg != newSg || oldVips != newVips || oldAAPs != newAAPs {

×

395

                        c.updatePodSecurityQueue.Add(key)

×

396

                        break

×

401

func (c *Controller) getPodKubeovnNets(pod *v1.Pod) ([]*kubeovnNet, error) {

×

402

        defaultSubnet, err := c.getPodDefaultSubnet(pod)

×

403

        if err != nil {

×

404

                klog.Error(err)

×

405

                return nil, err

×

406

×

408

        attachmentNets, err := c.getPodAttachmentNet(pod)

×

409

        if err != nil {

×

410

                klog.Error(err)

×

411

                return nil, err

×

412

×

414

        podNets := attachmentNets

×

415

        if _, hasOtherDefaultNet := pod.Annotations[util.DefaultNetworkAnnotation]; !hasOtherDefaultNet {

×

416

                podNets = append(attachmentNets, &kubeovnNet{

×

417

                        Type:         providerTypeOriginal,

×

418

                        ProviderName: util.OvnProvider,

×

419

                        Subnet:       defaultSubnet,

×

420

                        IsDefault:    true,

×

421

})

×

422

×

424

        return podNets, nil

×

427

func (c *Controller) handleAddOrUpdatePod(key string) (err error) {

×

428

        namespace, name, err := cache.SplitMetaNamespaceKey(key)

×

429

        if err != nil {

×

430

                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))

×

431

                return nil

×

432

×

434

        c.podKeyMutex.LockKey(key)

×

435

        defer func() { _ = c.podKeyMutex.UnlockKey(key) }()

×

436

        klog.Infof("handle add/update pod %s", key)

×

437

×

438

        pod, err := c.podsLister.Pods(namespace).Get(name)

×

439

        if err != nil {

×

440

                if k8serrors.IsNotFound(err) {

×

441

                        return nil

×

442

×

443

                klog.Error(err)

×

444

                return err

×

446

        if err := util.ValidatePodNetwork(pod.Annotations); err != nil {

×

447

                klog.Errorf("validate pod %s/%s failed: %v", namespace, name, err)

×

448

                c.recorder.Eventf(pod, v1.EventTypeWarning, "ValidatePodNetworkFailed", err.Error())

×

449

                return err

×

450

×

452

        podNets, err := c.getPodKubeovnNets(pod)

×

453

        if err != nil {

×

454

                klog.Errorf("failed to get pod nets %v", err)

×

455

                return err

×

456

×

459

        if pod, err = c.syncKubeOvnNet(pod, podNets); err != nil {

×

460

                klog.Errorf("failed to sync pod nets %v", err)

×

461

                return err

×

462

×

463

        if pod == nil {

×

464

                // pod has been deleted

×

465

                return nil

×

466

×

467

        needAllocatePodNets := needAllocateSubnets(pod, podNets)

×

468

        if len(needAllocatePodNets) != 0 {

×

469

                if pod, err = c.reconcileAllocateSubnets(pod, needAllocatePodNets); err != nil {

×

470

                        klog.Error(err)

×

471

                        return err

×

472

×

473

                if pod == nil {

×

474

                        // pod has been deleted

×

475

                        return nil

×

476

×

480

        return c.reconcileRouteSubnets(pod, needRouteSubnets(pod, podNets))

×

484

func (c *Controller) reconcileAllocateSubnets(pod *v1.Pod, needAllocatePodNets []*kubeovnNet) (*v1.Pod, error) {

×

485

        namespace := pod.Namespace

×

486

        name := pod.Name

×

487

        klog.Infof("sync pod %s/%s allocated", namespace, name)

×

488

×

489

        vipsMap := c.getVirtualIPs(pod, needAllocatePodNets)

×

490

        isVMPod, vmName := isVMPod(pod)

×

491

        podType := getPodType(pod)

×

492

        podName := c.getNameByPod(pod)

×

493

        // todo: isVmPod, getPodType, getNameByPod has duplicated logic

×

494

×

495

        var err error

×

496

        var vmKey string

×

497

        if isVMPod && c.config.EnableKeepVMIP {

×

498

                vmKey = fmt.Sprintf("%s/%s", namespace, vmName)

×

499

×

501

        patch := util.KVPatch{}

×

502

        for _, podNet := range needAllocatePodNets {

×

503

                // the subnet may changed when alloc static ip from the latter subnet after ns supports multi subnets

×

504

                v4IP, v6IP, mac, subnet, err := c.acquireAddress(pod, podNet)

×

505

                if err != nil {

×

506

                        c.recorder.Eventf(pod, v1.EventTypeWarning, "AcquireAddressFailed", err.Error())

×

507

                        klog.Error(err)

×

508

                        return nil, err

×

509

×

510

                ipStr := util.GetStringIP(v4IP, v6IP)

×

511

                patch[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] = ipStr

×

512

                if mac == "" {

×

513

                        patch[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = nil

×

514

                } else {

×

515

                        patch[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = mac

×

516

×

517

                patch[fmt.Sprintf(util.CidrAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.CIDRBlock

×

518

                patch[fmt.Sprintf(util.GatewayAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.Gateway

×

519

                if isOvnSubnet(podNet.Subnet) {

×

520

                        patch[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName)] = subnet.Name

×

521

                        if pod.Annotations[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] == "" {

×

522

                                patch[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] = c.config.PodNicType

×

523

×

524

                } else {

×

525

                        patch[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName)] = nil

×

526

                        patch[fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName)] = nil

×

527

×

528

                patch[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] = "true"

×

529

                if vmKey != "" {

×

530

                        patch[fmt.Sprintf(util.VMAnnotationTemplate, podNet.ProviderName)] = vmName

×

531

×

532

                if err := util.ValidateNetworkBroadcast(podNet.Subnet.Spec.CIDRBlock, ipStr); err != nil {

×

533

                        klog.Errorf("validate pod %s/%s failed: %v", namespace, name, err)

×

534

                        c.recorder.Eventf(pod, v1.EventTypeWarning, "ValidatePodNetworkFailed", err.Error())

×

535

                        return nil, err

×

536

×

538

                if podNet.Type != providerTypeIPAM {

×

539

                        if (subnet.Spec.Vlan == "" || subnet.Spec.LogicalGateway || subnet.Spec.U2OInterconnection) && subnet.Spec.Vpc != "" {

×

540

                                patch[fmt.Sprintf(util.LogicalRouterAnnotationTemplate, podNet.ProviderName)] = subnet.Spec.Vpc

×

541

×

543

                        if subnet.Spec.Vlan != "" {

×

544

                                vlan, err := c.vlansLister.Get(subnet.Spec.Vlan)

×

545

                                if err != nil {

×

546

                                        klog.Error(err)

×

547

                                        c.recorder.Eventf(pod, v1.EventTypeWarning, "GetVlanInfoFailed", err.Error())

×

548

                                        return nil, err

×

549

×

550

                                patch[fmt.Sprintf(util.VlanIDAnnotationTemplate, podNet.ProviderName)] = strconv.Itoa(vlan.Spec.ID)

×

551

                                patch[fmt.Sprintf(util.ProviderNetworkTemplate, podNet.ProviderName)] = vlan.Spec.Provider

×

554

                        portSecurity := false

×

555

                        if pod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)] == "true" {

×

556

                                portSecurity = true

×

557

×

559

                        vips := vipsMap[fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)]

×

560

                        for _, ip := range strings.Split(vips, ",") {

×

561

                                if ip != "" && net.ParseIP(ip) == nil {

×

562

                                        klog.Errorf("invalid vip address '%s' for pod %s", ip, name)

×

563

                                        vips = ""

×

564

                                        break

×

568

                        portName := ovs.PodNameToPortName(podName, namespace, podNet.ProviderName)

×

569

                        dhcpOptions := &ovs.DHCPOptionsUUIDs{

×

570

                                DHCPv4OptionsUUID: subnet.Status.DHCPv4OptionsUUID,

×

571

                                DHCPv6OptionsUUID: subnet.Status.DHCPv6OptionsUUID,

×

572

×

573

×

574

                        securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]

×

575

                        securityGroups := strings.ReplaceAll(securityGroupAnnotation, " ", "")

×

576

                        if err := c.OVNNbClient.CreateLogicalSwitchPort(subnet.Name, portName, ipStr, mac, podName, pod.Namespace,

×

577

                                portSecurity, securityGroupAnnotation, vips, podNet.Subnet.Spec.EnableDHCP, dhcpOptions, subnet.Spec.Vpc); err != nil {

×

578

                                c.recorder.Eventf(pod, v1.EventTypeWarning, "CreateOVNPortFailed", err.Error())

×

579

                                klog.Errorf("%v", err)

×

580

                                return nil, err

×

581

×

583

                        if pod.Annotations[fmt.Sprintf(util.Layer2ForwardAnnotationTemplate, podNet.ProviderName)] == "true" {

×

584

                                if err := c.OVNNbClient.EnablePortLayer2forward(portName); err != nil {

×

585

                                        c.recorder.Eventf(pod, v1.EventTypeWarning, "SetOVNPortL2ForwardFailed", err.Error())

×

586

                                        klog.Errorf("%v", err)

×

587

                                        return nil, err

×

588

×

591

                        if securityGroupAnnotation != "" {

×

592

                                sgNames := strings.Split(securityGroups, ",")

×

593

                                for _, sgName := range sgNames {

×

594

                                        if sgName != "" {

×

595

                                                c.syncSgPortsQueue.Add(sgName)

×

596

×

600

                        if vips != "" {

×

601

                                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)

×

602

×

605

                ipCRName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

606

                if err := c.createOrUpdateIPCR(ipCRName, podName, ipStr, mac, subnet.Name, pod.Namespace, pod.Spec.NodeName, podType); err != nil {

×

607

                        err = fmt.Errorf("failed to create ips CR %s.%s: %w", podName, pod.Namespace, err)

×

608

                        klog.Error(err)

×

609

                        return nil, err

×

610

×

612

        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(namespace), name, patch); err != nil {

×

613

                if k8serrors.IsNotFound(err) {

×

614

                        // Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.

×

615

                        // Then we need to recycle the resource again.

×

616

                        key := strings.Join([]string{namespace, name}, "/")

×

617

                        c.deletingPodObjMap.Store(key, pod)

×

618

                        c.deletePodQueue.AddRateLimited(key)

×

619

                        return nil, nil

×

620

×

621

                klog.Errorf("failed to patch pod %s/%s: %v", namespace, name, err)

×

622

                return nil, err

×

625

        if pod, err = c.config.KubeClient.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{}); err != nil {

×

626

                if k8serrors.IsNotFound(err) {

×

627

                        key := strings.Join([]string{namespace, name}, "/")

×

628

                        c.deletingPodObjMap.Store(key, pod)

×

629

                        c.deletePodQueue.AddRateLimited(key)

×

630

                        return nil, nil

×

631

×

632

                klog.Errorf("failed to get pod %s/%s: %v", namespace, name, err)

×

633

                return nil, err

×

636

        if vpcGwName, isVpcNatGw := pod.Annotations[util.VpcNatGatewayAnnotation]; isVpcNatGw {

×

637

                c.initVpcNatGatewayQueue.Add(vpcGwName)

×

638

×

639

        return pod, nil

×

643

func (c *Controller) reconcileRouteSubnets(pod *v1.Pod, needRoutePodNets []*kubeovnNet) error {

×

644

        // the lb-svc pod has dependencies on Running state, check it when pod state get updated

×

645

        if err := c.checkAndReInitLbSvcPod(pod); err != nil {

×

646

                klog.Errorf("failed to init iptable rules for load-balancer pod %s/%s: %v", pod.Namespace, pod.Name, err)

×

647

×

649

        if len(needRoutePodNets) == 0 {

×

650

                return nil

×

651

×

653

        namespace := pod.Namespace

×

654

        name := pod.Name

×

655

        podName := c.getNameByPod(pod)

×

656

×

657

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

×

658

×

659

        var podIP string

×

660

        var subnet *kubeovnv1.Subnet

×

661

        patch := util.KVPatch{}

×

662

        for _, podNet := range needRoutePodNets {

×

663

                // in case update handler overlap the annotation when cache is not in sync

×

664

                if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "" {

×

665

                        return fmt.Errorf("no address has been allocated to %s/%s", namespace, name)

×

666

×

668

                podIP = pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]

×

669

                subnet = podNet.Subnet

×

670

×

671

                // Check if pod uses nodeSwitch subnet

×

672

                if subnet.Name == c.config.NodeSwitch {

×

673

                        return fmt.Errorf("NodeSwitch subnet %s is unavailable for pod", subnet.Name)

×

674

×

676

                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

677

                if (!c.config.EnableLb || !(subnet.Spec.EnableLb != nil && *subnet.Spec.EnableLb)) &&

×

678

                        subnet.Spec.Vpc == c.config.ClusterRouter &&

×

679

                        subnet.Spec.U2OInterconnection &&

×

680

                        subnet.Spec.Vlan != "" &&

×

681

                        !subnet.Spec.LogicalGateway {

×

682

                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, pod.Spec.NodeName)

×

683

                        if err := c.OVNNbClient.PortGroupAddPorts(pgName, portName); err != nil {

×

684

                                klog.Errorf("failed to add port to u2o port group %s: %v", pgName, err)

×

685

                                return err

×

686

×

689

                if podIP != "" && (subnet.Spec.Vlan == "" || subnet.Spec.LogicalGateway) && subnet.Spec.Vpc == c.config.ClusterRouter {

×

690

                        node, err := c.nodesLister.Get(pod.Spec.NodeName)

×

691

                        if err != nil {

×

692

                                klog.Errorf("failed to get node %s: %v", pod.Spec.NodeName, err)

×

693

                                return err

×

694

×

696

                        pgName := getOverlaySubnetsPortGroupName(subnet.Name, node.Name)

×

697

                        if c.config.EnableEipSnat && (pod.Annotations[util.EipAnnotation] != "" || pod.Annotations[util.SnatAnnotation] != "") {

×

698

                                cm, err := c.configMapsLister.ConfigMaps(c.config.ExternalGatewayConfigNS).Get(util.ExternalGatewayConfig)

×

699

                                if err != nil {

×

700

                                        klog.Errorf("failed to get ex-gateway config, %v", err)

×

701

                                        return err

×

702

×

703

                                nextHop := cm.Data["external-gw-addr"]

×

704

                                if nextHop == "" {

×

705

                                        externalSubnet, err := c.subnetsLister.Get(c.config.ExternalGatewaySwitch)

×

706

                                        if err != nil {

×

707

                                                klog.Errorf("failed to get subnet %s, %v", c.config.ExternalGatewaySwitch, err)

×

708

                                                return err

×

709

×

710

                                        nextHop = externalSubnet.Spec.Gateway

×

711

                                        if nextHop == "" {

×

712

                                                klog.Errorf("no available gateway address")

×

713

                                                return errors.New("no available gateway address")

×

714

×

716

                                if strings.Contains(nextHop, "/") {

×

717

                                        nextHop = strings.Split(nextHop, "/")[0]

×

718

×

720

                                if err := c.addPolicyRouteToVpc(

×

721

                                        subnet.Spec.Vpc,

×

722

                                        &kubeovnv1.PolicyRoute{

×

723

                                                Priority:  util.NorthGatewayRoutePolicyPriority,

×

724

                                                Match:     fmt.Sprintf("ip4.src == %s", podIP),

×

725

                                                Action:    kubeovnv1.PolicyRouteActionReroute,

×

726

                                                NextHopIP: nextHop,

×

727

},

×

728

                                        map[string]string{

×

729

                                                "vendor": util.CniTypeName,

×

730

                                                "subnet": subnet.Name,

×

731

},

×

732

                                ); err != nil {

×

733

                                        klog.Errorf("failed to add policy route, %v", err)

×

734

                                        return err

×

735

×

738

                                if err = c.OVNNbClient.PortGroupRemovePorts(pgName, portName); err != nil {

×

739

                                        klog.Error(err)

×

740

                                        return err

×

741

×

742

                        } else {

×

743

                                if subnet.Spec.GatewayType == kubeovnv1.GWDistributedType && pod.Annotations[util.NorthGatewayAnnotation] == "" {

×

744

                                        nodeTunlIPAddr, err := getNodeTunlIP(node)

×

745

                                        if err != nil {

×

746

                                                klog.Error(err)

×

747

                                                return err

×

748

×

750

                                        var added bool

×

751

×

752

                                        for _, nodeAddr := range nodeTunlIPAddr {

×

753

                                                for _, podAddr := range strings.Split(podIP, ",") {

×

754

                                                        if util.CheckProtocol(nodeAddr.String()) != util.CheckProtocol(podAddr) {

×

755

                                                                continue

×

758

                                                        if err := c.OVNNbClient.PortGroupAddPorts(pgName, portName); err != nil {

×

759

                                                                klog.Errorf("add port to port group %s: %v", pgName, err)

×

760

                                                                return err

×

761

×

763

                                                        added = true

×

764

                                                        break

×

766

                                                if added {

×

767

                                                        break

×

772

                                if pod.Annotations[util.NorthGatewayAnnotation] != "" && pod.Annotations[util.IPAddressAnnotation] != "" {

×

773

                                        for _, podAddr := range strings.Split(pod.Annotations[util.IPAddressAnnotation], ",") {

×

774

                                                if util.CheckProtocol(podAddr) != util.CheckProtocol(pod.Annotations[util.NorthGatewayAnnotation]) {

×

775

                                                        continue

×

777

                                                ipSuffix := "ip4"

×

778

                                                if util.CheckProtocol(podAddr) == kubeovnv1.ProtocolIPv6 {

×

779

                                                        ipSuffix = "ip6"

×

780

×

782

                                                if err := c.addPolicyRouteToVpc(

×

783

                                                        subnet.Spec.Vpc,

×

784

                                                        &kubeovnv1.PolicyRoute{

×

785

                                                                Priority:  util.NorthGatewayRoutePolicyPriority,

×

786

                                                                Match:     fmt.Sprintf("%s.src == %s", ipSuffix, podAddr),

×

787

                                                                Action:    kubeovnv1.PolicyRouteActionReroute,

×

788

                                                                NextHopIP: pod.Annotations[util.NorthGatewayAnnotation],

×

789

},

×

790

                                                        map[string]string{

×

791

                                                                "vendor": util.CniTypeName,

×

792

                                                                "subnet": subnet.Name,

×

793

},

×

794

                                                ); err != nil {

×

795

                                                        klog.Errorf("failed to add policy route, %v", err)

×

796

                                                        return err

×

797

×

799

                                } else if c.config.EnableEipSnat {

×

800

                                        if err = c.deleteStaticRouteFromVpc(

×

801

                                                c.config.ClusterRouter,

×

802

                                                subnet.Spec.RouteTable,

×

803

                                                podIP,

×

804

"",

×

805

                                                kubeovnv1.PolicyDst,

×

806

                                        ); err != nil {

×

807

                                                klog.Error(err)

×

808

                                                return err

×

809

×

813

                        if c.config.EnableEipSnat {

×

814

                                for _, ipStr := range strings.Split(podIP, ",") {

×

815

                                        if eip := pod.Annotations[util.EipAnnotation]; eip == "" {

×

816

                                                if err = c.OVNNbClient.DeleteNats(c.config.ClusterRouter, ovnnb.NATTypeDNATAndSNAT, ipStr); err != nil {

×

817

                                                        klog.Errorf("failed to delete nat rules: %v", err)

×

818

×

819

                                        } else if util.CheckProtocol(eip) == util.CheckProtocol(ipStr) {

×

820

                                                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 {

×

821

                                                        klog.Errorf("failed to add nat rules, %v", err)

×

822

                                                        return err

×

823

×

825

                                        if eip := pod.Annotations[util.SnatAnnotation]; eip == "" {

×

826

                                                if err = c.OVNNbClient.DeleteNats(c.config.ClusterRouter, ovnnb.NATTypeSNAT, ipStr); err != nil {

×

827

                                                        klog.Errorf("failed to delete nat rules: %v", err)

×

828

×

829

                                        } else if util.CheckProtocol(eip) == util.CheckProtocol(ipStr) {

×

830

                                                if err = c.OVNNbClient.UpdateSnat(c.config.ClusterRouter, eip, ipStr); err != nil {

×

831

                                                        klog.Errorf("failed to add nat rules, %v", err)

×

832

                                                        return err

×

833

×

839

                if pod.Annotations[fmt.Sprintf(util.ActivationStrategyTemplate, podNet.ProviderName)] != "" {

×

840

                        if err := c.OVNNbClient.SetLogicalSwitchPortActivationStrategy(portName, pod.Spec.NodeName); err != nil {

×

841

                                klog.Errorf("failed to set activation strategy for lsp %s: %v", portName, err)

×

842

                                return err

×

843

×

846

                patch[fmt.Sprintf(util.RoutedAnnotationTemplate, podNet.ProviderName)] = "true"

×

848

        if err := util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(namespace), name, patch); err != nil {

×

849

                if k8serrors.IsNotFound(err) {

×

850

                        // Sometimes pod is deleted between kube-ovn configure ovn-nb and patch pod.

×

851

                        // Then we need to recycle the resource again.

×

852

                        key := strings.Join([]string{namespace, name}, "/")

×

853

                        c.deletingPodObjMap.Store(key, pod)

×

854

                        c.deletePodQueue.AddRateLimited(key)

×

855

                        return nil

×

856

×

857

                klog.Errorf("failed to patch pod %s/%s: %v", namespace, name, err)

×

858

                return err

×

860

        return nil

×

863

func (c *Controller) handleDeletePod(key string) (err error) {

×

864

        pod, ok := c.deletingPodObjMap.Load(key)

×

865

        if !ok {

×

866

                return nil

×

867

×

868

        podName := c.getNameByPod(pod)

×

869

        c.podKeyMutex.LockKey(key)

×

870

        defer func() {

×

871

                _ = c.podKeyMutex.UnlockKey(key)

×

872

                if err == nil {

×

873

                        c.deletingPodObjMap.Delete(key)

×

874

×

876

        klog.Infof("handle delete pod %s", key)

×

877

×

878

        p, _ := c.podsLister.Pods(pod.Namespace).Get(pod.Name)

×

879

        if p != nil && p.UID != pod.UID {

×

880

                // Pod with same name exists, just return here

×

881

                return nil

×

882

×

884

        if aaps := pod.Annotations[util.AAPsAnnotation]; aaps != "" {

×

885

                for _, vipName := range strings.Split(aaps, ",") {

×

886

                        if vip, err := c.virtualIpsLister.Get(vipName); err == nil {

×

887

                                if vip.Spec.Namespace != pod.Namespace {

×

888

                                        continue

×

890

                                klog.Infof("enqueue update virtual parents for %s", vipName)

×

891

                                c.updateVirtualParentsQueue.Add(vipName)

×

896

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

×

897

×

898

        var keepIPCR bool

×

899

        if ok, stsName, stsUID := isStatefulSetPod(pod); ok {

×

900

                if pod.DeletionTimestamp != nil {

×

901

                        klog.Infof("handle deletion of sts pod %s", podName)

×

902

                        toDel := isStatefulSetPodToDel(c.config.KubeClient, pod, stsName, stsUID)

×

903

                        if !toDel {

×

904

                                klog.Infof("try keep ip for sts pod %s", podKey)

×

905

                                keepIPCR = true

×

906

×

908

                if keepIPCR {

×

909

                        isDelete, err := appendCheckPodToDel(c, pod, stsName, util.StatefulSet)

×

910

                        if err != nil {

×

911

                                klog.Error(err)

×

912

                                return err

×

913

×

914

                        if isDelete {

×

915

                                klog.Infof("not keep ip for sts pod %s", podKey)

×

916

                                keepIPCR = false

×

917

×

920

        isVMPod, vmName := isVMPod(pod)

×

921

        if isVMPod && c.config.EnableKeepVMIP {

×

922

                ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": podKey})

×

923

                if err != nil {

×

924

                        klog.Errorf("failed to list lsps of pod '%s', %v", pod.Name, err)

×

925

                        return err

×

926

×

927

                for _, port := range ports {

×

928

                        if err := c.OVNNbClient.CleanLogicalSwitchPortMigrateOptions(port.Name); err != nil {

×

929

                                err = fmt.Errorf("failed to clean migrate options for vm lsp %s, %w", port.Name, err)

×

930

                                klog.Error(err)

×

931

                                return err

×

932

×

934

                if pod.DeletionTimestamp != nil {

×

935

                        klog.Infof("handle deletion of vm pod %s", podName)

×

936

                        vmToBeDel := c.isVMToDel(pod, vmName)

×

937

                        if !vmToBeDel {

×

938

                                klog.Infof("try keep ip for vm pod %s", podKey)

×

939

                                keepIPCR = true

×

940

×

942

                if keepIPCR {

×

943

                        isDelete, err := appendCheckPodToDel(c, pod, vmName, util.VMInstance)

×

944

                        if err != nil {

×

945

                                klog.Error(err)

×

946

                                return err

×

947

×

948

                        if isDelete {

×

949

                                klog.Infof("not keep ip for vm pod %s", podKey)

×

950

                                keepIPCR = false

×

951

×

955

        podNets, err := c.getPodKubeovnNets(pod)

×

956

        if err != nil {

×

957

                klog.Errorf("failed to get pod nets %v", err)

×

958

×

959

        if !keepIPCR {

×

960

                ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": podKey})

×

961

                if err != nil {

×

962

                        klog.Errorf("failed to list lsps of pod '%s', %v", pod.Name, err)

×

963

                        return err

×

964

×

966

                if len(ports) != 0 {

×

967

                        addresses := c.ipam.GetPodAddress(podKey)

×

968

                        for _, address := range addresses {

×

969

                                if strings.TrimSpace(address.IP) == "" {

×

970

                                        continue

×

972

                                subnet, err := c.subnetsLister.Get(address.Subnet.Name)

×

973

                                if k8serrors.IsNotFound(err) {

×

974

                                        continue

×

975

                                } else if err != nil {

×

976

                                        klog.Error(err)

×

977

                                        return err

×

978

×

979

                                vpc, err := c.vpcsLister.Get(subnet.Spec.Vpc)

×

980

                                if k8serrors.IsNotFound(err) {

×

981

                                        continue

×

982

                                } else if err != nil {

×

983

                                        klog.Error(err)

×

984

                                        return err

×

985

×

987

                                ipSuffix := "ip4"

×

988

                                if util.CheckProtocol(address.IP) == kubeovnv1.ProtocolIPv6 {

×

989

                                        ipSuffix = "ip6"

×

990

×

991

                                if err = c.deletePolicyRouteFromVpc(

×

992

                                        vpc.Name,

×

993

                                        util.NorthGatewayRoutePolicyPriority,

×

994

                                        fmt.Sprintf("%s.src == %s", ipSuffix, address.IP),

×

995

                                ); err != nil {

×

996

                                        klog.Errorf("failed to delete static route, %v", err)

×

997

                                        return err

×

998

×

1000

                                if c.config.EnableEipSnat {

×

1001

                                        if pod.Annotations[util.EipAnnotation] != "" {

×

1002

                                                if err = c.OVNNbClient.DeleteNat(c.config.ClusterRouter, ovnnb.NATTypeDNATAndSNAT, pod.Annotations[util.EipAnnotation], address.IP); err != nil {

×

1003

                                                        klog.Errorf("failed to delete nat rules: %v", err)

×

1004

×

1006

                                        if pod.Annotations[util.SnatAnnotation] != "" {

×

1007

                                                if err = c.OVNNbClient.DeleteNat(c.config.ClusterRouter, ovnnb.NATTypeSNAT, "", address.IP); err != nil {

×

1008

                                                        klog.Errorf("failed to delete nat rules: %v", err)

×

1009

×

1014

                for _, port := range ports {

×

1015

                        // when lsp is deleted, the port of pod is deleted from any port-group automatically.

×

1016

                        klog.Infof("delete logical switch port %s", port.Name)

×

1017

                        if err := c.OVNNbClient.DeleteLogicalSwitchPort(port.Name); err != nil {

×

1018

                                klog.Errorf("failed to delete lsp %s, %v", port.Name, err)

×

1019

                                return err

×

1020

×

1022

                klog.Infof("try release all ip address for deleting pod %s", podKey)

×

1023

                for _, podNet := range podNets {

×

1024

                        portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1025

                        ipCR, err := c.ipsLister.Get(portName)

×

1026

                        if err != nil {

×

1027

                                if k8serrors.IsNotFound(err) {

×

1028

                                        continue

×

1030

                                klog.Errorf("failed to get ip %s, %v", portName, err)

×

1031

                                return err

×

1033

                        if ipCR.Labels[util.IPReservedLabel] != "true" {

×

1034

                                klog.Infof("delete ip CR %s", ipCR.Name)

×

1035

                                if err := c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), ipCR.Name, metav1.DeleteOptions{}); err != nil {

×

1036

                                        if !k8serrors.IsNotFound(err) {

×

1037

                                                klog.Errorf("failed to delete ip %s, %v", ipCR.Name, err)

×

1038

                                                return err

×

1039

×

1042

                                c.ipam.ReleaseAddressByPod(podKey, podNet.Subnet.Name)

×

1045

                if pod.Annotations[util.VipAnnotation] != "" {

×

1046

                        if err = c.releaseVip(pod.Annotations[util.VipAnnotation]); err != nil {

×

1047

                                klog.Errorf("failed to clean label from vip %s, %v", pod.Annotations[util.VipAnnotation], err)

×

1048

                                return err

×

1049

×

1052

        for _, podNet := range podNets {

×

1053

                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)

×

1054

                securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]

×

1055

                if securityGroupAnnotation != "" {

×

1056

                        securityGroups := strings.ReplaceAll(securityGroupAnnotation, " ", "")

×

1057

                        for _, sgName := range strings.Split(securityGroups, ",") {

×

1058

                                if sgName != "" {

×

1059

                                        c.syncSgPortsQueue.Add(sgName)

×

1060

×

1064

        return nil

×

1067

func (c *Controller) handleUpdatePodSecurity(key string) error {

×

1068

        namespace, name, err := cache.SplitMetaNamespaceKey(key)

×

1069

        if err != nil {

×

1070

                utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))

×

1071

                return nil

×

1072

×

1074

        c.podKeyMutex.LockKey(key)

×

1075

        defer func() { _ = c.podKeyMutex.UnlockKey(key) }()

×

1077

        pod, err := c.podsLister.Pods(namespace).Get(name)

×

1078

        if err != nil {

×

1079

                if k8serrors.IsNotFound(err) {

×

1080

                        return nil

×

1081

×

1082

                klog.Error(err)

×

1083

                return err

×

1085

        podName := c.getNameByPod(pod)

×

1086

×

1087

        klog.Infof("update pod %s/%s security", namespace, name)

×

1088

×

1089

        podNets, err := c.getPodKubeovnNets(pod)

×

1090

        if err != nil {

×

1091

                klog.Errorf("failed to pod nets %v", err)

×

1092

                return err

×

1093

×

1095

        vipsMap := c.getVirtualIPs(pod, podNets)

×

1096

×

1097

        // associated with security group

×

1098

        for _, podNet := range podNets {

×

1099

                portSecurity := false

×

1100

                if pod.Annotations[fmt.Sprintf(util.PortSecurityAnnotationTemplate, podNet.ProviderName)] == "true" {

×

1101

                        portSecurity = true

×

1102

×

1104

                mac := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]

×

1105

                ipStr := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]

×

1106

                vips := vipsMap[fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)]

×

1107

×

1108

                if err = c.OVNNbClient.SetLogicalSwitchPortSecurity(portSecurity, ovs.PodNameToPortName(podName, namespace, podNet.ProviderName), mac, ipStr, vips); err != nil {

×

1109

                        klog.Errorf("set logical switch port security: %v", err)

×

1110

                        return err

×

1111

×

1113

                c.syncVirtualPortsQueue.Add(podNet.Subnet.Name)

×

1114

                securityGroupAnnotation := pod.Annotations[fmt.Sprintf(util.SecurityGroupAnnotationTemplate, podNet.ProviderName)]

×

1115

                var securityGroups string

×

1116

                if securityGroupAnnotation != "" {

×

1117

                        securityGroups = strings.ReplaceAll(securityGroupAnnotation, " ", "")

×

1118

                        for _, sgName := range strings.Split(securityGroups, ",") {

×

1119

                                if sgName != "" {

×

1120

                                        c.syncSgPortsQueue.Add(sgName)

×

1121

×

1124

                if err = c.reconcilePortSg(ovs.PodNameToPortName(podName, namespace, podNet.ProviderName), securityGroups); err != nil {

×

1125

                        klog.Errorf("reconcilePortSg failed. %v", err)

×

1126

                        return err

×

1127

×

1129

        return nil

×

1132

func (c *Controller) syncKubeOvnNet(pod *v1.Pod, podNets []*kubeovnNet) (*v1.Pod, error) {

×

1133

        podName := c.getNameByPod(pod)

×

1134

        key := fmt.Sprintf("%s/%s", pod.Namespace, podName)

×

1135

        targetPortNameList := strset.NewWithSize(len(podNets))

×

1136

        portsNeedToDel := []string{}

×

1137

        annotationsNeedToDel := []string{}

×

1138

        annotationsNeedToAdd := make(map[string]string)

×

1139

        subnetUsedByPort := make(map[string]string)

×

1140

×

1141

        for _, podNet := range podNets {

×

1142

                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1143

                targetPortNameList.Add(portName)

×

1144

                if podNet.IPRequest != "" {

×

1145

                        klog.Infof("pod %s/%s use custom IP %s for provider %s", pod.Namespace, pod.Name, podNet.IPRequest, podNet.ProviderName)

×

1146

                        annotationsNeedToAdd[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] = podNet.IPRequest

×

1147

×

1149

                if podNet.MacRequest != "" {

×

1150

                        klog.Infof("pod %s/%s use custom MAC %s for provider %s", pod.Namespace, pod.Name, podNet.MacRequest, podNet.ProviderName)

×

1151

                        annotationsNeedToAdd[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)] = podNet.MacRequest

×

1152

×

1155

        ports, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{"pod": key})

×

1156

        if err != nil {

×

1157

                klog.Errorf("failed to list lsps of pod '%s', %v", pod.Name, err)

×

1158

                return nil, err

×

1159

×

1161

        for _, port := range ports {

×

1162

                if !targetPortNameList.Has(port.Name) {

×

1163

                        portsNeedToDel = append(portsNeedToDel, port.Name)

×

1164

                        subnetUsedByPort[port.Name] = port.ExternalIDs["ls"]

×

1165

                        portNameSlice := strings.Split(port.Name, ".")

×

1166

                        providerName := strings.Join(portNameSlice[2:], ".")

×

1167

                        if providerName == util.OvnProvider {

×

1168

                                continue

×

1170

                        annotationsNeedToDel = append(annotationsNeedToDel, providerName)

×

1174

        if len(portsNeedToDel) == 0 && len(annotationsNeedToAdd) == 0 {

×

1175

                return pod, nil

×

1176

×

1178

        for _, portNeedDel := range portsNeedToDel {

×

1179

                klog.Infof("release port %s for pod %s", portNeedDel, podName)

×

1180

                if subnet, ok := c.ipam.Subnets[subnetUsedByPort[portNeedDel]]; ok {

×

1181

                        subnet.ReleaseAddressWithNicName(podName, portNeedDel)

×

1182

×

1183

                if err := c.OVNNbClient.DeleteLogicalSwitchPort(portNeedDel); err != nil {

×

1184

                        klog.Errorf("failed to delete lsp %s, %v", portNeedDel, err)

×

1185

                        return nil, err

×

1186

×

1187

                if err := c.config.KubeOvnClient.KubeovnV1().IPs().Delete(context.Background(), portNeedDel, metav1.DeleteOptions{}); err != nil {

×

1188

                        if !k8serrors.IsNotFound(err) {

×

1189

                                klog.Errorf("failed to delete ip %s, %v", portNeedDel, err)

×

1190

                                return nil, err

×

1191

×

1195

        patch := util.KVPatch{}

×

1196

        for _, providerName := range annotationsNeedToDel {

×

1197

                for key := range pod.Annotations {

×

1198

                        if strings.HasPrefix(key, providerName) {

×

1199

                                patch[key] = nil

×

1200

×

1204

        for key, value := range annotationsNeedToAdd {

×

1205

                patch[key] = value

×

1206

×

1208

        if len(patch) == 0 {

×

1209

                return pod, nil

×

1210

×

1212

        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Pods(pod.Namespace), pod.Name, patch); err != nil {

×

1213

                if k8serrors.IsNotFound(err) {

×

1214

                        return nil, nil

×

1215

×

1216

                klog.Errorf("failed to clean annotations for pod %s/%s: %v", pod.Namespace, pod.Name, err)

×

1217

                return nil, err

×

1220

        if pod, err = c.config.KubeClient.CoreV1().Pods(pod.Namespace).Get(context.TODO(), pod.Name, metav1.GetOptions{}); err != nil {

×

1221

                if k8serrors.IsNotFound(err) {

×

1222

                        return nil, nil

×

1223

×

1224

                klog.Errorf("failed to get pod %s/%s: %v", pod.Namespace, pod.Name, err)

×

1225

                return nil, err

×

1228

        return pod, nil

×

1231

func isStatefulSetPod(pod *v1.Pod) (bool, string, types.UID) {

×

1232

        for _, owner := range pod.OwnerReferences {

×

1233

                if owner.Kind == util.StatefulSet && strings.HasPrefix(owner.APIVersion, "apps/") {

×

1234

                        if strings.HasPrefix(pod.Name, owner.Name) {

×

1235

                                return true, owner.Name, owner.UID

×

1236

×

1239

        return false, "", ""

×

1242

func isStatefulSetPodToDel(c kubernetes.Interface, pod *v1.Pod, statefulSetName string, statefulSetUID types.UID) bool {

×

1243

        // only delete statefulset pod lsp when statefulset deleted or down scaled

×

1244

        sts, err := c.AppsV1().StatefulSets(pod.Namespace).Get(context.Background(), statefulSetName, metav1.GetOptions{})

×

1245

        if err != nil {

×

1246

                // statefulset is deleted

×

1247

                if k8serrors.IsNotFound(err) {

×

1248

                        klog.Infof("statefulset %s is deleted", statefulSetName)

×

1249

                        return true

×

1250

×

1251

                klog.Errorf("failed to get statefulset %v", err)

×

1252

                return false

×

1256

        if !sts.DeletionTimestamp.IsZero() || sts.UID != statefulSetUID {

×

1257

                klog.Infof("statefulset %s is being deleted", statefulSetName)

×

1258

                return true

×

1259

×

1262

        tempStrs := strings.Split(pod.Name, "-")

×

1263

        numStr := tempStrs[len(tempStrs)-1]

×

1264

        index, err := strconv.ParseInt(numStr, 10, 0)

×

1265

        if err != nil {

×

1266

                klog.Errorf("failed to parse %s to int", numStr)

×

1267

                return false

×

1268

×

1270

        if index >= int64(*sts.Spec.Replicas) {

×

1271

                klog.Infof("statefulset %s is down scaled", statefulSetName)

×

1272

                return true

×

1273

×

1274

        return false

×

1277

func getNodeTunlIP(node *v1.Node) ([]net.IP, error) {

×

1278

        var nodeTunlIPAddr []net.IP

×

1279

        nodeTunlIP := node.Annotations[util.IPAddressAnnotation]

×

1280

        if nodeTunlIP == "" {

×

1281

                return nil, errors.New("node has no tunnel ip annotation")

×

1282

×

1284

        for _, ip := range strings.Split(nodeTunlIP, ",") {

×

1285

                nodeTunlIPAddr = append(nodeTunlIPAddr, net.ParseIP(ip))

×

1286

×

1287

        return nodeTunlIPAddr, nil

×

1290

func getNextHopByTunnelIP(gw []net.IP) string {

×

1291

        // validation check by caller

×

1292

        nextHop := gw[0].String()

×

1293

        if len(gw) == 2 {

×

1294

                nextHop = gw[0].String() + "," + gw[1].String()

×

1295

×

1296

        return nextHop

×

1299

func needAllocateSubnets(pod *v1.Pod, nets []*kubeovnNet) []*kubeovnNet {

×

1300

        // check if allocate from subnet is need.

×

1301

        // allocate subnet when change subnet to hotplug nic

×

1302

        // allocate subnet when migrate vm

×

1303

        if !isPodAlive(pod) {

×

1304

                return nil

×

1305

×

1307

        if pod.Annotations == nil {

×

1308

                return nets

×

1309

×

1311

        migrate := false

×

1312

        if job, ok := pod.Annotations[util.MigrationJobAnnotation]; ok {

×

1313

                klog.Infof("pod %s/%s is in the migration job %s", pod.Namespace, pod.Name, job)

×

1314

                migrate = true

×

1315

×

1317

        result := make([]*kubeovnNet, 0, len(nets))

×

1318

        for _, n := range nets {

×

1319

                if migrate || pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, n.ProviderName)] != "true" {

×

1320

                        result = append(result, n)

×

1321

×

1323

        return result

×

1326

func (c *Controller) podNeedSync(pod *v1.Pod) (bool, error) {

×

1327

        // 1. check annotations

×

1328

        if pod.Annotations == nil {

×

1329

                return true, nil

×

1330

×

1332

        if pod.Annotations[util.RoutedAnnotation] != "true" {

×

1333

                return true, nil

×

1334

×

1336

        attachmentNets, err := c.getPodAttachmentNet(pod)

×

1337

        if err != nil {

×

1338

                klog.Error(err)

×

1339

                return false, err

×

1340

×

1341

        for _, n := range attachmentNets {

×

1342

                if pod.Annotations[fmt.Sprintf(util.RoutedAnnotationTemplate, n.ProviderName)] != "true" {

×

1343

                        return true, nil

×

1344

×

1345

                ipName := ovs.PodNameToPortName(pod.Name, pod.Namespace, n.ProviderName)

×

1346

                if _, err = c.ipsLister.Get(ipName); err != nil {

×

1347

                        err = fmt.Errorf("pod has no ip %s: %w", ipName, err)

×

1348

                        // need to sync to create ip

×

1349

                        klog.Error(err)

×

1350

                        return true, nil

×

1351

×

1353

        return false, nil

×

1356

func needRouteSubnets(pod *v1.Pod, nets []*kubeovnNet) []*kubeovnNet {

×

1357

        if !isPodAlive(pod) {

×

1358

                return nil

×

1359

×

1361

        if pod.Annotations == nil {

×

1362

                return nets

×

1363

×

1365

        result := make([]*kubeovnNet, 0, len(nets))

×

1366

        for _, n := range nets {

×

1367

                if !isOvnSubnet(n.Subnet) {

×

1368

                        continue

×

1371

                if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, n.ProviderName)] == "true" && pod.Spec.NodeName != "" {

×

1372

                        if pod.Annotations[fmt.Sprintf(util.RoutedAnnotationTemplate, n.ProviderName)] != "true" {

×

1373

                                result = append(result, n)

×

1374

×

1377

        return result

×

1380

func (c *Controller) getPodDefaultSubnet(pod *v1.Pod) (*kubeovnv1.Subnet, error) {

×

1381

        var subnet *kubeovnv1.Subnet

×

1382

        var err error

×

1383

        // 1. check annotation subnet

×

1384

        lsName, lsExist := pod.Annotations[util.LogicalSwitchAnnotation]

×

1385

        if lsExist {

×

1386

                subnet, err = c.subnetsLister.Get(lsName)

×

1387

                if err != nil {

×

1388

                        klog.Errorf("failed to get subnet %v", err)

×

1389

                        return nil, err

×

1390

×

1391

        } else {

×

1392

                ns, err := c.namespacesLister.Get(pod.Namespace)

×

1393

                if err != nil {

×

1394

                        klog.Errorf("failed to get namespace %s, %v", pod.Namespace, err)

×

1395

                        return nil, err

×

1396

×

1397

                if ns.Annotations == nil {

×

1398

                        err = fmt.Errorf("namespace %s network annotations is nil", pod.Namespace)

×

1399

                        klog.Error(err)

×

1400

                        return nil, err

×

1401

×

1403

                subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]

×

1404

                for _, subnetName := range strings.Split(subnetNames, ",") {

×

1405

                        if subnetName == "" {

×

1406

                                err = fmt.Errorf("namespace %s default logical switch is not found", pod.Namespace)

×

1407

                                klog.Error(err)

×

1408

                                return nil, err

×

1409

×

1410

                        subnet, err = c.subnetsLister.Get(subnetName)

×

1411

                        if err != nil {

×

1412

                                klog.Errorf("failed to get subnet %v", err)

×

1413

                                return nil, err

×

1414

×

1416

                        switch subnet.Spec.Protocol {

×

1417

                        case kubeovnv1.ProtocolIPv4:

×

1418

                                fallthrough

×

1419

                        case kubeovnv1.ProtocolDual:

×

1420

                                if subnet.Status.V4AvailableIPs == 0 {

×

1421

                                        klog.V(3).Infof("there's no available ips for subnet %v, try next subnet", subnet.Name)

×

1422

                                        continue

×

1424

                        case kubeovnv1.ProtocolIPv6:

×

1425

                                if subnet.Status.V6AvailableIPs == 0 {

×

1426

                                        klog.Infof("there's no available ips for subnet %v, try next subnet", subnet.Name)

×

1427

                                        continue

×

1430

                        break

×

1433

        return subnet, nil

×

1436

func loadNetConf(bytes []byte) (*multustypes.DelegateNetConf, error) {

×

1437

        delegateConf := &multustypes.DelegateNetConf{}

×

1438

        if err := json.Unmarshal(bytes, &delegateConf.Conf); err != nil {

×

1439

                return nil, logging.Errorf("LoadDelegateNetConf: error unmarshalling delegate config: %v", err)

×

1440

×

1442

        if delegateConf.Conf.Type == "" {

×

1443

                if err := multustypes.LoadDelegateNetConfList(bytes, delegateConf); err != nil {

×

1444

                        return nil, logging.Errorf("LoadDelegateNetConf: failed with: %v", err)

×

1445

×

1447

        return delegateConf, nil

×

1467

func (c *Controller) getPodAttachmentNet(pod *v1.Pod) ([]*kubeovnNet, error) {

×

1468

        var multusNets []*nadv1.NetworkSelectionElement

×

1469

        defaultAttachNetworks := pod.Annotations[util.DefaultNetworkAnnotation]

×

1470

        if defaultAttachNetworks != "" {

×

1471

                attachments, err := nadutils.ParseNetworkAnnotation(defaultAttachNetworks, pod.Namespace)

×

1472

                if err != nil {

×

1473

                        klog.Errorf("failed to parse default attach net for pod '%s', %v", pod.Name, err)

×

1474

                        return nil, err

×

1475

×

1476

                multusNets = attachments

×

1479

        attachNetworks := pod.Annotations[nadv1.NetworkAttachmentAnnot]

×

1480

        if attachNetworks != "" {

×

1481

                attachments, err := nadutils.ParseNetworkAnnotation(attachNetworks, pod.Namespace)

×

1482

                if err != nil {

×

1483

                        klog.Errorf("failed to parse attach net for pod '%s', %v", pod.Name, err)

×

1484

                        return nil, err

×

1485

×

1486

                multusNets = append(multusNets, attachments...)

×

1488

        subnets, err := c.subnetsLister.List(labels.Everything())

×

1489

        if err != nil {

×

1490

                return nil, err

×

1491

×

1493

        result := make([]*kubeovnNet, 0, len(multusNets))

×

1494

        for _, attach := range multusNets {

×

1495

                networkClient := c.config.AttachNetClient.K8sCniCncfIoV1().NetworkAttachmentDefinitions(attach.Namespace)

×

1496

                network, err := networkClient.Get(context.Background(), attach.Name, metav1.GetOptions{})

×

1497

                if err != nil {

×

1498

                        klog.Errorf("failed to get net-attach-def %s, %v", attach.Name, err)

×

1499

                        return nil, err

×

1500

×

1502

                netCfg, err := loadNetConf([]byte(network.Spec.Config))

×

1503

                if err != nil {

×

1504

                        klog.Errorf("failed to load config of net-attach-def %s, %v", attach.Name, err)

×

1505

                        return nil, err

×

1506

×

1509

                var providerName string

×

1510

                if util.IsOvnNetwork(netCfg) {

×

1511

                        allowLiveMigration := false

×

1512

                        isDefault := util.IsDefaultNet(pod.Annotations[util.DefaultNetworkAnnotation], attach)

×

1513

×

1514

                        providerName = fmt.Sprintf("%s.%s.%s", attach.Name, attach.Namespace, util.OvnProvider)

×

1515

                        if pod.Annotations[util.MigrationJobAnnotation] != "" {

×

1516

                                allowLiveMigration = true

×

1517

×

1519

                        subnetName := pod.Annotations[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, providerName)]

×

1520

                        if subnetName == "" {

×

1521

                                for _, subnet := range subnets {

×

1522

                                        if subnet.Spec.Provider == providerName {

×

1523

                                                subnetName = subnet.Name

×

1524

                                                break

×

1528

                        var subnet *kubeovnv1.Subnet

×

1529

                        if subnetName == "" {

×

1530

                                subnet, err = c.getPodDefaultSubnet(pod)

×

1531

                                if err != nil {

×

1532

                                        klog.Errorf("failed to pod default subnet, %v", err)

×

1533

                                        return nil, err

×

1534

×

1535

                        } else {

×

1536

                                subnet, err = c.subnetsLister.Get(subnetName)

×

1537

                                if err != nil {

×

1538

                                        klog.Errorf("failed to get subnet %s, %v", subnetName, err)

×

1539

                                        return nil, err

×

1540

×

1543

                        ret := &kubeovnNet{

×

1544

                                Type:               providerTypeOriginal,

×

1545

                                ProviderName:       providerName,

×

1546

                                Subnet:             subnet,

×

1547

                                IsDefault:          isDefault,

×

1548

                                AllowLiveMigration: allowLiveMigration,

×

1549

                                MacRequest:         attach.MacRequest,

×

1550

                                IPRequest:          strings.Join(attach.IPRequest, ","),

×

1551

×

1552

                        result = append(result, ret)

×

1553

                } else {

×

1554

                        providerName = fmt.Sprintf("%s.%s", attach.Name, attach.Namespace)

×

1555

                        for _, subnet := range subnets {

×

1556

                                if subnet.Spec.Provider == providerName {

×

1557

                                        result = append(result, &kubeovnNet{

×

1558

                                                Type:         providerTypeIPAM,

×

1559

                                                ProviderName: providerName,

×

1560

                                                Subnet:       subnet,

×

1561

                                                MacRequest:   attach.MacRequest,

×

1562

                                                IPRequest:    strings.Join(attach.IPRequest, ","),

×

1563

})

×

1564

                                        break

×

1569

        return result, nil

×

1572

func (c *Controller) validatePodIP(podName, subnetName, ipv4, ipv6 string) (bool, bool, error) {

×

1573

        subnet, err := c.subnetsLister.Get(subnetName)

×

1574

        if err != nil {

×

1575

                klog.Errorf("failed to get subnet %s: %v", subnetName, err)

×

1576

                return false, false, err

×

1577

×

1579

        if subnet.Spec.Vlan == "" && subnet.Spec.Vpc == c.config.ClusterRouter {

×

1580

                nodes, err := c.nodesLister.List(labels.Everything())

×

1581

                if err != nil {

×

1582

                        klog.Errorf("failed to list nodes: %v", err)

×

1583

                        return false, false, err

×

1584

×

1586

                for _, node := range nodes {

×

1587

                        nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)

×

1588

                        if ipv4 != "" && ipv4 == nodeIPv4 {

×

1589

                                klog.Errorf("IP address (%s) assigned to pod %s is the same with internal IP address of node %s, reallocating...", ipv4, podName, node.Name)

×

1590

                                return false, true, nil

×

1591

×

1592

                        if ipv6 != "" && ipv6 == nodeIPv6 {

×

1593

                                klog.Errorf("IP address (%s) assigned to pod %s is the same with internal IP address of node %s, reallocating...", ipv6, podName, node.Name)

×

1594

                                return true, false, nil

×

1595

×

1599

        return true, true, nil

×

1602

func (c *Controller) acquireAddress(pod *v1.Pod, podNet *kubeovnNet) (string, string, string, *kubeovnv1.Subnet, error) {

×

1603

        podName := c.getNameByPod(pod)

×

1604

        key := fmt.Sprintf("%s/%s", pod.Namespace, podName)

×

1605

×

1606

        var checkVMPod bool

×

1607

        isStsPod, _, _ := isStatefulSetPod(pod)

×

1608

        // if pod has static vip

×

1609

        vipName := pod.Annotations[util.VipAnnotation]

×

1610

        if vipName != "" {

×

1611

                vip, err := c.virtualIpsLister.Get(vipName)

×

1612

                if err != nil {

×

1613

                        klog.Errorf("failed to get static vip '%s', %v", vipName, err)

×

1614

                        return "", "", "", podNet.Subnet, err

×

1615

×

1616

                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1617

                if c.config.EnableKeepVMIP {

×

1618

                        checkVMPod, _ = isVMPod(pod)

×

1619

×

1620

                if err = c.podReuseVip(vipName, portName, isStsPod || checkVMPod); err != nil {

×

1621

                        return "", "", "", podNet.Subnet, err

×

1622

×

1623

                return vip.Status.V4ip, vip.Status.V6ip, vip.Status.Mac, podNet.Subnet, nil

×

1626

        var macPointer *string

×

1627

        if isOvnSubnet(podNet.Subnet) {

×

1628

                annoMAC := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]

×

1629

                if annoMAC != "" {

×

1630

                        if _, err := net.ParseMAC(annoMAC); err != nil {

×

1631

                                return "", "", "", podNet.Subnet, err

×

1632

×

1633

                        macPointer = &annoMAC

×

1635

        } else {

×

1636

                macPointer = ptr.To("")

×

1637

×

1639

        ippoolStr := pod.Annotations[fmt.Sprintf(util.IPPoolAnnotationTemplate, podNet.ProviderName)]

×

1640

        if ippoolStr == "" {

×

1641

                ns, err := c.namespacesLister.Get(pod.Namespace)

×

1642

                if err != nil {

×

1643

                        klog.Errorf("failed to get namespace %s: %v", pod.Namespace, err)

×

1644

                        return "", "", "", podNet.Subnet, err

×

1645

×

1647

                if len(ns.Annotations) != 0 {

×

NEW

1648

                        if ipPoolList, ok := ns.Annotations[util.IPPoolAnnotation]; ok {

×

NEW

1649

                                for _, ipPoolName := range strings.Split(ipPoolList, ",") {

×

NEW

1650

                                        ippool, err := c.ippoolLister.Get(ipPoolName)

×

NEW

1651

                                        if err != nil {

×

NEW

1652

                                                klog.Errorf("failed to get ippool %s: %v", ipPoolName, err)

×

NEW

1653

                                                return "", "", "", podNet.Subnet, err

×

NEW

1654

×

NEW

1656

                                        switch podNet.Subnet.Spec.Protocol {

×

NEW

1657

                                        case kubeovnv1.ProtocolDual:

×

NEW

1658

                                                if ippool.Status.V4AvailableIPs.Int64() == 0 || ippool.Status.V6AvailableIPs.Int64() == 0 {

×

NEW

1659

                                                        continue

×

NEW

1661

                                        case kubeovnv1.ProtocolIPv4:

×

NEW

1662

                                                if ippool.Status.V4AvailableIPs.Int64() == 0 {

×

NEW

1663

                                                        continue

×

NEW

1666

                                        default:

×

NEW

1667

                                                if ippool.Status.V6AvailableIPs.Int64() == 0 {

×

NEW

1668

                                                        continue

×

NEW

1672

                                        if ippool.Spec.Subnet == podNet.Subnet.Name {

×

NEW

1673

                                                ippoolStr = ippool.Name

×

NEW

1674

                                                break

×

1682

        if pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] == "" &&

×

1683

                ippoolStr == "" {

×

1684

                var skippedAddrs []string

×

1685

                for {

×

1686

                        portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1687

×

1688

                        ipv4, ipv6, mac, err := c.ipam.GetRandomAddress(key, portName, macPointer, podNet.Subnet.Name, "", skippedAddrs, !podNet.AllowLiveMigration)

×

1689

                        if err != nil {

×

1690

                                klog.Error(err)

×

1691

                                return "", "", "", podNet.Subnet, err

×

1692

×

1693

                        ipv4OK, ipv6OK, err := c.validatePodIP(pod.Name, podNet.Subnet.Name, ipv4, ipv6)

×

1694

                        if err != nil {

×

1695

                                klog.Error(err)

×

1696

                                return "", "", "", podNet.Subnet, err

×

1697

×

1698

                        if ipv4OK && ipv6OK {

×

1699

                                return ipv4, ipv6, mac, podNet.Subnet, nil

×

1700

×

1702

                        if !ipv4OK {

×

1703

                                skippedAddrs = append(skippedAddrs, ipv4)

×

1704

×

1705

                        if !ipv6OK {

×

1706

                                skippedAddrs = append(skippedAddrs, ipv6)

×

1707

×

1711

        portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1712

×

1713

        // The static ip can be assigned from any subnet after ns supports multi subnets

×

1714

        nsNets, _ := c.getNsAvailableSubnets(pod, podNet)

×

1715

        var v4IP, v6IP, mac string

×

1716

        var err error

×

1717

×

1718

        // Static allocate

×

1719

        if pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)] != "" {

×

1720

                ipStr := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]

×

1721

×

1722

                for _, net := range nsNets {

×

1723

                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipStr, macPointer, net.Subnet.Name, net.AllowLiveMigration)

×

1724

                        if err == nil {

×

1725

                                return v4IP, v6IP, mac, net.Subnet, nil

×

1726

×

1728

                return v4IP, v6IP, mac, podNet.Subnet, err

×

1732

        if ippoolStr != "" {

×

1733

                var ipPool []string

×

1734

                if strings.ContainsRune(ippoolStr, ';') {

×

1735

                        ipPool = strings.Split(ippoolStr, ";")

×

1736

                } else {

×

1737

                        ipPool = strings.Split(ippoolStr, ",")

×

1738

                        if len(ipPool) == 2 && util.CheckProtocol(ipPool[0]) != util.CheckProtocol(ipPool[1]) {

×

1739

                                ipPool = []string{ippoolStr}

×

1740

×

1742

                for i, ip := range ipPool {

×

1743

                        ipPool[i] = strings.TrimSpace(ip)

×

1744

×

1746

                if len(ipPool) == 1 && (!strings.ContainsRune(ipPool[0], ',') && net.ParseIP(ipPool[0]) == nil) {

×

1747

                        var skippedAddrs []string

×

1748

                        for {

×

1749

                                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

1750

                                ipv4, ipv6, mac, err := c.ipam.GetRandomAddress(key, portName, macPointer, podNet.Subnet.Name, ipPool[0], skippedAddrs, !podNet.AllowLiveMigration)

×

1751

                                if err != nil {

×

1752

                                        klog.Error(err)

×

1753

                                        return "", "", "", podNet.Subnet, err

×

1754

×

1755

                                ipv4OK, ipv6OK, err := c.validatePodIP(pod.Name, podNet.Subnet.Name, ipv4, ipv6)

×

1756

                                if err != nil {

×

1757

                                        klog.Error(err)

×

1758

                                        return "", "", "", podNet.Subnet, err

×

1759

×

1760

                                if ipv4OK && ipv6OK {

×

1761

                                        return ipv4, ipv6, mac, podNet.Subnet, nil

×

1762

×

1764

                                if !ipv4OK {

×

1765

                                        skippedAddrs = append(skippedAddrs, ipv4)

×

1766

×

1767

                                if !ipv6OK {

×

1768

                                        skippedAddrs = append(skippedAddrs, ipv6)

×

1769

×

1773

                if !isStsPod {

×

1774

                        for _, net := range nsNets {

×

1775

                                for _, staticIP := range ipPool {

×

1776

                                        var checkIP string

×

1777

                                        ipProtocol := util.CheckProtocol(staticIP)

×

1778

                                        if ipProtocol == kubeovnv1.ProtocolDual {

×

1779

                                                checkIP = strings.Split(staticIP, ",")[0]

×

1780

                                        } else {

×

1781

                                                checkIP = staticIP

×

1782

×

1784

                                        if assignedPod, ok := c.ipam.IsIPAssignedToOtherPod(checkIP, net.Subnet.Name, key); ok {

×

1785

                                                klog.Errorf("static address %s for %s has been assigned to %s", staticIP, key, assignedPod)

×

1786

                                                continue

×

1789

                                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, staticIP, macPointer, net.Subnet.Name, net.AllowLiveMigration)

×

1790

                                        if err == nil {

×

1791

                                                return v4IP, v6IP, mac, net.Subnet, nil

×

1792

×

1795

                        klog.Errorf("acquire address from ippool %s for %s failed, %v", ippoolStr, key, err)

×

1796

                } else {

×

1797

                        tempStrs := strings.Split(pod.Name, "-")

×

1798

                        numStr := tempStrs[len(tempStrs)-1]

×

1799

                        index, _ := strconv.Atoi(numStr)

×

1800

×

1801

                        if index < len(ipPool) {

×

1802

                                for _, net := range nsNets {

×

1803

                                        v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipPool[index], macPointer, net.Subnet.Name, net.AllowLiveMigration)

×

1804

                                        if err == nil {

×

1805

                                                return v4IP, v6IP, mac, net.Subnet, nil

×

1806

×

1808

                                klog.Errorf("acquire address %s for %s failed, %v", ipPool[index], key, err)

×

1812

        klog.Errorf("allocate address for %s failed, return NoAvailableAddress", key)

×

1813

        return "", "", "", podNet.Subnet, ipam.ErrNoAvailable

×

1816

func (c *Controller) acquireStaticAddress(key, nicName, ip string, mac *string, subnet string, liveMigration bool) (string, string, string, error) {

×

1817

        var v4IP, v6IP, macStr string

×

1818

        var err error

×

1819

        for _, ipStr := range strings.Split(ip, ",") {

×

1820

                if net.ParseIP(ipStr) == nil {

×

1821

                        return "", "", "", fmt.Errorf("failed to parse IP %s", ipStr)

×

1822

×

1825

        if v4IP, v6IP, macStr, err = c.ipam.GetStaticAddress(key, nicName, ip, mac, subnet, !liveMigration); err != nil {

×

1826

                klog.Errorf("failed to get static ip %v, mac %v, subnet %v, err %v", ip, mac, subnet, err)

×

1827

                return "", "", "", err

×

1828

×

1829

        return v4IP, v6IP, macStr, nil

×

1832

func appendCheckPodToDel(c *Controller, pod *v1.Pod, ownerRefName, ownerRefKind string) (bool, error) {

×

1833

        // subnet for ns has been changed, and statefulset pod's ip is not in the range of subnet's cidr anymore

×

1834

        podNs, err := c.namespacesLister.Get(pod.Namespace)

×

1835

        if err != nil {

×

1836

                klog.Errorf("failed to get namespace %s, %v", pod.Namespace, err)

×

1837

                return false, err

×

1838

×

1841

        var ownerRefSubnetExist bool

×

1842

        var ownerRefSubnet string

×

1843

        switch ownerRefKind {

×

1844

        case util.StatefulSet:

×

1845

                ss, err := c.config.KubeClient.AppsV1().StatefulSets(pod.Namespace).Get(context.Background(), ownerRefName, metav1.GetOptions{})

×

1846

                if err != nil {

×

1847

                        if k8serrors.IsNotFound(err) {

×

1848

                                klog.Infof("Statefulset %s is not found", ownerRefName)

×

1849

                                return true, nil

×

1850

×

1851

                        klog.Errorf("failed to get StatefulSet %s, %v", ownerRefName, err)

×

1853

                if ss.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation] != "" {

×

1854

                        ownerRefSubnetExist = true

×

1855

                        ownerRefSubnet = ss.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation]

×

1856

×

1858

        case util.VMInstance:

×

1859

                vm, err := c.config.KubevirtClient.VirtualMachine(pod.Namespace).Get(context.Background(), ownerRefName, metav1.GetOptions{})

×

1860

                if err != nil {

×

1861

                        if k8serrors.IsNotFound(err) {

×

1862

                                klog.Infof("VirtualMachine %s is not found", ownerRefName)

×

1863

                                return true, nil

×

1864

×

1865

                        klog.Errorf("failed to get VirtualMachine %s, %v", ownerRefName, err)

×

1867

                if vm != nil &&

×

1868

                        vm.Spec.Template != nil &&

×

1869

                        vm.Spec.Template.ObjectMeta.Annotations != nil &&

×

1870

                        vm.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation] != "" {

×

1871

                        ownerRefSubnetExist = true

×

1872

                        ownerRefSubnet = vm.Spec.Template.ObjectMeta.Annotations[util.LogicalSwitchAnnotation]

×

1873

×

1875

        podSwitch := strings.TrimSpace(pod.Annotations[util.LogicalSwitchAnnotation])

×

1876

        if !ownerRefSubnetExist {

×

1877

                nsSubnetNames := podNs.Annotations[util.LogicalSwitchAnnotation]

×

1878

                // check if pod use the subnet of its ns

×

1879

                if nsSubnetNames != "" && podSwitch != "" && !slices.Contains(strings.Split(nsSubnetNames, ","), podSwitch) {

×

1880

                        klog.Infof("ns %s annotation subnet is %s, which is inconstant with subnet for pod %s, delete pod", pod.Namespace, nsSubnetNames, pod.Name)

×

1881

                        return true, nil

×

1882

×

1886

        podSubnet, err := c.subnetsLister.Get(podSwitch)

×

1887

        if err != nil {

×

1888

                klog.Errorf("failed to get subnet %s, %v, not auto clean ip", podSwitch, err)

×

1889

                return false, err

×

1890

×

1891

        if podSubnet == nil {

×

1892

                // TODO: remove: CRD get interface will retrun a nil subnet ?

×

1893

                klog.Errorf("pod %s/%s subnet %s is nil, not auto clean ip", pod.Namespace, pod.Name, podSwitch)

×

1894

                return false, nil

×

1895

×

1896

        podIP := pod.Annotations[util.IPAddressAnnotation]

×

1897

        if podIP == "" {

×

1898

                // delete pod just after it created < 1ms

×

1899

                klog.Infof("pod %s/%s annotaions has no ip address, not auto clean ip", pod.Namespace, pod.Name)

×

1900

                return false, nil

×

1901

×

1902

        podSubnetCidr := podSubnet.Spec.CIDRBlock

×

1903

        if podSubnetCidr != "" {

×

1904

                // subnet spec cidr changed by user

×

1905

                klog.Errorf("invalid pod subnet %s empty cidr %s, not auto clean ip", podSwitch, podSubnetCidr)

×

1906

                return false, nil

×

1907

×

1908

        if !util.CIDRContainIP(podSubnetCidr, podIP) {

×

1909

                klog.Infof("pod's ip %s is not in the range of subnet %s, delete pod", pod.Annotations[util.IPAddressAnnotation], podSubnet.Name)

×

1910

                return true, nil

×

1911

×

1913

        if ownerRefSubnet != "" && podSubnet.Name != ownerRefSubnet {

×

1914

                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)

×

1915

                return true, nil

×

1916

×

1918

        return false, nil

×

1921

func isVMPod(pod *v1.Pod) (bool, string) {

×

1922

        for _, owner := range pod.OwnerReferences {

×

1923

                // The name of vmi is consistent with vm's name.

×

1924

                if owner.Kind == util.VMInstance && strings.HasPrefix(owner.APIVersion, "kubevirt.io") {

×

1925

                        return true, owner.Name

×

1926

×

1928

        return false, ""

×

1931

func isOwnsByTheVM(vmi metav1.Object) (bool, string) {

×

1932

        for _, owner := range vmi.GetOwnerReferences() {

×

1933

                if owner.Kind == util.VM && strings.HasPrefix(owner.APIVersion, "kubevirt.io") {

×

1934

                        return true, owner.Name

×

1935

×

1937

        return false, ""

×

1940

func (c *Controller) isVMToDel(pod *v1.Pod, vmiName string) bool {

×

1941

        var (

×

1942

                vmiAlive bool

×

1943

                vmName   string

×

1944

×

1945

        // The vmi is also deleted when pod is deleted, only left vm exists.

×

1946

        vmi, err := c.config.KubevirtClient.VirtualMachineInstance(pod.Namespace).Get(context.Background(), vmiName, metav1.GetOptions{})

×

1947

        if err != nil {

×

1948

                if k8serrors.IsNotFound(err) {

×

1949

                        vmiAlive = false

×

1950

                        // The name of vmi is consistent with vm's name.

×

1951

                        vmName = vmiName

×

1952

                        klog.ErrorS(err, "failed to get vmi, will try to get the vm directly", "name", vmiName)

×

1953

                } else {

×

1954

                        klog.ErrorS(err, "failed to get vmi", "name", vmiName)

×

1955

                        return false

×

1956

×

1957

        } else {

×

1958

                var ownsByVM bool

×

1959

                ownsByVM, vmName = isOwnsByTheVM(vmi)

×

1960

                if !ownsByVM && !vmi.DeletionTimestamp.IsZero() {

×

1961

                        klog.Infof("ephemeral vmi %s is deleting", vmiName)

×

1962

                        return true

×

1963

×

1964

                vmiAlive = vmi.DeletionTimestamp.IsZero()

×

1967

        if vmiAlive {

×

1968

                return false

×

1969

×

1971

        vm, err := c.config.KubevirtClient.VirtualMachine(pod.Namespace).Get(context.Background(), vmName, metav1.GetOptions{})

×

1972

        if err != nil {

×

1973

                // the vm has gone

×

1974

                if k8serrors.IsNotFound(err) {

×

1975

                        klog.ErrorS(err, "failed to get vm", "name", vmName)

×

1976

                        return true

×

1977

×

1978

                klog.ErrorS(err, "failed to get vm", "name", vmName)

×

1979

                return false

×

1982

        if !vm.DeletionTimestamp.IsZero() {

×

1983

                klog.Infof("vm %s is deleting", vmName)

×

1984

                return true

×

1985

×

1986

        return false

×

1989

func (c *Controller) getNameByPod(pod *v1.Pod) string {

×

1990

        if c.config.EnableKeepVMIP {

×

1991

                if isVMPod, vmName := isVMPod(pod); isVMPod {

×

1992

                        return vmName

×

1993

×

1995

        return pod.Name

×

1999

func (c *Controller) getNsAvailableSubnets(pod *v1.Pod, podNet *kubeovnNet) ([]*kubeovnNet, error) {

×

2000

        var result []*kubeovnNet

×

2001

        // keep the annotation subnet of the pod in first position

×

2002

        result = append(result, podNet)

×

2003

×

2004

        ns, err := c.namespacesLister.Get(pod.Namespace)

×

2005

        if err != nil {

×

2006

                klog.Errorf("failed to get namespace %s, %v", pod.Namespace, err)

×

2007

                return nil, err

×

2008

×

2009

        if ns.Annotations == nil {

×

2010

                return nil, nil

×

2011

×

2013

        subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]

×

2014

        for _, subnetName := range strings.Split(subnetNames, ",") {

×

2015

                if subnetName == "" || subnetName == podNet.Subnet.Name {

×

2016

                        continue

×

2018

                subnet, err := c.subnetsLister.Get(subnetName)

×

2019

                if err != nil {

×

2020

                        klog.Errorf("failed to get subnet %v", err)

×

2021

                        return nil, err

×

2022

×

2024

                result = append(result, &kubeovnNet{

×

2025

                        Type:         providerTypeOriginal,

×

2026

                        ProviderName: subnet.Spec.Provider,

×

2027

                        Subnet:       subnet,

×

2028

})

×

2031

        return result, nil

×

2034

func getPodType(pod *v1.Pod) string {

×

2035

        if ok, _, _ := isStatefulSetPod(pod); ok {

×

2036

                return util.StatefulSet

×

2037

×

2039

        if isVMPod, _ := isVMPod(pod); isVMPod {

×

2040

                return util.VM

×

2041

×

2042

        return ""

×

2045

func (c *Controller) getVirtualIPs(pod *v1.Pod, podNets []*kubeovnNet) map[string]string {

×

2046

        vipsListMap := make(map[string][]string)

×

2047

        var vipNamesList []string

×

2048

        for _, vipName := range strings.Split(strings.TrimSpace(pod.Annotations[util.AAPsAnnotation]), ",") {

×

2049

                if vipName = strings.TrimSpace(vipName); vipName == "" {

×

2050

                        continue

×

2052

                if !slices.Contains(vipNamesList, vipName) {

×

2053

                        vipNamesList = append(vipNamesList, vipName)

×

2054

                } else {

×

2055

                        continue

×

2057

                vip, err := c.virtualIpsLister.Get(vipName)

×

2058

                if err != nil {

×

2059

                        klog.Errorf("failed to get vip %s, %v", vipName, err)

×

2060

                        continue

×

2062

                if vip.Spec.Namespace != pod.Namespace || (vip.Status.V4ip == "" && vip.Status.V6ip == "") {

×

2063

                        continue

×

2065

                for _, podNet := range podNets {

×

2066

                        if podNet.Subnet.Name == vip.Spec.Subnet {

×

2067

                                key := fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)

×

2068

                                vipsList := vipsListMap[key]

×

2069

                                if vipsList == nil {

×

2070

                                        vipsList = []string{}

×

2071

×

2073

                                if util.IsValidIP(vip.Status.V4ip) {

×

2074

                                        vipsList = append(vipsList, vip.Status.V4ip)

×

2075

×

2076

                                if util.IsValidIP(vip.Status.V6ip) {

×

2077

                                        vipsList = append(vipsList, vip.Status.V6ip)

×

2078

×

2080

                                vipsListMap[key] = vipsList

×

2085

        for _, podNet := range podNets {

×

2086

                vipStr := pod.Annotations[fmt.Sprintf(util.PortVipAnnotationTemplate, podNet.ProviderName)]

×

2087

                if vipStr == "" {

×

2088

                        continue

×

2090

                key := fmt.Sprintf("%s.%s", podNet.Subnet.Name, podNet.ProviderName)

×

2091

                vipsList := vipsListMap[key]

×

2092

                if vipsList == nil {

×

2093

                        vipsList = []string{}

×

2094

×

2096

                for _, vip := range strings.Split(vipStr, ",") {

×

2097

                        if util.IsValidIP(vip) && !slices.Contains(vipsList, vip) {

×

2098

                                vipsList = append(vipsList, vip)

×

2099

×

2102

                vipsListMap[key] = vipsList

×

2105

        vipsMap := make(map[string]string)

×

2106

        for key, vipsList := range vipsListMap {

×

2107

                vipsMap[key] = strings.Join(vipsList, ",")

×

2108

×

2109

        return vipsMap

×

2112

func setPodRoutesAnnotation(annotations map[string]string, provider string, routes []request.Route) error {

×

2113

        key := fmt.Sprintf(util.RoutesAnnotationTemplate, provider)

×

2114

        if len(routes) == 0 {

×

2115

                delete(annotations, key)

×

2116

                return nil

×

2117

×

2119

        buf, err := json.Marshal(routes)

×

2120

        if err != nil {

×

2121

                err = fmt.Errorf("failed to marshal routes %+v: %w", routes, err)

×

2122

                klog.Error(err)

×

2123

                return err

×

2124

×

2125

        annotations[key] = string(buf)

×

2126

×

2127

        return nil

×

kubeovn / kube-ovn / 12545917939

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

kubeovn / kube-ovn / 12545917939

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

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