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

44

func NewNamedPort() *NamedPort {

×

45

        return &NamedPort{

×

46

                mutex:        sync.RWMutex{},

×

47

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

×

48

×

49

×

51

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

×

52

        n.mutex.Lock()

×

53

        defer n.mutex.Unlock()

×

54

        ns := pod.Namespace

×

55

        podName := pod.Name

×

56

×

57

        if pod.Spec.Containers == nil {

×

58

                return

×

59

×

61

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

×

62

                if container.Ports == nil {

×

63

                        continue

×

66

                for _, port := range container.Ports {

×

67

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

×

68

                                continue

×

71

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

×

72

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

×

73

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

×

74

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

×

75

                                        } else {

×

76

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

×

77

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

×

78

×

79

                                        continue

×

81

                        } else {

×

82

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

×

83

×

84

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

×

85

                                PortID: port.ContainerPort,

×

86

                                Pods:   strset.New(podName),

×

87

×

92

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

×

93

        n.mutex.Lock()

×

94

        defer n.mutex.Unlock()

×

95

×

96

        ns := pod.Namespace

×

97

        podName := pod.Name

×

98

×

99

        if pod.Spec.Containers == nil {

×

100

                return

×

101

×

103

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

×

104

                if container.Ports == nil {

×

105

                        continue

×

108

                for _, port := range container.Ports {

×

109

                        if port.Name == "" {

×

110

                                continue

×

113

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

×

114

                                continue

×

117

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

×

118

                                continue

×

121

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

×

122

                                continue

×

125

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

×

126

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

×

127

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

×

128

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

×

129

                                        delete(n.namedPortMap, ns)

×

130

×

136

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

×

137

        n.mutex.RLock()

×

138

        defer n.mutex.RUnlock()

×

139

×

140

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

×

141

                for portName, info := range result {

×

142

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

×

143

×

144

                return result

×

146

        return nil

×

149

func isPodAlive(p *v1.Pod) bool {

×

150

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

×

151

                now := time.Now()

×

152

                deletionTime := p.DeletionTimestamp.Time

×

153

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

×

154

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

×

155

                        return false

×

156

×

158

        return isPodStatusPhaseAlive(p)

×

161

func isPodStatusPhaseAlive(p *v1.Pod) bool {

×

162

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

×

163

                return false

×

164

×

166

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

×

167

                return false

×

168

×

170

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

×

171

                return false

×

172

×

173

        return true

×

176

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

×

177

        var key string

×

178

        var err error

×

179

        if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {

×

180

                utilruntime.HandleError(err)

×

181

                return

×

182

×

184

        p := obj.(*v1.Pod)

×

185

        if p.Spec.HostNetwork {

×

186

                return

×

187

×

190

        if c.config.EnableNP {

×

191

                c.namedPort.AddNamedPortByPod(p)

×

192

                if p.Status.PodIP != "" {

×

193

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

×

194

                                c.updateNpQueue.Add(np)

×

195

×

199

        if !isPodAlive(p) {

×

200

                isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(p)

×

201

                isVMPod, vmName := isVMPod(p)

×

202

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

×

203

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

×

204

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

×

205

                                c.deletingPodObjMap.Store(key, p)

×

206

                                c.deletePodQueue.Add(key)

×

207

×

208

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

×

209

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

×

210

                                c.deletingPodObjMap.Store(key, p)

×

211

                                c.deletePodQueue.Add(key)

×

212

×

213

                } else {

×

214

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

×

215

                        c.deletingPodObjMap.Store(key, p)

×

216

                        c.deletePodQueue.Add(key)

×

217

×

218

                return

×

221

        need, err := c.podNeedSync(p)

×

222

        if err != nil {

×

223

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

×

224

                return

×

225

×

226

        if need {

×

227

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

×

228

                c.addOrUpdatePodQueue.Add(key)

×

229

×

232

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

×

233

        var key string

×

234

        var err error

×

235

        if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {

×

236

                utilruntime.HandleError(err)

×

237

                return

×

238

×

240

        p := obj.(*v1.Pod)

×

241

        if p.Spec.HostNetwork {

×

242

                return

×

243

×

245

        if c.config.EnableNP {

×

246

                c.namedPort.DeleteNamedPortByPod(p)

×

247

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

×

248

                        c.updateNpQueue.Add(np)

×

249

×

252

        if c.config.EnableANP {

×

253

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

×

254

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

×

255

×

257

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

×

258

        c.deletingPodObjMap.Store(key, p)

×

259

        c.deletePodQueue.Add(key)

×

262

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

×

263

        oldPod := oldObj.(*v1.Pod)

×

264

        newPod := newObj.(*v1.Pod)

×

265

×

266

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

×

267

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

×

268

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

×

269

                var vipNames []string

×

270

                for _, vipName := range oldAAPs {

×

271

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

×

272

×

273

                for _, vipName := range newAAPs {

×

274

                        vipName = strings.TrimSpace(vipName)

×

275

                        if !slices.Contains(vipNames, vipName) {

×

276

                                vipNames = append(vipNames, vipName)

×

277

×

279

                for _, vipName := range vipNames {

×

280

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

×

281

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

×

282

                                        continue

×

284

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

×

285

                                c.updateVirtualParentsQueue.Add(vipName)

×

290

        if oldPod.ResourceVersion == newPod.ResourceVersion {

×

291

                return

×

292

×

293

        if newPod.Spec.HostNetwork {

×

294

                return

×

295

×

297

        key, err := cache.MetaNamespaceKeyFunc(newObj)

×

298

        if err != nil {

×

299

                utilruntime.HandleError(err)

×

300

                return

×

301

×

303

        podNets, err := c.getPodKubeovnNets(newPod)

×

304

        if err != nil {

×

305

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

×

306

                return

×

307

×

309

        if c.config.EnableNP {

×

310

                c.namedPort.AddNamedPortByPod(newPod)

×

311

                newNp := c.podMatchNetworkPolicies(newPod)

×

312

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

×

313

                        oldNp := c.podMatchNetworkPolicies(oldPod)

×

314

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

×

315

                                c.updateNpQueue.Add(np)

×

316

×

319

                for _, podNet := range podNets {

×

320

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

×

321

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

×

322

                        if oldAllocated != newAllocated {

×

323

                                for _, np := range newNp {

×

324

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

×

325

                                        c.updateNpQueue.Add(np)

×

326

×

327

                                break

×

332

        if c.config.EnableANP {

×

333

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

×

334

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

×

335

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

×

336

×

338

                for _, podNet := range podNets {

×

339

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

×

340

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

×

341

                        if oldAllocated != newAllocated {

×

342

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

×

343

                                break

×

348

        isStateful, statefulSetName, statefulSetUID := isStatefulSetPod(newPod)

×

349

        isVMPod, vmName := isVMPod(newPod)

×

350

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

×

351

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

×

352

                c.deletingPodObjMap.Store(key, newPod)

×

353

                c.deletePodQueue.Add(key)

×

354

                return

×

355

×

358

        var delay time.Duration

×

359

        if newPod.Spec.TerminationGracePeriodSeconds != nil {

×

360

                if !newPod.DeletionTimestamp.IsZero() {

×

361

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

×

362

                } else {

×

363

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

×

364

×

367

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

×

368

                go func() {

×

369

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

×

370

                        // the ip address will not be recycled

×

371

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

×

372

                        c.deletingPodObjMap.Store(key, newPod)

×

373

                        c.deletePodQueue.AddAfter(key, delay)

×

374

}()

×

375

                return

×

379

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

×

380

                go func() {

×

381

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

×

382

                        c.deletingPodObjMap.Store(key, newPod)

×

383

                        c.deletePodQueue.AddAfter(key, delay)

×

384

}()

×

385

                return

×

387

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

×

388

                go func() {

×

389

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

×

390

                        c.deletingPodObjMap.Store(key, newPod)

×

391

                        c.deletePodQueue.AddAfter(key, delay)

×

392

}()

×

393

                return

×

395

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

×

396

        c.addOrUpdatePodQueue.Add(key)

×

397

×

398

        // security policy changed

×

399

        for _, podNet := range podNets {

×

400

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

×

401

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

×

402

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

×

403

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

×

404

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

×

405

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

×

406

                oldAAPs := oldPod.Annotations[util.AAPsAnnotation]

×

407

                newAAPs := newPod.Annotations[util.AAPsAnnotation]

×

408

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

×

409

                        c.updatePodSecurityQueue.Add(key)

×

410

                        break

×

415

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

×

416

        defaultSubnet, err := c.getPodDefaultSubnet(pod)

×

417

        if err != nil {

×

418

                klog.Error(err)

×

419

                return nil, err

×

420

×

422

        attachmentNets, err := c.getPodAttachmentNet(pod)

×

423

        if err != nil {

×

424

                klog.Error(err)

×

425

                return nil, err

×

426

×

428

        podNets := attachmentNets

×

429

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

×

430

                podNets = append(attachmentNets, &kubeovnNet{

×

431

                        Type:         providerTypeOriginal,

×

432

                        ProviderName: util.OvnProvider,

×

433

                        Subnet:       defaultSubnet,

×

434

                        IsDefault:    true,

×

435

})

×

436

×

438

        return podNets, nil

×

441

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

×

442

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

×

443

        if err != nil {

×

444

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

×

445

                return nil

×

446

×

448

        c.podKeyMutex.LockKey(key)

×

449

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

×

450

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

×

451

×

452

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

×

453

        if err != nil {

×

454

                if k8serrors.IsNotFound(err) {

×

455

                        return nil

×

456

×

457

                klog.Error(err)

×

458

                return err

×

460

        pod := cachedPod.DeepCopy()

×

461

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

×

462

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

×

463

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

×

464

                return err

×

465

×

467

        podNets, err := c.getPodKubeovnNets(pod)

×

468

        if err != nil {

×

469

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

×

470

                return err

×

471

×

472

        if len(pod.Annotations) == 0 {

×

473

                pod.Annotations = map[string]string{}

×

474

×

477

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

×

478

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

×

479

                return err

×

480

×

481

        if cachedPod == nil {

×

482

                // pod has been deleted

×

483

                return nil

×

484

×

485

        pod = cachedPod.DeepCopy()

×

486

        needAllocatePodNets := needAllocateSubnets(pod, podNets)

×

487

        if len(needAllocatePodNets) != 0 {

×

488

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

×

489

                        klog.Error(err)

×

490

                        return err

×

491

×

492

                if cachedPod == nil {

×

493

                        // pod has been deleted

×

494

                        return nil

×

495

×

499

        pod = cachedPod.DeepCopy()

×

500

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

×

504

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

×

505

        namespace := pod.Namespace

×

506

        name := pod.Name

×

507

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

×

508

×

509

        vipsMap := c.getVirtualIPs(pod, needAllocatePodNets)

×

510

        isVMPod, vmName := isVMPod(pod)

×

511

        podType := getPodType(pod)

×

512

        podName := c.getNameByPod(pod)

×

513

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

×

514

×

515

        var err error

×

516

        var vmKey string

×

517

        if isVMPod && c.config.EnableKeepVMIP {

×

518

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

×

519

×

521

        for _, podNet := range needAllocatePodNets {

×

522

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

×

523

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

×

524

                if err != nil {

×

525

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

×

526

                        klog.Error(err)

×

527

                        return nil, err

×

528

×

529

                ipStr := util.GetStringIP(v4IP, v6IP)

×

530

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

×

531

                if mac == "" {

×

532

                        delete(pod.Annotations, fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName))

×

533

                } else {

×

534

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

×

535

×

536

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

×

537

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

×

538

                if isOvnSubnet(podNet.Subnet) {

×

539

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

×

540

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

×

541

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

×

542

×

543

                } else {

×

544

                        delete(pod.Annotations, fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, podNet.ProviderName))

×

545

                        delete(pod.Annotations, fmt.Sprintf(util.PodNicAnnotationTemplate, podNet.ProviderName))

×

546

×

547

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

×

548

                if vmKey != "" {

×

549

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

×

550

×

551

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

×

552

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

×

553

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

×

554

                        return nil, err

×

555

×

557

                if podNet.Type != providerTypeIPAM {

×

558

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

×

559

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

×

560

×

562

                        if subnet.Spec.Vlan != "" {

×

563

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

×

564

                                if err != nil {

×

565

                                        klog.Error(err)

×

566

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

×

567

                                        return nil, err

×

568

×

569

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

×

570

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

×

573

                        portSecurity := false

×

574

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

×

575

                                portSecurity = true

×

576

×

578

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

×

579

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

×

580

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

×

581

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

×

582

                                        vips = ""

×

583

                                        break

×

587

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

×

588

                        dhcpOptions := &ovs.DHCPOptionsUUIDs{

×

589

                                DHCPv4OptionsUUID: subnet.Status.DHCPv4OptionsUUID,

×

590

                                DHCPv6OptionsUUID: subnet.Status.DHCPv6OptionsUUID,

×

591

×

592

×

593

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

×

594

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

×

595

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

×

596

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

×

597

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

×

598

                                klog.Errorf("%v", err)

×

599

                                return nil, err

×

600

×

602

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

×

603

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

×

604

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

×

605

                                        klog.Errorf("%v", err)

×

606

                                        return nil, err

×

607

×

610

                        if securityGroupAnnotation != "" {

×

611

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

×

612

                                for _, sgName := range sgNames {

×

613

                                        if sgName != "" {

×

614

                                                c.syncSgPortsQueue.Add(sgName)

×

615

×

619

                        if vips != "" {

×

620

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

×

621

×

624

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

×

625

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

×

626

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

×

627

                        klog.Error(err)

×

628

                        return nil, err

×

629

×

631

        patch, err := util.GenerateMergePatchPayload(cachedPod, pod)

×

632

        if err != nil {

×

633

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

×

634

                return nil, err

×

635

×

636

        patchedPod, err := c.config.KubeClient.CoreV1().Pods(namespace).Patch(context.Background(), name,

×

637

                types.MergePatchType, patch, metav1.PatchOptions{}, "")

×

638

        if err != nil {

×

639

                if k8serrors.IsNotFound(err) {

×

640

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

×

641

                        // Then we need to recycle the resource again.

×

642

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

×

643

                        c.deletingPodObjMap.Store(key, pod)

×

644

                        c.deletePodQueue.AddRateLimited(key)

×

645

                        return nil, nil

×

646

×

647

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

×

648

                return nil, err

×

651

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

×

652

                c.initVpcNatGatewayQueue.Add(vpcGwName)

×

653

×

654

        return patchedPod.DeepCopy(), nil

×

658

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

×

659

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

×

660

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

×

661

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

×

662

×

664

        if len(needRoutePodNets) == 0 {

×

665

                return nil

×

666

×

668

        namespace := pod.Namespace

×

669

        name := pod.Name

×

670

        podName := c.getNameByPod(pod)

×

671

×

672

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

×

673

×

674

        var podIP string

×

675

        var subnet *kubeovnv1.Subnet

×

676

×

677

        for _, podNet := range needRoutePodNets {

×

678

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

×

679

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

×

680

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

×

681

×

683

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

×

684

                subnet = podNet.Subnet

×

685

×

686

                // Check if pod uses nodeSwitch subnet

×

687

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

×

688

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

×

689

×

691

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

×

692

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

×

693

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

×

694

                        subnet.Spec.U2OInterconnection &&

×

695

                        subnet.Spec.Vlan != "" &&

×

696

                        !subnet.Spec.LogicalGateway {

×

697

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

×

698

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

×

699

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

×

700

                                return err

×

701

×

704

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

×

705

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

×

706

                        if err != nil {

×

707

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

×

708

                                return err

×

709

×

711

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

×

712

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

×

713

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

×

714

                                if err != nil {

×

715

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

×

716

                                        return err

×

717

×

718

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

×

719

                                if nextHop == "" {

×

720

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

×

721

                                        if err != nil {

×

722

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

×

723

                                                return err

×

724

×

725

                                        nextHop = externalSubnet.Spec.Gateway

×

726

                                        if nextHop == "" {

×

727

                                                klog.Errorf("no available gateway address")

×

728

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

×

729

×

731

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

×

732

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

×

733

×

735

                                if err := c.addPolicyRouteToVpc(

×

736

                                        subnet.Spec.Vpc,

×

737

                                        &kubeovnv1.PolicyRoute{

×

738

                                                Priority:  util.NorthGatewayRoutePolicyPriority,

×

739

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

×

740

                                                Action:    kubeovnv1.PolicyRouteActionReroute,

×

741

                                                NextHopIP: nextHop,

×

742

},

×

743

                                        map[string]string{

×

744

                                                "vendor": util.CniTypeName,

×

745

                                                "subnet": subnet.Name,

×

746

},

×

747

                                ); err != nil {

×

748

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

×

749

                                        return err

×

750

×

753

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

×

754

                                        klog.Error(err)

×

755

                                        return err

×

756

×

757

                        } else {

×

758

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

×

759

                                        nodeTunlIPAddr, err := getNodeTunlIP(node)

×

760

                                        if err != nil {

×

761

                                                klog.Error(err)

×

762

                                                return err

×

763

×

765

                                        var added bool

×

766

×

767

                                        for _, nodeAddr := range nodeTunlIPAddr {

×

768

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

×

769

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

×

770

                                                                continue

×

773

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

×

774

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

×

775

                                                                return err

×

776

×

778

                                                        added = true

×

779

                                                        break

×

781

                                                if added {

×

782

                                                        break

×

787

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

×

788

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

×

789

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

×

790

                                                        continue

×

792

                                                ipSuffix := "ip4"

×

793

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

×

794

                                                        ipSuffix = "ip6"

×

795

×

797

                                                if err := c.addPolicyRouteToVpc(

×

798

                                                        subnet.Spec.Vpc,

×

799

                                                        &kubeovnv1.PolicyRoute{

×

800

                                                                Priority:  util.NorthGatewayRoutePolicyPriority,

×

801

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

×

802

                                                                Action:    kubeovnv1.PolicyRouteActionReroute,

×

803

                                                                NextHopIP: pod.Annotations[util.NorthGatewayAnnotation],

×

804

},

×

805

                                                        map[string]string{

×

806

                                                                "vendor": util.CniTypeName,

×

807

                                                                "subnet": subnet.Name,

×

808

},

×

809

                                                ); err != nil {

×

810

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

×

811

                                                        return err

×

812

×

814

                                } else if c.config.EnableEipSnat {

×

815

                                        if err = c.deleteStaticRouteFromVpc(

×

816

                                                c.config.ClusterRouter,

×

817

                                                subnet.Spec.RouteTable,

×

818

                                                podIP,

×

819

"",

×

820

                                                kubeovnv1.PolicyDst,

×

821

                                        ); err != nil {

×

822

                                                klog.Error(err)

×

823

                                                return err

×

824

×

828

                        if c.config.EnableEipSnat {

×

829

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

×

830

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

×

831

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

×

832

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

×

833

×

834

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

×

835

                                                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 {

×

836

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

×

837

                                                        return err

×

838

×

840

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

×

841

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

×

842

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

×

843

×

844

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

×

845

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

×

846

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

×

847

                                                        return err

×

848

×

854

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

×

855

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

×

856

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

×

857

                                return err

×

858

×

861

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

×

863

        patch, err := util.GenerateMergePatchPayload(cachedPod, pod)

×

864

        if err != nil {

×

865

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

×

866

                return err

×

867

×

868

        if _, err := c.config.KubeClient.CoreV1().Pods(namespace).Patch(context.Background(), name,

×

869

                types.MergePatchType, patch, metav1.PatchOptions{}, ""); err != nil {

×

870

                if k8serrors.IsNotFound(err) {

×

871

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

×

872

                        // Then we need to recycle the resource again.

×

873

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

×

874

                        c.deletingPodObjMap.Store(key, pod)

×

875

                        c.deletePodQueue.AddRateLimited(key)

×

876

                        return nil

×

877

×

878

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

×

879

                return err

×

881

        return nil

×

884

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

×

885

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

×

886

        if !ok {

×

887

                return nil

×

888

×

889

        podName := c.getNameByPod(pod)

×

890

        c.podKeyMutex.LockKey(key)

×

891

        defer func() {

×

892

                _ = c.podKeyMutex.UnlockKey(key)

×

893

                if err == nil {

×

894

                        c.deletingPodObjMap.Delete(key)

×

895

×

897

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

×

898

×

899

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

×

900

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

×

901

                // Pod with same name exists, just return here

×

902

                return nil

×

903

×

905

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

×

906

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

×

907

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

×

908

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

×

909

                                        continue

×

911

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

×

912

                                c.updateVirtualParentsQueue.Add(vipName)

×

917

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

×

918

×

919

        var keepIPCR bool

×

920

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

×

921

                if pod.DeletionTimestamp != nil {

×

922

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

×

923

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

×

924

                        if !toDel {

×

925

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

×

926

                                keepIPCR = true

×

927

×

929

                if keepIPCR {

×

930

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

×

931

                        if err != nil {

×

932

                                klog.Error(err)

×

933

                                return err

×

934

×

935

                        if isDelete {

×

936

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

×

937

                                keepIPCR = false

×

938

×

941

        isVMPod, vmName := isVMPod(pod)

×

942

        if isVMPod && c.config.EnableKeepVMIP {

×

943

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

×

944

                if err != nil {

×

945

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

×

946

                        return err

×

947

×

948

                for _, port := range ports {

×

949

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

×

950

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

×

951

                                klog.Error(err)

×

952

                                return err

×

953

×

955

                if pod.DeletionTimestamp != nil {

×

956

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

×

957

                        vmToBeDel := c.isVMToDel(pod, vmName)

×

958

                        if !vmToBeDel {

×

959

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

×

960

                                keepIPCR = true

×

961

×

963

                if keepIPCR {

×

964

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

×

965

                        if err != nil {

×

966

                                klog.Error(err)

×

967

                                return err

×

968

×

969

                        if isDelete {

×

970

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

×

971

                                keepIPCR = false

×

972

×

976

        podNets, err := c.getPodKubeovnNets(pod)

×

977

        if err != nil {

×

978

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

×

979

×

980

        if !keepIPCR {

×

981

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

×

982

                if err != nil {

×

983

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

×

984

                        return err

×

985

×

987

                if len(ports) != 0 {

×

988

                        addresses := c.ipam.GetPodAddress(podKey)

×

989

                        for _, address := range addresses {

×

990

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

×

991

                                        continue

×

993

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

×

994

                                if k8serrors.IsNotFound(err) {

×

995

                                        continue

×

996

                                } else if err != nil {

×

997

                                        klog.Error(err)

×

998

                                        return err

×

999

×

1000

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

×

1001

                                if k8serrors.IsNotFound(err) {

×

1002

                                        continue

×

1003

                                } else if err != nil {

×

1004

                                        klog.Error(err)

×

1005

                                        return err

×

1006

×

1008

                                ipSuffix := "ip4"

×

1009

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

×

1010

                                        ipSuffix = "ip6"

×

1011

×

1012

                                if err = c.deletePolicyRouteFromVpc(

×

1013

                                        vpc.Name,

×

1014

                                        util.NorthGatewayRoutePolicyPriority,

×

1015

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

×

1016

                                ); err != nil {

×

1017

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

×

1018

                                        return err

×

1019

×

1021

                                if c.config.EnableEipSnat {

×

1022

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

×

1023

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

×

1024

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

×

1025

×

1027

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

×

1028

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

×

1029

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

×

1030

×

1035

                for _, port := range ports {

×

1036

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

×

1037

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

×

1038

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

×

1039

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

×

1040

                                return err

×

1041

×

1043

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

×

1044

                for _, podNet := range podNets {

×

1045

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

×

1046

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

×

1047

                        if err != nil {

×

1048

                                if k8serrors.IsNotFound(err) {

×

1049

                                        continue

×

1051

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

×

1052

                                return err

×

1054

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

×

1055

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

×

1056

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

×

1057

                                        if !k8serrors.IsNotFound(err) {

×

1058

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

×

1059

                                                return err

×

1060

×

1063

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

×

1066

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

×

1067

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

×

1068

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

×

1069

                                return err

×

1070

×

1073

        for _, podNet := range podNets {

×

1074

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

×

1075

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

×

1076

                if securityGroupAnnotation != "" {

×

1077

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

×

1078

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

×

1079

                                if sgName != "" {

×

1080

                                        c.syncSgPortsQueue.Add(sgName)

×

1081

×

1085

        return nil

×

1088

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

×

1089

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

×

1090

        if err != nil {

×

1091

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

×

1092

                return nil

×

1093

×

1095

        c.podKeyMutex.LockKey(key)

×

1096

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

×

1098

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

×

1099

        if err != nil {

×

1100

                if k8serrors.IsNotFound(err) {

×

1101

                        return nil

×

1102

×

1103

                klog.Error(err)

×

1104

                return err

×

1106

        podName := c.getNameByPod(pod)

×

1107

×

1108

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

×

1109

×

1110

        podNets, err := c.getPodKubeovnNets(pod)

×

1111

        if err != nil {

×

1112

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

×

1113

                return err

×

1114

×

1116

        vipsMap := c.getVirtualIPs(pod, podNets)

×

1117

×

1118

        // associated with security group

×

1119

        for _, podNet := range podNets {

×

1120

                portSecurity := false

×

1121

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

×

1122

                        portSecurity = true

×

1123

×

1125

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

×

1126

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

×

1127

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

×

1128

×

1129

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

×

1130

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

×

1131

                        return err

×

1132

×

1134

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

×

1135

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

×

1136

                var securityGroups string

×

1137

                if securityGroupAnnotation != "" {

×

1138

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

×

1139

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

×

1140

                                if sgName != "" {

×

1141

                                        c.syncSgPortsQueue.Add(sgName)

×

1142

×

1145

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

×

1146

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

×

1147

                        return err

×

1148

×

1150

        return nil

×

1153

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

×

1154

        podName := c.getNameByPod(pod)

×

1155

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

×

1156

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

×

1157

        portsNeedToDel := []string{}

×

1158

        annotationsNeedToDel := []string{}

×

NEW

1159

        annotationsNeedToAdd := make(map[string]string)

×

1160

        subnetUsedByPort := make(map[string]string)

×

1161

×

1162

        for _, podNet := range podNets {

×

1163

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

×

1164

                targetPortNameList.Add(portName)

×

NEW

1165

                if podNet.IPRequest != "" {

×

NEW

1166

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

×

NEW

1167

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

×

NEW

1168

×

NEW

1170

                if podNet.MacRequest != "" {

×

NEW

1171

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

×

NEW

1172

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

×

NEW

1173

×

1176

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

×

1177

        if err != nil {

×

1178

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

×

1179

                return nil, err

×

1180

×

1182

        for _, port := range ports {

×

1183

                if !targetPortNameList.Has(port.Name) {

×

1184

                        portsNeedToDel = append(portsNeedToDel, port.Name)

×

1185

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

×

1186

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

×

1187

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

×

1188

                        if providerName == util.OvnProvider {

×

1189

                                continue

×

1191

                        annotationsNeedToDel = append(annotationsNeedToDel, providerName)

×

NEW

1195

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

×

1196

                return pod, nil

×

1197

×

1199

        for _, portNeedDel := range portsNeedToDel {

×

1200

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

×

1201

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

×

1202

                        subnet.ReleaseAddressWithNicName(podName, portNeedDel)

×

1203

×

1204

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

×

1205

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

×

1206

                        return nil, err

×

1207

×

1208

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

×

1209

                        if !k8serrors.IsNotFound(err) {

×

1210

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

×

1211

                                return nil, err

×

1212

×

1216

        for _, providerName := range annotationsNeedToDel {

×

1217

                for annotationKey := range pod.Annotations {

×

1218

                        if strings.HasPrefix(annotationKey, providerName) {

×

1219

                                delete(pod.Annotations, annotationKey)

×

1220

×

NEW

1224

        for annotationKey, annotationValue := range annotationsNeedToAdd {

×

NEW

1225

                pod.Annotations[annotationKey] = annotationValue

×

NEW

1226

×

1228

        if len(cachedPod.Annotations) == len(pod.Annotations) {

×

1229

                return pod, nil

×

1230

×

1232

        patch, err := util.GenerateMergePatchPayload(cachedPod, pod)

×

1233

        if err != nil {

×

1234

                klog.Errorf("failed to generate patch payload for pod '%s', %v", pod.Name, err)

×

1235

                return nil, err

×

1236

×

1237

        patchedPod, err := c.config.KubeClient.CoreV1().Pods(pod.Namespace).Patch(context.Background(), pod.Name,

×

1238

                types.MergePatchType, patch, metav1.PatchOptions{}, "")

×

1239

        if err != nil {

×

1240

                if k8serrors.IsNotFound(err) {

×

1241

                        return nil, nil

×

1242

×

1243

                klog.Errorf("failed to delete useless annotations for pod %s: %v", pod.Name, err)

×

1244

                return nil, err

×

1247

        return patchedPod.DeepCopy(), nil

×

1250

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

×

1251

        for _, owner := range pod.OwnerReferences {

×

1252

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

×

1253

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

×

1254

                                return true, owner.Name, owner.UID

×

1255

×

1258

        return false, "", ""

×

1261

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

×

1262

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

×

1263

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

×

1264

        if err != nil {

×

1265

                // statefulset is deleted

×

1266

                if k8serrors.IsNotFound(err) {

×

1267

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

×

1268

                        return true

×

1269

×

1270

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

×

1271

                return false

×

1275

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

×

1276

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

×

1277

                return true

×

1278

×

1281

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

×

1282

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

×

1283

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

×

1284

        if err != nil {

×

1285

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

×

1286

                return false

×

1287

×

1289

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

×

1290

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

×

1291

                return true

×

1292

×

1293

        return false

×

1296

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

×

1297

        var nodeTunlIPAddr []net.IP

×

1298

        nodeTunlIP := node.Annotations[util.IPAddressAnnotation]

×

1299

        if nodeTunlIP == "" {

×

1300

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

×

1301

×

1303

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

×

1304

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

×

1305

×

1306

        return nodeTunlIPAddr, nil

×

1309

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

×

1310

        // validation check by caller

×

1311

        nextHop := gw[0].String()

×

1312

        if len(gw) == 2 {

×

1313

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

×

1314

×

1315

        return nextHop

×

1318

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

×

1319

        // check if allocate from subnet is need.

×

1320

        // allocate subnet when change subnet to hotplug nic

×

1321

        // allocate subnet when migrate vm

×

1322

        if !isPodAlive(pod) {

×

1323

                return nil

×

1324

×

1326

        if pod.Annotations == nil {

×

1327

                return nets

×

1328

×

1330

        migrate := false

×

1331

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

×

1332

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

×

1333

                migrate = true

×

1334

×

1336

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

×

1337

        for _, n := range nets {

×

1338

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

×

1339

                        result = append(result, n)

×

1340

×

1342

        return result

×

1345

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

×

1346

        // 1. check annotations

×

1347

        if pod.Annotations == nil {

×

1348

                return true, nil

×

1349

×

1351

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

×

1352

                return true, nil

×

1353

×

1355

        attachmentNets, err := c.getPodAttachmentNet(pod)

×

1356

        if err != nil {

×

1357

                klog.Error(err)

×

1358

                return false, err

×

1359

×

1360

        for _, n := range attachmentNets {

×

1361

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

×

1362

                        return true, nil

×

1363

×

1364

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

×

1365

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

×

1366

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

×

1367

                        // need to sync to create ip

×

1368

                        klog.Error(err)

×

1369

                        return true, nil

×

1370

×

1372

        return false, nil

×

1375

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

×

1376

        if !isPodAlive(pod) {

×

1377

                return nil

×

1378

×

1380

        if pod.Annotations == nil {

×

1381

                return nets

×

1382

×

1384

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

×

1385

        for _, n := range nets {

×

1386

                if !isOvnSubnet(n.Subnet) {

×

1387

                        continue

×

1390

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

×

1391

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

×

1392

                                result = append(result, n)

×

1393

×

1396

        return result

×

1399

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

×

1400

        var subnet *kubeovnv1.Subnet

×

1401

        var err error

×

1402

        // 1. check annotation subnet

×

1403

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

×

1404

        if lsExist {

×

1405

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

×

1406

                if err != nil {

×

1407

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

×

1408

                        return nil, err

×

1409

×

1410

        } else {

×

1411

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

×

1412

                if err != nil {

×

1413

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

×

1414

                        return nil, err

×

1415

×

1416

                if ns.Annotations == nil {

×

1417

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

×

1418

                        klog.Error(err)

×

1419

                        return nil, err

×

1420

×

1422

                subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]

×

1423

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

×

1424

                        if subnetName == "" {

×

1425

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

×

1426

                                klog.Error(err)

×

1427

                                return nil, err

×

1428

×

1429

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

×

1430

                        if err != nil {

×

1431

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

×

1432

                                return nil, err

×

1433

×

1435

                        switch subnet.Spec.Protocol {

×

1436

                        case kubeovnv1.ProtocolIPv4:

×

1437

                                fallthrough

×

1438

                        case kubeovnv1.ProtocolDual:

×

1439

                                if subnet.Status.V4AvailableIPs == 0 {

×

1440

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

×

1441

                                        continue

×

1443

                        case kubeovnv1.ProtocolIPv6:

×

1444

                                if subnet.Status.V6AvailableIPs == 0 {

×

1445

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

×

1446

                                        continue

×

1449

                        break

×

1452

        return subnet, nil

×

1455

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

×

1456

        delegateConf := &multustypes.DelegateNetConf{}

×

1457

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

×

1458

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

×

1459

×

1461

        if delegateConf.Conf.Type == "" {

×

1462

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

×

1463

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

×

1464

×

1466

        return delegateConf, nil

×

1486

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

×

1487

        var multusNets []*multustypes.NetworkSelectionElement

×

1488

        defaultAttachNetworks := pod.Annotations[util.DefaultNetworkAnnotation]

×

1489

        if defaultAttachNetworks != "" {

×

1490

                attachments, err := util.ParsePodNetworkAnnotation(defaultAttachNetworks, pod.Namespace)

×

1491

                if err != nil {

×

1492

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

×

1493

                        return nil, err

×

1494

×

1495

                multusNets = attachments

×

1498

        attachNetworks := pod.Annotations[util.AttachmentNetworkAnnotation]

×

1499

        if attachNetworks != "" {

×

1500

                attachments, err := util.ParsePodNetworkAnnotation(attachNetworks, pod.Namespace)

×

1501

                if err != nil {

×

1502

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

×

1503

                        return nil, err

×

1504

×

1505

                multusNets = append(multusNets, attachments...)

×

1507

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

×

1508

        if err != nil {

×

1509

                return nil, err

×

1510

×

1512

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

×

1513

        for _, attach := range multusNets {

×

1514

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

×

1515

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

×

1516

                if err != nil {

×

1517

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

×

1518

                        return nil, err

×

1519

×

1521

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

×

1522

                if err != nil {

×

1523

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

×

1524

                        return nil, err

×

1525

×

1528

                var providerName string

×

1529

                if util.IsOvnNetwork(netCfg) {

×

1530

                        allowLiveMigration := false

×

1531

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

×

1532

×

1533

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

×

1534

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

×

1535

                                allowLiveMigration = true

×

1536

×

1538

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

×

1539

                        if subnetName == "" {

×

1540

                                for _, subnet := range subnets {

×

1541

                                        if subnet.Spec.Provider == providerName {

×

1542

                                                subnetName = subnet.Name

×

1543

                                                break

×

1547

                        var subnet *kubeovnv1.Subnet

×

1548

                        if subnetName == "" {

×

1549

                                subnet, err = c.getPodDefaultSubnet(pod)

×

1550

                                if err != nil {

×

1551

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

×

1552

                                        return nil, err

×

1553

×

1554

                        } else {

×

1555

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

×

1556

                                if err != nil {

×

1557

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

×

1558

                                        return nil, err

×

1559

×

NEW

1562

                        ret := &kubeovnNet{

×

1563

                                Type:               providerTypeOriginal,

×

1564

                                ProviderName:       providerName,

×

1565

                                Subnet:             subnet,

×

1566

                                IsDefault:          isDefault,

×

1567

                                AllowLiveMigration: allowLiveMigration,

×

NEW

1568

×

NEW

1569

×

NEW

1570

                        if len(attach.IPRequest) != 0 {

×

NEW

1571

                                ret.IPRequest = attach.IPRequest[0]

×

NEW

1572

×

NEW

1574

                        ret.MacRequest = attach.MacRequest

×

NEW

1575

                        result = append(result, ret)

×

1576

                } else {

×

1577

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

×

1578

                        for _, subnet := range subnets {

×

1579

                                if subnet.Spec.Provider == providerName {

×

1580

                                        result = append(result, &kubeovnNet{

×

1581

                                                Type:         providerTypeIPAM,

×

1582

                                                ProviderName: providerName,

×

1583

                                                Subnet:       subnet,

×

1584

})

×

1585

                                        break

×

1590

        return result, nil

×

1593

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

×

1594

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

×

1595

        if err != nil {

×

1596

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

×

1597

                return false, false, err

×

1598

×

1600

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

×

1601

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

×

1602

                if err != nil {

×

1603

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

×

1604

                        return false, false, err

×

1605

×

1607

                for _, node := range nodes {

×

1608

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

×

1609

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

×

1610

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

×

1611

                                return false, true, nil

×

1612

×

1613

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

×

1614

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

×

1615

                                return true, false, nil

×

1616

×

1620

        return true, true, nil

×

1623

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

×

1624

        podName := c.getNameByPod(pod)

×

1625

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

×

1626

×

1627

        var checkVMPod bool

×

1628

        isStsPod, _, _ := isStatefulSetPod(pod)

×

1629

        // if pod has static vip

×

1630

        vipName := pod.Annotations[util.VipAnnotation]

×

1631

        if vipName != "" {

×

1632

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

×

1633

                if err != nil {

×

1634

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

×

1635

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

×

1636

×

1637

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

×

1638

                if c.config.EnableKeepVMIP {

×

1639

                        checkVMPod, _ = isVMPod(pod)

×

1640

×

1641

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

×

1642

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

×

1643

×

1644

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

×

1647

        var macPointer *string

×

1648

        if isOvnSubnet(podNet.Subnet) {

×

1649

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

×

1650

                if annoMAC != "" {

×

1651

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

×

1652

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

×

1653

×

1654

                        macPointer = &annoMAC

×

1656

        } else {

×

1657

                macPointer = ptr.To("")

×

1658

×

1660

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

×

1661

        if ippoolStr == "" {

×

1662

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

×

1663

                if err != nil {

×

1664

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

×

1665

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

×

1666

×

1667

                if len(ns.Annotations) != 0 {

×

1668

                        ippoolStr = ns.Annotations[util.IPPoolAnnotation]

×

1669

×

1673

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

×

1674

                ippoolStr == "" {

×

1675

                var skippedAddrs []string

×

1676

                for {

×

1677

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

×

1678

×

1679

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

×

1680

                        if err != nil {

×

1681

                                klog.Error(err)

×

1682

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

×

1683

×

1684

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

×

1685

                        if err != nil {

×

1686

                                klog.Error(err)

×

1687

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

×

1688

×

1689

                        if ipv4OK && ipv6OK {

×

1690

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

×

1691

×

1693

                        if !ipv4OK {

×

1694

                                skippedAddrs = append(skippedAddrs, ipv4)

×

1695

×

1696

                        if !ipv6OK {

×

1697

                                skippedAddrs = append(skippedAddrs, ipv6)

×

1698

×

1702

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

×

1703

×

1704

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

×

1705

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

×

1706

        var v4IP, v6IP, mac string

×

1707

        var err error

×

1708

×

1709

        // Static allocate

×

1710

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

×

1711

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

×

1712

×

1713

                for _, net := range nsNets {

×

1714

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

×

1715

                        if err == nil {

×

1716

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

×

1717

×

1719

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

×

1723

        if ippoolStr != "" {

×

1724

                var ipPool []string

×

1725

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

×

1726

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

×

1727

                } else {

×

1728

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

×

1729

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

×

1730

                                ipPool = []string{ippoolStr}

×

1731

×

1733

                for i, ip := range ipPool {

×

1734

                        ipPool[i] = strings.TrimSpace(ip)

×

1735

×

1737

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

×

1738

                        var skippedAddrs []string

×

1739

                        for {

×

1740

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

×

1741

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

×

1742

                                if err != nil {

×

1743

                                        klog.Error(err)

×

1744

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

×

1745

×

1746

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

×

1747

                                if err != nil {

×

1748

                                        klog.Error(err)

×

1749

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

×

1750

×

1751

                                if ipv4OK && ipv6OK {

×

1752

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

×

1753

×

1755

                                if !ipv4OK {

×

1756

                                        skippedAddrs = append(skippedAddrs, ipv4)

×

1757

×

1758

                                if !ipv6OK {

×

1759

                                        skippedAddrs = append(skippedAddrs, ipv6)

×

1760

×

1764

                if !isStsPod {

×

1765

                        for _, net := range nsNets {

×

1766

                                for _, staticIP := range ipPool {

×

1767

                                        var checkIP string

×

1768

                                        ipProtocol := util.CheckProtocol(staticIP)

×

1769

                                        if ipProtocol == kubeovnv1.ProtocolDual {

×

1770

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

×

1771

                                        } else {

×

1772

                                                checkIP = staticIP

×

1773

×

1775

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

×

1776

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

×

1777

                                                continue

×

1780

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

×

1781

                                        if err == nil {

×

1782

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

×

1783

×

1786

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

×

1787

                } else {

×

1788

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

×

1789

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

×

1790

                        index, _ := strconv.Atoi(numStr)

×

1791

×

1792

                        if index < len(ipPool) {

×

1793

                                for _, net := range nsNets {

×

1794

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

×

1795

                                        if err == nil {

×

1796

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

×

1797

×

1799

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

×

1803

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

×

1804

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

×

1807

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

×

1808

        var v4IP, v6IP, macStr string

×

1809

        var err error

×

1810

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

×

1811

                if net.ParseIP(ipStr) == nil {

×

1812

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

×

1813

×

1816

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

×

1817

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

×

1818

                return "", "", "", err

×

1819

×

1820

        return v4IP, v6IP, macStr, nil

×

1823

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

×

1824

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

×

1825

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

×

1826

        if err != nil {

×

1827

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

×

1828

                return false, err

×

1829

×

1832

        var ownerRefSubnetExist bool

×

1833

        var ownerRefSubnet string

×

1834

        switch ownerRefKind {

×

1835

        case util.StatefulSet:

×

1836

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

×

1837

                if err != nil {

×

1838

                        if k8serrors.IsNotFound(err) {

×

1839

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

×

1840

                                return true, nil

×

1841

×

1842

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

×

1844

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

×

1845

                        ownerRefSubnetExist = true

×

1846

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

×

1847

×

1849

        case util.VMInstance:

×

1850

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

×

1851

                if err != nil {

×

1852

                        if k8serrors.IsNotFound(err) {

×

1853

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

×

1854

                                return true, nil

×

1855

×

1856

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

×

1858

                if vm != nil &&

×

1859

                        vm.Spec.Template != nil &&

×

1860

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

×

1861

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

×

1862

                        ownerRefSubnetExist = true

×

1863

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

×

1864

×

1866

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

×

1867

        if !ownerRefSubnetExist {

×

1868

                nsSubnetNames := podNs.Annotations[util.LogicalSwitchAnnotation]

×

1869

                // check if pod use the subnet of its ns

×

1870

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

×

1871

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

×

1872

                        return true, nil

×

1873

×

1877

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

×

1878

        if err != nil {

×

1879

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

×

1880

                return false, err

×

1881

×

1882

        if podSubnet == nil {

×

1883

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

×

1884

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

×

1885

                return false, nil

×

1886

×

1887

        podIP := pod.Annotations[util.IPAddressAnnotation]

×

1888

        if podIP == "" {

×

1889

                // delete pod just after it created < 1ms

×

1890

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

×

1891

                return false, nil

×

1892

×

1893

        podSubnetCidr := podSubnet.Spec.CIDRBlock

×

1894

        if podSubnetCidr != "" {

×

1895

                // subnet spec cidr changed by user

×

1896

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

×

1897

                return false, nil

×

1898

×

1899

        if !util.CIDRContainIP(podSubnetCidr, podIP) {

×

1900

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

×

1901

                return true, nil

×

1902

×

1904

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

×

1905

                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)

×

1906

                return true, nil

×

1907

×

1909

        return false, nil

×

1912

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

×

1913

        for _, owner := range pod.OwnerReferences {

×

1914

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

×

1915

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

×

1916

                        return true, owner.Name

×

1917

×

1919

        return false, ""

×

1922

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

×

1923

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

×

1924

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

×

1925

                        return true, owner.Name

×

1926

×

1928

        return false, ""

×

1931

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

×

1932

        var (

×

1933

                vmiAlive bool

×

1934

                vmName   string

×

1935

×

1936

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

×

1937

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

×

1938

        if err != nil {

×

1939

                if k8serrors.IsNotFound(err) {

×

1940

                        vmiAlive = false

×

1941

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

×

1942

                        vmName = vmiName

×

1943

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

×

1944

                } else {

×

1945

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

×

1946

                        return false

×

1947

×

1948

        } else {

×

1949

                var ownsByVM bool

×

1950

                ownsByVM, vmName = isOwnsByTheVM(vmi)

×

1951

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

×

1952

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

×

1953

                        return true

×

1954

×

1955

                vmiAlive = vmi.DeletionTimestamp.IsZero()

×

1958

        if vmiAlive {

×

1959

                return false

×

1960

×

1962

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

×

1963

        if err != nil {

×

1964

                // the vm has gone

×

1965

                if k8serrors.IsNotFound(err) {

×

1966

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

×

1967

                        return true

×

1968

×

1969

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

×

1970

                return false

×

1973

        if !vm.DeletionTimestamp.IsZero() {

×

1974

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

×

1975

                return true

×

1976

×

1977

        return false

×

1980

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

×

1981

        if c.config.EnableKeepVMIP {

×

1982

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

×

1983

                        return vmName

×

1984

×

1986

        return pod.Name

×

1990

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

×

1991

        var result []*kubeovnNet

×

1992

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

×

1993

        result = append(result, podNet)

×

1994

×

1995

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

×

1996

        if err != nil {

×

1997

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

×

1998

                return nil, err

×

1999

×

2000

        if ns.Annotations == nil {

×

2001

                return nil, nil

×

2002

×

2004

        subnetNames := ns.Annotations[util.LogicalSwitchAnnotation]

×

2005

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

×

2006

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

×

2007

                        continue

×

2009

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

×

2010

                if err != nil {

×

2011

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

×

2012

                        return nil, err

×

2013

×

2015

                result = append(result, &kubeovnNet{

×

2016

                        Type:         providerTypeOriginal,

×

2017

                        ProviderName: subnet.Spec.Provider,

×

2018

                        Subnet:       subnet,

×

2019

})

×

2022

        return result, nil

×

2025

func getPodType(pod *v1.Pod) string {

×

2026

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

×

2027

                return util.StatefulSet

×

2028

×

2030

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

×

2031

                return util.VM

×

2032

×

2033

        return ""

×

2036

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

×

2037

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

×

2038

        var vipNamesList []string

×

2039

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

×

2040

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

×

2041

                        continue

×

2043

                if !slices.Contains(vipNamesList, vipName) {

×

2044

                        vipNamesList = append(vipNamesList, vipName)

×

2045

                } else {

×

2046

                        continue

×

2048

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

×

2049

                if err != nil {

×

2050

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

×

2051

                        continue

×

2053

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

×

2054

                        continue

×

2056

                for _, podNet := range podNets {

×

2057

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

×

2058

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

×

2059

                                vipsList := vipsListMap[key]

×

2060

                                if vipsList == nil {

×

2061

                                        vipsList = []string{}

×

2062

×

2064

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

×

2065

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

×

2066

×

2067

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

×

2068

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

×

2069

×

2071

                                vipsListMap[key] = vipsList

×

2076

        for _, podNet := range podNets {

×

2077

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

×

2078

                if vipStr == "" {

×

2079

                        continue

×

2081

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

×

2082

                vipsList := vipsListMap[key]

×

2083

                if vipsList == nil {

×

2084

                        vipsList = []string{}

×

2085

×

2087

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

×

2088

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

×

2089

                                vipsList = append(vipsList, vip)

×

2090

×

2093

                vipsListMap[key] = vipsList

×

2096

        vipsMap := make(map[string]string)

×

2097

        for key, vipsList := range vipsListMap {

×

2098

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

×

2099

×

2100

        return vipsMap

×

2103

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

×

2104

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

×

2105

        if len(routes) == 0 {

×

2106

                delete(annotations, key)

×

2107

                return nil

×

2108

×

2110

        buf, err := json.Marshal(routes)

×

2111

        if err != nil {

×

2112

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

×

2113

                klog.Error(err)

×

2114

                return err

×

2115

×

2116

        annotations[key] = string(buf)

×

2117

×

2118

        return nil

×

kubeovn / kube-ovn / 12275027635

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

kubeovn / kube-ovn / 12275027635

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