kubeovn/kube-ovn | Build 17228354799 | pkg/controller/ip.go | Coveralls

28

func (c *Controller) enqueueAddIP(obj any) {

×

29

        ipObj := obj.(*kubeovnv1.IP)

×

30

        if strings.HasPrefix(ipObj.Name, util.U2OInterconnName[0:19]) {

×

31

                return

×

32

×

33

        klog.V(3).Infof("enqueue update status subnet %s", ipObj.Spec.Subnet)

×

34

        c.updateSubnetStatusQueue.Add(ipObj.Spec.Subnet)

×

35

        for _, as := range ipObj.Spec.AttachSubnets {

×

36

                klog.V(3).Infof("enqueue update attach status for subnet %s", as)

×

37

                c.updateSubnetStatusQueue.Add(as)

×

38

×

40

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

×

41

        klog.V(3).Infof("enqueue add ip %s", key)

×

42

        c.addIPQueue.Add(key)

×

45

func (c *Controller) enqueueUpdateIP(oldObj, newObj any) {

×

46

        oldIP := oldObj.(*kubeovnv1.IP)

×

47

        newIP := newObj.(*kubeovnv1.IP)

×

48

        // ip can not change these specs below

×

49

        if oldIP.Spec.Subnet != "" && newIP.Spec.Subnet != oldIP.Spec.Subnet {

×

50

                klog.Errorf("ip %s subnet can not change", newIP.Name)

×

51

                return

×

52

×

53

        if oldIP.Spec.Namespace != "" && newIP.Spec.Namespace != oldIP.Spec.Namespace {

×

54

                klog.Errorf("ip %s namespace can not change", newIP.Name)

×

55

                return

×

56

×

57

        if oldIP.Spec.PodName != "" && newIP.Spec.PodName != oldIP.Spec.PodName {

×

58

                klog.Errorf("ip %s podName can not change", newIP.Name)

×

59

                return

×

60

×

61

        if oldIP.Spec.PodType != "" && newIP.Spec.PodType != oldIP.Spec.PodType {

×

62

                klog.Errorf("ip %s podType can not change", newIP.Name)

×

63

                return

×

64

×

65

        if oldIP.Spec.MacAddress != "" && newIP.Spec.MacAddress != oldIP.Spec.MacAddress {

×

66

                klog.Errorf("ip %s macAddress can not change", newIP.Name)

×

67

                return

×

68

×

69

        if oldIP.Spec.V4IPAddress != "" && newIP.Spec.V4IPAddress != oldIP.Spec.V4IPAddress {

×

70

                klog.Errorf("ip %s v4IPAddress can not change", newIP.Name)

×

71

                return

×

72

×

73

        if oldIP.Spec.V6IPAddress != "" {

×

74

                // v6 ip address can not use upper case

×

75

                if util.ContainsUppercase(newIP.Spec.V6IPAddress) {

×

76

                        err := fmt.Errorf("ip %s v6 ip address %s can not contain upper case", newIP.Name, newIP.Spec.V6IPAddress)

×

77

                        klog.Error(err)

×

78

                        return

×

79

×

80

                if newIP.Spec.V6IPAddress != oldIP.Spec.V6IPAddress {

×

81

                        klog.Errorf("ip %s v6IPAddress can not change", newIP.Name)

×

82

                        return

×

83

×

85

        if !newIP.DeletionTimestamp.IsZero() {

×

86

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

×

87

                klog.V(3).Infof("enqueue update ip %s", key)

×

88

                c.updateIPQueue.Add(key)

×

89

                return

×

90

×

91

        if !slices.Equal(oldIP.Spec.AttachSubnets, newIP.Spec.AttachSubnets) {

×

92

                klog.V(3).Infof("enqueue update status subnet %s", newIP.Spec.Subnet)

×

93

                for _, as := range newIP.Spec.AttachSubnets {

×

94

                        klog.V(3).Infof("enqueue update status for attach subnet %s", as)

×

95

                        c.updateSubnetStatusQueue.Add(as)

×

96

×

100

func (c *Controller) enqueueDelIP(obj any) {

×

NEW

101

        var ipObj *kubeovnv1.IP

×

NEW

102

        switch t := obj.(type) {

×

NEW

103

        case *kubeovnv1.IP:

×

NEW

104

                ipObj = t

×

NEW

105

        case cache.DeletedFinalStateUnknown:

×

NEW

106

                ip, ok := t.Obj.(*kubeovnv1.IP)

×

NEW

107

                if !ok {

×

NEW

108

                        klog.Warningf("unexpected object type: %T", t.Obj)

×

NEW

109

                        return

×

NEW

110

×

NEW

111

                ipObj = ip

×

NEW

112

        default:

×

NEW

113

                klog.Warningf("unexpected type: %T", obj)

×

NEW

114

                return

×

117

        if strings.HasPrefix(ipObj.Name, util.U2OInterconnName[0:19]) {

×

118

                return

×

119

×

121

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

×

122

        klog.V(3).Infof("enqueue del ip %s", key)

×

123

        c.delIPQueue.Add(ipObj)

×

126

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

×

127

        ip, err := c.ipsLister.Get(key)

×

128

        if err != nil {

×

129

                if k8serrors.IsNotFound(err) {

×

130

                        return nil

×

131

×

132

                klog.Error(err)

×

133

                return err

×

135

        ip = ip.DeepCopy()

×

136

        klog.V(3).Infof("handle add reserved ip %s", ip.Name)

×

137

×

138

        if !ip.DeletionTimestamp.IsZero() {

×

139

                klog.Infof("handle add process stop for deleting ip %s", ip.Name)

×

140

                return nil

×

141

×

142

        if ip.Spec.Subnet == "" {

×

143

                err := errors.New("subnet parameter cannot be empty")

×

144

                klog.Error(err)

×

145

                return err

×

146

×

148

        if ip.Spec.PodType != "" && ip.Spec.PodType != util.VM && ip.Spec.PodType != util.StatefulSet {

×

149

                err := fmt.Errorf("podType %s is not supported", ip.Spec.PodType)

×

150

                klog.Error(err)

×

151

                return err

×

152

×

154

        subnet, err := c.subnetsLister.Get(ip.Spec.Subnet)

×

155

        if err != nil {

×

156

                err = fmt.Errorf("failed to get subnet %s: %w", ip.Spec.Subnet, err)

×

157

                klog.Error(err)

×

158

                return err

×

159

×

161

        portName := ovs.PodNameToPortName(ip.Spec.PodName, ip.Spec.Namespace, subnet.Spec.Provider)

×

162

        if portName != ip.Name {

×

163

                // invalid ip or node ip, no need to handle it here

×

164

                klog.V(3).Infof("port name %s is not equal to ip name %s", portName, ip.Name)

×

165

                return nil

×

166

×

169

        lsp, err := c.OVNNbClient.GetLogicalSwitchPort(portName, true)

×

170

        if err != nil {

×

171

                klog.Errorf("failed to list logical switch ports %s, %v", portName, err)

×

172

                return err

×

173

×

174

        if lsp != nil {

×

175

                // port already exists means the ip already created

×

176

                klog.V(3).Infof("ip %s is ready", portName)

×

177

                return nil

×

178

×

181

        if util.ContainsUppercase(ip.Spec.V6IPAddress) {

×

182

                err := fmt.Errorf("ip %s v6 ip address %s can not contain upper case", ip.Name, ip.Spec.V6IPAddress)

×

183

                klog.Error(err)

×

184

                return err

×

185

×

186

        v4IP, v6IP, mac, err := c.ipAcquireAddress(ip, subnet)

×

187

        if err != nil {

×

188

                err = fmt.Errorf("failed to acquire ip address %w", err)

×

189

                klog.Error(err)

×

190

                return err

×

191

×

192

        ipStr := util.GetStringIP(v4IP, v6IP)

×

193

        if err := c.createOrUpdateIPCR(ip.Name, ip.Spec.PodName, ipStr, mac, subnet.Name, ip.Spec.Namespace, ip.Spec.NodeName, ip.Spec.PodType); err != nil {

×

194

                err = fmt.Errorf("failed to create ips CR %s.%s: %w", ip.Spec.PodName, ip.Spec.Namespace, err)

×

195

                klog.Error(err)

×

196

                return err

×

197

×

198

        if ip.Labels == nil {

×

199

                ip.Labels = map[string]string{}

×

200

×

201

        if ip.Labels[util.IPReservedLabel] != "false" {

×

202

                ip.Labels[util.IPReservedLabel] = "true"

×

203

                patchPayloadTemplate := `[{ "op": "%s", "path": "/metadata/labels", "value": %s }]`

×

204

                raw, err := json.Marshal(ip.Labels)

×

205

                if err != nil {

×

206

                        klog.Error(err)

×

207

                        return err

×

208

×

209

                op := "replace"

×

210

                patchPayload := fmt.Sprintf(patchPayloadTemplate, op, raw)

×

211

                if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), ip.Name,

×

212

                        types.JSONPatchType, []byte(patchPayload), metav1.PatchOptions{}); err != nil {

×

213

                        klog.Errorf("failed to patch label for ip %s, %v", ip.Name, err)

×

214

                        return err

×

215

×

217

        return nil

×

220

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

×

221

        cachedIP, err := c.ipsLister.Get(key)

×

222

        if err != nil {

×

223

                if k8serrors.IsNotFound(err) {

×

224

                        return nil

×

225

×

226

                klog.Error(err)

×

227

                return err

×

229

        if !cachedIP.DeletionTimestamp.IsZero() {

×

230

                klog.Infof("handle deleting ip %s", cachedIP.Name)

×

231

                subnet, err := c.subnetsLister.Get(cachedIP.Spec.Subnet)

×

232

                if err != nil {

×

233

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

×

234

                        return err

×

235

×

236

                portName := cachedIP.Name

×

237

                if isOvnSubnet(subnet) {

×

238

                        port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, true)

×

239

                        if err != nil {

×

240

                                klog.Errorf("failed to get logical switch port %s: %v", portName, err)

×

241

                                return err

×

242

×

243

                        if port != nil {

×

244

                                klog.Infof("delete ip cr lsp %s from switch %s", portName, subnet.Name)

×

245

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

×

246

                                        klog.Errorf("failed to delete ip cr lsp %s from switch %s: %v", portName, subnet.Name, err)

×

247

                                        return err

×

248

×

249

                                klog.V(3).Infof("sync sg for deleted port %s", portName)

×

250

                                sgList, err := c.getPortSg(port)

×

251

                                if err != nil {

×

252

                                        klog.Errorf("get port sg failed, %v", err)

×

253

                                        return err

×

254

×

255

                                for _, sgName := range sgList {

×

256

                                        if sgName != "" {

×

257

                                                c.syncSgPortsQueue.Add(sgName)

×

258

×

262

                podKey := fmt.Sprintf("%s/%s", cachedIP.Spec.Namespace, cachedIP.Spec.PodName)

×

263

                klog.Infof("ip cr %s release ipam pod key %s from subnet %s", cachedIP.Name, podKey, cachedIP.Spec.Subnet)

×

264

                c.ipam.ReleaseAddressByNic(podKey, portName, cachedIP.Spec.Subnet)

×

265

                if err = c.handleDelIPFinalizer(cachedIP); err != nil {

×

266

                        klog.Errorf("failed to handle del ip finalizer %v", err)

×

267

                        return err

×

268

×

269

                c.updateSubnetStatusQueue.Add(cachedIP.Spec.Subnet)

×

271

        return nil

×

274

func (c *Controller) handleDelIP(ip *kubeovnv1.IP) error {

×

275

        klog.Infof("deleting ip %s enqueue update status subnet %s", ip.Name, ip.Spec.Subnet)

×

276

        c.updateSubnetStatusQueue.Add(ip.Spec.Subnet)

×

277

        for _, as := range ip.Spec.AttachSubnets {

×

278

                klog.V(3).Infof("enqueue update attach status for subnet %s", as)

×

279

                c.updateSubnetStatusQueue.Add(as)

×

280

×

281

        return nil

×

284

func (c *Controller) syncIPFinalizer(cl client.Client) error {

×

285

        // migrate depreciated finalizer to new finalizer

×

286

        ips := &kubeovnv1.IPList{}

×

287

        return migrateFinalizers(cl, ips, func(i int) (client.Object, client.Object) {

×

288

                if i < 0 || i >= len(ips.Items) {

×

289

                        return nil, nil

×

290

×

291

                return ips.Items[i].DeepCopy(), ips.Items[i].DeepCopy()

×

295

func (c *Controller) handleAddIPFinalizer(cachedIP *kubeovnv1.IP) error {

×

296

        if !cachedIP.DeletionTimestamp.IsZero() || len(cachedIP.GetFinalizers()) != 0 {

×

297

                return nil

×

298

×

299

        newIP := cachedIP.DeepCopy()

×

300

        controllerutil.AddFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

301

        patch, err := util.GenerateMergePatchPayload(cachedIP, newIP)

×

302

        if err != nil {

×

303

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

×

304

                return err

×

305

×

306

        if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), cachedIP.Name,

×

307

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

×

308

                if k8serrors.IsNotFound(err) {

×

309

                        return nil

×

310

×

311

                klog.Errorf("failed to add finalizer for ip %s, %v", cachedIP.Name, err)

×

312

                return err

×

314

        return nil

×

317

func (c *Controller) handleDelIPFinalizer(cachedIP *kubeovnv1.IP) error {

×

318

        if len(cachedIP.GetFinalizers()) == 0 {

×

319

                return nil

×

320

×

321

        newIP := cachedIP.DeepCopy()

×

322

        controllerutil.RemoveFinalizer(newIP, util.DepreciatedFinalizerName)

×

323

        controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

324

        patch, err := util.GenerateMergePatchPayload(cachedIP, newIP)

×

325

        if err != nil {

×

326

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

×

327

                return err

×

328

×

329

        if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), cachedIP.Name,

×

330

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

×

331

                if k8serrors.IsNotFound(err) {

×

332

                        return nil

×

333

×

334

                klog.Errorf("failed to remove finalizer from ip %s, %v", cachedIP.Name, err)

×

335

                return err

×

337

        return nil

×

340

func (c *Controller) acquireIPAddress(subnetName, name, nicName string) (string, string, string, error) {

×

341

        var skippedAddrs []string

×

342

        var v4ip, v6ip, mac string

×

343

        checkConflict := true

×

344

        var err error

×

345

        for {

×

346

                v4ip, v6ip, mac, err = c.ipam.GetRandomAddress(name, nicName, nil, subnetName, "", skippedAddrs, checkConflict)

×

347

                if err != nil {

×

348

                        klog.Error(err)

×

349

                        return "", "", "", err

×

350

×

352

                ipv4OK, ipv6OK, err := c.validatePodIP(name, subnetName, v4ip, v6ip)

×

353

                if err != nil {

×

354

                        klog.Error(err)

×

355

                        return "", "", "", err

×

356

×

358

                if ipv4OK && ipv6OK {

×

359

                        return v4ip, v6ip, mac, nil

×

360

×

362

                if !ipv4OK {

×

363

                        skippedAddrs = append(skippedAddrs, v4ip)

×

364

×

365

                if !ipv6OK {

×

366

                        skippedAddrs = append(skippedAddrs, v6ip)

×

367

×

371

func (c *Controller) acquireStaticIPAddress(subnetName, name, nicName, ip string, macPointer *string) (string, string, string, error) {

×

372

        checkConflict := true

×

373

        var v4ip, v6ip, mac string

×

374

        var err error

×

375

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

×

376

                if net.ParseIP(ipStr) == nil {

×

377

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

×

378

×

381

        if v4ip, v6ip, mac, err = c.ipam.GetStaticAddress(name, nicName, ip, macPointer, subnetName, checkConflict); err != nil {

×

382

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

×

383

                return "", "", "", err

×

384

×

385

        return v4ip, v6ip, mac, nil

×

388

func (c *Controller) createOrUpdateIPCR(ipCRName, podName, ip, mac, subnetName, ns, nodeName, podType string) error {

×

389

        // `ipCRName`: pod or vm IP name must set ip CR name when creating ip CR

×

390

        var key, ipName string

×

391

        if ipCRName != "" {

×

392

                // pod IP

×

393

                key = podName

×

394

                ipName = ipCRName

×

395

        } else {

×

396

                // node IP or interconn IP

×

397

                switch {

×

398

                case subnetName == c.config.NodeSwitch:

×

399

                        key = nodeName

×

400

                        ipName = util.NodeLspName(nodeName)

×

401

                case strings.HasPrefix(podName, util.U2OInterconnName[0:19]):

×

402

                        key = podName // interconn IP name

×

403

                        ipName = podName

×

404

                case strings.HasPrefix(podName, util.McastQuerierName[0:13]):

×

405

                        key = podName // mcast querier IP name

×

406

                        ipName = podName

×

410

        var err error

×

411

        var ipCR *kubeovnv1.IP

×

412

        ipCR, err = c.ipsLister.Get(ipName)

×

413

        if err != nil {

×

414

                if !k8serrors.IsNotFound(err) {

×

415

                        err := fmt.Errorf("failed to get ip CR %s: %w", ipName, err)

×

416

                        klog.Error(err)

×

417

                        return err

×

418

×

420

                ipCR = nil

×

422

        if ipCR != nil && !ipCR.DeletionTimestamp.IsZero() {

×

423

                // this ip is being deleted, no need to update

×

424

                klog.Infof("enqueue update for removing finalizer to delete ip %s", ipCR.Name)

×

425

                c.updateIPQueue.Add(ipCR.Name)

×

426

                return nil

×

427

×

428

        v4IP, v6IP := util.SplitStringIP(ip)

×

429

        if ipCR == nil {

×

430

                ipCR, err = c.config.KubeOvnClient.KubeovnV1().IPs().Create(context.Background(), &kubeovnv1.IP{

×

431

                        ObjectMeta: metav1.ObjectMeta{

×

432

                                Name: ipName,

×

433

                                Labels: map[string]string{

×

434

                                        util.SubnetNameLabel: subnetName,

×

435

                                        util.NodeNameLabel:   nodeName,

×

436

                                        subnetName:           "",

×

437

                                        util.IPReservedLabel: "false", // ip create with pod or node, ip not reserved

×

438

},

×

439

},

×

440

                        Spec: kubeovnv1.IPSpec{

×

441

                                PodName:       key,

×

442

                                Subnet:        subnetName,

×

443

                                NodeName:      nodeName,

×

444

                                Namespace:     ns,

×

445

                                IPAddress:     ip,

×

446

                                V4IPAddress:   v4IP,

×

447

                                V6IPAddress:   v6IP,

×

448

                                MacAddress:    mac,

×

449

                                AttachIPs:     []string{},

×

450

                                AttachMacs:    []string{},

×

451

                                AttachSubnets: []string{},

×

452

                                PodType:       podType,

×

453

},

×

454

                }, metav1.CreateOptions{})

×

455

                if err != nil {

×

456

                        errMsg := fmt.Errorf("failed to create ip CR %s: %w", ipName, err)

×

457

                        klog.Error(errMsg)

×

458

                        return errMsg

×

459

×

460

        } else {

×

461

                newIPCR := ipCR.DeepCopy()

×

462

                if newIPCR.Labels != nil {

×

463

                        newIPCR.Labels[util.SubnetNameLabel] = subnetName

×

464

                        newIPCR.Labels[util.NodeNameLabel] = nodeName

×

465

                } else {

×

466

                        newIPCR.Labels = map[string]string{

×

467

                                util.SubnetNameLabel: subnetName,

×

468

                                util.NodeNameLabel:   nodeName,

×

469

×

470

                        // update not touch IP Reserved Label

×

471

×

472

                newIPCR.Spec.PodName = key

×

473

                newIPCR.Spec.Namespace = ns

×

474

                newIPCR.Spec.Subnet = subnetName

×

475

                newIPCR.Spec.NodeName = nodeName

×

476

                newIPCR.Spec.IPAddress = ip

×

477

                newIPCR.Spec.V4IPAddress = v4IP

×

478

                newIPCR.Spec.V6IPAddress = v6IP

×

479

                newIPCR.Spec.MacAddress = mac

×

480

                newIPCR.Spec.AttachIPs = []string{}

×

481

                newIPCR.Spec.AttachMacs = []string{}

×

482

                newIPCR.Spec.AttachSubnets = []string{}

×

483

                newIPCR.Spec.PodType = podType

×

484

                if maps.Equal(newIPCR.Labels, ipCR.Labels) && reflect.DeepEqual(newIPCR.Spec, ipCR.Spec) {

×

485

                        return nil

×

486

×

488

                ipCR, err = c.config.KubeOvnClient.KubeovnV1().IPs().Update(context.Background(), newIPCR, metav1.UpdateOptions{})

×

489

                if err != nil {

×

490

                        err := fmt.Errorf("failed to update ip CR %s: %w", ipCRName, err)

×

491

                        klog.Error(err)

×

492

                        return err

×

493

×

495

        if err := c.handleAddIPFinalizer(ipCR); err != nil {

×

496

                klog.Errorf("failed to handle add ip finalizer %v", err)

×

497

                return err

×

498

×

499

        return nil

×

502

func (c *Controller) ipAcquireAddress(ip *kubeovnv1.IP, subnet *kubeovnv1.Subnet) (string, string, string, error) {

×

503

        key := fmt.Sprintf("%s/%s", ip.Spec.Namespace, ip.Spec.PodName)

×

504

        portName := ovs.PodNameToPortName(ip.Spec.PodName, ip.Spec.Namespace, subnet.Spec.Provider)

×

505

        ipStr := util.GetStringIP(ip.Spec.V4IPAddress, ip.Spec.V6IPAddress)

×

506

×

507

        var v4IP, v6IP, mac string

×

508

        var err error

×

509

        var macPtr *string

×

510

        if isOvnSubnet(subnet) {

×

511

                if ip.Spec.MacAddress != "" {

×

512

                        macPtr = &ip.Spec.MacAddress

×

513

×

514

        } else {

×

515

                macPtr = ptr.To("")

×

516

×

518

        if ipStr == "" {

×

519

                // allocate address

×

520

                v4IP, v6IP, mac, err = c.acquireIPAddress(subnet.Name, ip.Name, portName)

×

521

                if err == nil {

×

522

                        return v4IP, v6IP, mac, err

×

523

×

524

                err = fmt.Errorf("failed to get random address for ip %s, %w", ip.Name, err)

×

525

        } else {

×

526

                // static address

×

527

                v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipStr, macPtr, subnet.Name, true)

×

528

                if err == nil {

×

529

                        return v4IP, v6IP, mac, nil

×

530

×

531

                err = fmt.Errorf("failed to get static address for ip %s, %w", ip.Name, err)

×

533

        klog.Error(err)

×

534

        return "", "", "", err

×

kubeovn / kube-ovn / 17228354799

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

1	package controller
2
3	import (
4	"context"
5	"encoding/json"
6	"errors"
7	"fmt"
8	"maps"
9	"net"
10	"reflect"
11	"slices"
12	"strings"
13
14	k8serrors "k8s.io/apimachinery/pkg/api/errors"
15	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
16	"k8s.io/apimachinery/pkg/types"
17	"k8s.io/client-go/tools/cache"
18	"k8s.io/klog/v2"
19	"k8s.io/utils/ptr"
20	"sigs.k8s.io/controller-runtime/pkg/client"
21	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
22
23	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
24	"github.com/kubeovn/kube-ovn/pkg/ovs"
25	"github.com/kubeovn/kube-ovn/pkg/util"
26	)
27
28	func (c *Controller) enqueueAddIP(obj any) {	×
29	ipObj := obj.(*kubeovnv1.IP)	×
30	if strings.HasPrefix(ipObj.Name, util.U2OInterconnName[0:19]) {	×
31	return	×
32	}	×
33	klog.V(3).Infof("enqueue update status subnet %s", ipObj.Spec.Subnet)	×
34	c.updateSubnetStatusQueue.Add(ipObj.Spec.Subnet)	×
35	for _, as := range ipObj.Spec.AttachSubnets {	×
36	klog.V(3).Infof("enqueue update attach status for subnet %s", as)	×
37	c.updateSubnetStatusQueue.Add(as)	×
38	}	×
39
40	key := cache.MetaObjectToName(ipObj).String()	×
41	klog.V(3).Infof("enqueue add ip %s", key)	×
42	c.addIPQueue.Add(key)	×
43	}
44
45	func (c *Controller) enqueueUpdateIP(oldObj, newObj any) {	×
46	oldIP := oldObj.(*kubeovnv1.IP)	×
47	newIP := newObj.(*kubeovnv1.IP)	×
48	// ip can not change these specs below	×
49	if oldIP.Spec.Subnet != "" && newIP.Spec.Subnet != oldIP.Spec.Subnet {	×
50	klog.Errorf("ip %s subnet can not change", newIP.Name)	×
51	return	×
52	}	×
53	if oldIP.Spec.Namespace != "" && newIP.Spec.Namespace != oldIP.Spec.Namespace {	×
54	klog.Errorf("ip %s namespace can not change", newIP.Name)	×
55	return	×
56	}	×
57	if oldIP.Spec.PodName != "" && newIP.Spec.PodName != oldIP.Spec.PodName {	×
58	klog.Errorf("ip %s podName can not change", newIP.Name)	×
59	return	×
60	}	×
61	if oldIP.Spec.PodType != "" && newIP.Spec.PodType != oldIP.Spec.PodType {	×
62	klog.Errorf("ip %s podType can not change", newIP.Name)	×
63	return	×
64	}	×
65	if oldIP.Spec.MacAddress != "" && newIP.Spec.MacAddress != oldIP.Spec.MacAddress {	×
66	klog.Errorf("ip %s macAddress can not change", newIP.Name)	×
67	return	×
68	}	×
69	if oldIP.Spec.V4IPAddress != "" && newIP.Spec.V4IPAddress != oldIP.Spec.V4IPAddress {	×
70	klog.Errorf("ip %s v4IPAddress can not change", newIP.Name)	×
71	return	×
72	}	×
73	if oldIP.Spec.V6IPAddress != "" {	×
74	// v6 ip address can not use upper case	×
75	if util.ContainsUppercase(newIP.Spec.V6IPAddress) {	×
76	err := fmt.Errorf("ip %s v6 ip address %s can not contain upper case", newIP.Name, newIP.Spec.V6IPAddress)	×
77	klog.Error(err)	×
78	return	×
79	}	×
80	if newIP.Spec.V6IPAddress != oldIP.Spec.V6IPAddress {	×
81	klog.Errorf("ip %s v6IPAddress can not change", newIP.Name)	×
82	return	×
83	}	×
84	}
85	if !newIP.DeletionTimestamp.IsZero() {	×
86	key := cache.MetaObjectToName(newIP).String()	×
87	klog.V(3).Infof("enqueue update ip %s", key)	×
88	c.updateIPQueue.Add(key)	×
89	return	×
90	}	×
91	if !slices.Equal(oldIP.Spec.AttachSubnets, newIP.Spec.AttachSubnets) {	×
92	klog.V(3).Infof("enqueue update status subnet %s", newIP.Spec.Subnet)	×
93	for _, as := range newIP.Spec.AttachSubnets {	×
94	klog.V(3).Infof("enqueue update status for attach subnet %s", as)	×
95	c.updateSubnetStatusQueue.Add(as)	×
96	}	×
97	}
98	}
99
100	func (c *Controller) enqueueDelIP(obj any) {	×
NEW 101	var ipObj *kubeovnv1.IP	×
NEW 102	switch t := obj.(type) {	×
NEW 103	case *kubeovnv1.IP:	×
NEW 104	ipObj = t	×
NEW 105	case cache.DeletedFinalStateUnknown:	×
NEW 106	ip, ok := t.Obj.(*kubeovnv1.IP)	×
NEW 107	if !ok {	×
NEW 108	klog.Warningf("unexpected object type: %T", t.Obj)	×
NEW 109	return	×
NEW 110	}	×
NEW 111	ipObj = ip	×
NEW 112	default:	×
NEW 113	klog.Warningf("unexpected type: %T", obj)	×
NEW 114	return	×
115	}
116
117	if strings.HasPrefix(ipObj.Name, util.U2OInterconnName[0:19]) {	×
118	return	×
119	}	×
120
121	key := cache.MetaObjectToName(ipObj).String()	×
122	klog.V(3).Infof("enqueue del ip %s", key)	×
123	c.delIPQueue.Add(ipObj)	×
124	}
125
126	func (c *Controller) handleAddReservedIP(key string) error {	×
127	ip, err := c.ipsLister.Get(key)	×
128	if err != nil {	×
129	if k8serrors.IsNotFound(err) {	×
130	return nil	×
131	}	×
132	klog.Error(err)	×
133	return err	×
134	}
135	ip = ip.DeepCopy()	×
136	klog.V(3).Infof("handle add reserved ip %s", ip.Name)	×
137		×
138	if !ip.DeletionTimestamp.IsZero() {	×
139	klog.Infof("handle add process stop for deleting ip %s", ip.Name)	×
140	return nil	×
141	}	×
142	if ip.Spec.Subnet == "" {	×
143	err := errors.New("subnet parameter cannot be empty")	×
144	klog.Error(err)	×
145	return err	×
146	}	×
147
148	if ip.Spec.PodType != "" && ip.Spec.PodType != util.VM && ip.Spec.PodType != util.StatefulSet {	×
149	err := fmt.Errorf("podType %s is not supported", ip.Spec.PodType)	×
150	klog.Error(err)	×
151	return err	×
152	}	×
153
154	subnet, err := c.subnetsLister.Get(ip.Spec.Subnet)	×
155	if err != nil {	×
156	err = fmt.Errorf("failed to get subnet %s: %w", ip.Spec.Subnet, err)	×
157	klog.Error(err)	×
158	return err	×
159	}	×
160
161	portName := ovs.PodNameToPortName(ip.Spec.PodName, ip.Spec.Namespace, subnet.Spec.Provider)	×
162	if portName != ip.Name {	×
163	// invalid ip or node ip, no need to handle it here	×
164	klog.V(3).Infof("port name %s is not equal to ip name %s", portName, ip.Name)	×
165	return nil	×
166	}	×
167
168	// not handle add the ip, which created in pod process, lsp created before ip
169	lsp, err := c.OVNNbClient.GetLogicalSwitchPort(portName, true)	×
170	if err != nil {	×
171	klog.Errorf("failed to list logical switch ports %s, %v", portName, err)	×
172	return err	×
173	}	×
174	if lsp != nil {	×
175	// port already exists means the ip already created	×
176	klog.V(3).Infof("ip %s is ready", portName)	×
177	return nil	×
178	}	×
179
180	// v6 ip address can not use upper case
181	if util.ContainsUppercase(ip.Spec.V6IPAddress) {	×
182	err := fmt.Errorf("ip %s v6 ip address %s can not contain upper case", ip.Name, ip.Spec.V6IPAddress)	×
183	klog.Error(err)	×
184	return err	×
185	}	×
186	v4IP, v6IP, mac, err := c.ipAcquireAddress(ip, subnet)	×
187	if err != nil {	×
188	err = fmt.Errorf("failed to acquire ip address %w", err)	×
189	klog.Error(err)	×
190	return err	×
191	}	×
192	ipStr := util.GetStringIP(v4IP, v6IP)	×
193	if err := c.createOrUpdateIPCR(ip.Name, ip.Spec.PodName, ipStr, mac, subnet.Name, ip.Spec.Namespace, ip.Spec.NodeName, ip.Spec.PodType); err != nil {	×
194	err = fmt.Errorf("failed to create ips CR %s.%s: %w", ip.Spec.PodName, ip.Spec.Namespace, err)	×
195	klog.Error(err)	×
196	return err	×
197	}	×
198	if ip.Labels == nil {	×
199	ip.Labels = map[string]string{}	×
200	}	×
201	if ip.Labels[util.IPReservedLabel] != "false" {	×
202	ip.Labels[util.IPReservedLabel] = "true"	×
203	patchPayloadTemplate := `[{ "op": "%s", "path": "/metadata/labels", "value": %s }]`	×
204	raw, err := json.Marshal(ip.Labels)	×
205	if err != nil {	×
206	klog.Error(err)	×
207	return err	×
208	}	×
209	op := "replace"	×
210	patchPayload := fmt.Sprintf(patchPayloadTemplate, op, raw)	×
211	if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), ip.Name,	×
212	types.JSONPatchType, []byte(patchPayload), metav1.PatchOptions{}); err != nil {	×
213	klog.Errorf("failed to patch label for ip %s, %v", ip.Name, err)	×
214	return err	×
215	}	×
216	}
217	return nil	×
218	}
219
220	func (c *Controller) handleUpdateIP(key string) error {	×
221	cachedIP, err := c.ipsLister.Get(key)	×
222	if err != nil {	×
223	if k8serrors.IsNotFound(err) {	×
224	return nil	×
225	}	×
226	klog.Error(err)	×
227	return err	×
228	}
229	if !cachedIP.DeletionTimestamp.IsZero() {	×
230	klog.Infof("handle deleting ip %s", cachedIP.Name)	×
231	subnet, err := c.subnetsLister.Get(cachedIP.Spec.Subnet)	×
232	if err != nil {	×
233	klog.Errorf("failed to get subnet %s: %v", cachedIP.Spec.Subnet, err)	×
234	return err	×
235	}	×
236	portName := cachedIP.Name	×
237	if isOvnSubnet(subnet) {	×
238	port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, true)	×
239	if err != nil {	×
240	klog.Errorf("failed to get logical switch port %s: %v", portName, err)	×
241	return err	×
242	}	×
243	if port != nil {	×
244	klog.Infof("delete ip cr lsp %s from switch %s", portName, subnet.Name)	×
245	if err := c.OVNNbClient.DeleteLogicalSwitchPort(portName); err != nil {	×
246	klog.Errorf("failed to delete ip cr lsp %s from switch %s: %v", portName, subnet.Name, err)	×
247	return err	×
248	}	×
249	klog.V(3).Infof("sync sg for deleted port %s", portName)	×
250	sgList, err := c.getPortSg(port)	×
251	if err != nil {	×
252	klog.Errorf("get port sg failed, %v", err)	×
253	return err	×
254	}	×
255	for _, sgName := range sgList {	×
256	if sgName != "" {	×
257	c.syncSgPortsQueue.Add(sgName)	×
258	}	×
259	}
260	}
261	}
262	podKey := fmt.Sprintf("%s/%s", cachedIP.Spec.Namespace, cachedIP.Spec.PodName)	×
263	klog.Infof("ip cr %s release ipam pod key %s from subnet %s", cachedIP.Name, podKey, cachedIP.Spec.Subnet)	×
264	c.ipam.ReleaseAddressByNic(podKey, portName, cachedIP.Spec.Subnet)	×
265	if err = c.handleDelIPFinalizer(cachedIP); err != nil {	×
266	klog.Errorf("failed to handle del ip finalizer %v", err)	×
267	return err	×
268	}	×
269	c.updateSubnetStatusQueue.Add(cachedIP.Spec.Subnet)	×
270	}
271	return nil	×
272	}
273
274	func (c Controller) handleDelIP(ip kubeovnv1.IP) error {	×
275	klog.Infof("deleting ip %s enqueue update status subnet %s", ip.Name, ip.Spec.Subnet)	×
276	c.updateSubnetStatusQueue.Add(ip.Spec.Subnet)	×
277	for _, as := range ip.Spec.AttachSubnets {	×
278	klog.V(3).Infof("enqueue update attach status for subnet %s", as)	×
279	c.updateSubnetStatusQueue.Add(as)	×
280	}	×
281	return nil	×
282	}
283
284	func (c *Controller) syncIPFinalizer(cl client.Client) error {	×
285	// migrate depreciated finalizer to new finalizer	×
286	ips := &kubeovnv1.IPList{}	×
287	return migrateFinalizers(cl, ips, func(i int) (client.Object, client.Object) {	×
288	if i < 0 \|\| i >= len(ips.Items) {	×
289	return nil, nil	×
290	}	×
291	return ips.Items[i].DeepCopy(), ips.Items[i].DeepCopy()	×
292	})
293	}
294
295	func (c Controller) handleAddIPFinalizer(cachedIP kubeovnv1.IP) error {	×
296	if !cachedIP.DeletionTimestamp.IsZero() \|\| len(cachedIP.GetFinalizers()) != 0 {	×
297	return nil	×
298	}	×
299	newIP := cachedIP.DeepCopy()	×
300	controllerutil.AddFinalizer(newIP, util.KubeOVNControllerFinalizer)	×
301	patch, err := util.GenerateMergePatchPayload(cachedIP, newIP)	×
302	if err != nil {	×
303	klog.Errorf("failed to generate patch payload for ip %s, %v", cachedIP.Name, err)	×
304	return err	×
305	}	×
306	if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), cachedIP.Name,	×
307	types.MergePatchType, patch, metav1.PatchOptions{}, ""); err != nil {	×
308	if k8serrors.IsNotFound(err) {	×
309	return nil	×
310	}	×
311	klog.Errorf("failed to add finalizer for ip %s, %v", cachedIP.Name, err)	×
312	return err	×
313	}
314	return nil	×
315	}
316
317	func (c Controller) handleDelIPFinalizer(cachedIP kubeovnv1.IP) error {	×
318	if len(cachedIP.GetFinalizers()) == 0 {	×
319	return nil	×
320	}	×
321	newIP := cachedIP.DeepCopy()	×
322	controllerutil.RemoveFinalizer(newIP, util.DepreciatedFinalizerName)	×
323	controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)	×
324	patch, err := util.GenerateMergePatchPayload(cachedIP, newIP)	×
325	if err != nil {	×
326	klog.Errorf("failed to generate patch payload for ip %s, %v", cachedIP.Name, err)	×
327	return err	×
328	}	×
329	if _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Patch(context.Background(), cachedIP.Name,	×
330	types.MergePatchType, patch, metav1.PatchOptions{}, ""); err != nil {	×
331	if k8serrors.IsNotFound(err) {	×
332	return nil	×
333	}	×
334	klog.Errorf("failed to remove finalizer from ip %s, %v", cachedIP.Name, err)	×
335	return err	×
336	}
337	return nil	×
338	}
339
340	func (c *Controller) acquireIPAddress(subnetName, name, nicName string) (string, string, string, error) {	×
341	var skippedAddrs []string	×
342	var v4ip, v6ip, mac string	×
343	checkConflict := true	×
344	var err error	×
345	for {	×
346	v4ip, v6ip, mac, err = c.ipam.GetRandomAddress(name, nicName, nil, subnetName, "", skippedAddrs, checkConflict)	×
347	if err != nil {	×
348	klog.Error(err)	×
349	return "", "", "", err	×
350	}	×
351
352	ipv4OK, ipv6OK, err := c.validatePodIP(name, subnetName, v4ip, v6ip)	×
353	if err != nil {	×
354	klog.Error(err)	×
355	return "", "", "", err	×
356	}	×
357
358	if ipv4OK && ipv6OK {	×
359	return v4ip, v6ip, mac, nil	×
360	}	×
361
362	if !ipv4OK {	×
363	skippedAddrs = append(skippedAddrs, v4ip)	×
364	}	×
365	if !ipv6OK {	×
366	skippedAddrs = append(skippedAddrs, v6ip)	×
367	}	×
368	}
369	}
370
371	func (c Controller) acquireStaticIPAddress(subnetName, name, nicName, ip string, macPointer string) (string, string, string, error) {	×
372	checkConflict := true	×
373	var v4ip, v6ip, mac string	×
374	var err error	×
375	for ipStr := range strings.SplitSeq(ip, ",") {	×
376	if net.ParseIP(ipStr) == nil {	×
377	return "", "", "", fmt.Errorf("failed to parse vip ip %s", ipStr)	×
378	}	×
379	}
380
381	if v4ip, v6ip, mac, err = c.ipam.GetStaticAddress(name, nicName, ip, macPointer, subnetName, checkConflict); err != nil {	×
382	klog.Errorf("failed to get static virtual ip '%s', mac '%s', subnet '%s', %v", ip, mac, subnetName, err)	×
383	return "", "", "", err	×
384	}	×
385	return v4ip, v6ip, mac, nil	×
386	}
387
388	func (c *Controller) createOrUpdateIPCR(ipCRName, podName, ip, mac, subnetName, ns, nodeName, podType string) error {	×
389	// `ipCRName`: pod or vm IP name must set ip CR name when creating ip CR	×
390	var key, ipName string	×
391	if ipCRName != "" {	×
392	// pod IP	×
393	key = podName	×
394	ipName = ipCRName	×
395	} else {	×
396	// node IP or interconn IP	×
397	switch {	×
398	case subnetName == c.config.NodeSwitch:	×
399	key = nodeName	×
400	ipName = util.NodeLspName(nodeName)	×
401	case strings.HasPrefix(podName, util.U2OInterconnName[0:19]):	×
402	key = podName // interconn IP name	×
403	ipName = podName	×
404	case strings.HasPrefix(podName, util.McastQuerierName[0:13]):	×
405	key = podName // mcast querier IP name	×
406	ipName = podName	×
407	}
408	}
409
410	var err error	×
411	var ipCR *kubeovnv1.IP	×
412	ipCR, err = c.ipsLister.Get(ipName)	×
413	if err != nil {	×
414	if !k8serrors.IsNotFound(err) {	×
415	err := fmt.Errorf("failed to get ip CR %s: %w", ipName, err)	×
416	klog.Error(err)	×
417	return err	×
418	}	×
419	// the returned pointer is not nil if the CR does not exist
420	ipCR = nil	×
421	}
422	if ipCR != nil && !ipCR.DeletionTimestamp.IsZero() {	×
423	// this ip is being deleted, no need to update	×
424	klog.Infof("enqueue update for removing finalizer to delete ip %s", ipCR.Name)	×
425	c.updateIPQueue.Add(ipCR.Name)	×
426	return nil	×
427	}	×
428	v4IP, v6IP := util.SplitStringIP(ip)	×
429	if ipCR == nil {	×
430	ipCR, err = c.config.KubeOvnClient.KubeovnV1().IPs().Create(context.Background(), &kubeovnv1.IP{	×
431	ObjectMeta: metav1.ObjectMeta{	×
432	Name: ipName,	×
433	Labels: map[string]string{	×
434	util.SubnetNameLabel: subnetName,	×
435	util.NodeNameLabel: nodeName,	×
436	subnetName: "",	×
437	util.IPReservedLabel: "false", // ip create with pod or node, ip not reserved	×
438	},	×
439	},	×
440	Spec: kubeovnv1.IPSpec{	×
441	PodName: key,	×
442	Subnet: subnetName,	×
443	NodeName: nodeName,	×
444	Namespace: ns,	×
445	IPAddress: ip,	×
446	V4IPAddress: v4IP,	×
447	V6IPAddress: v6IP,	×
448	MacAddress: mac,	×
449	AttachIPs: []string{},	×
450	AttachMacs: []string{},	×
451	AttachSubnets: []string{},	×
452	PodType: podType,	×
453	},	×
454	}, metav1.CreateOptions{})	×
455	if err != nil {	×
456	errMsg := fmt.Errorf("failed to create ip CR %s: %w", ipName, err)	×
457	klog.Error(errMsg)	×
458	return errMsg	×
459	}	×
460	} else {	×
461	newIPCR := ipCR.DeepCopy()	×
462	if newIPCR.Labels != nil {	×
463	newIPCR.Labels[util.SubnetNameLabel] = subnetName	×
464	newIPCR.Labels[util.NodeNameLabel] = nodeName	×
465	} else {	×
466	newIPCR.Labels = map[string]string{	×
467	util.SubnetNameLabel: subnetName,	×
468	util.NodeNameLabel: nodeName,	×
469	}	×
470	// update not touch IP Reserved Label	×
471	}	×
472	newIPCR.Spec.PodName = key	×
473	newIPCR.Spec.Namespace = ns	×
474	newIPCR.Spec.Subnet = subnetName	×
475	newIPCR.Spec.NodeName = nodeName	×
476	newIPCR.Spec.IPAddress = ip	×
477	newIPCR.Spec.V4IPAddress = v4IP	×
478	newIPCR.Spec.V6IPAddress = v6IP	×
479	newIPCR.Spec.MacAddress = mac	×
480	newIPCR.Spec.AttachIPs = []string{}	×
481	newIPCR.Spec.AttachMacs = []string{}	×
482	newIPCR.Spec.AttachSubnets = []string{}	×
483	newIPCR.Spec.PodType = podType	×
484	if maps.Equal(newIPCR.Labels, ipCR.Labels) && reflect.DeepEqual(newIPCR.Spec, ipCR.Spec) {	×
485	return nil	×
486	}	×
487
488	ipCR, err = c.config.KubeOvnClient.KubeovnV1().IPs().Update(context.Background(), newIPCR, metav1.UpdateOptions{})	×
489	if err != nil {	×
490	err := fmt.Errorf("failed to update ip CR %s: %w", ipCRName, err)	×
491	klog.Error(err)	×
492	return err	×
493	}	×
494	}
495	if err := c.handleAddIPFinalizer(ipCR); err != nil {	×
496	klog.Errorf("failed to handle add ip finalizer %v", err)	×
497	return err	×
498	}	×
499	return nil	×
500	}
501
502	func (c Controller) ipAcquireAddress(ip kubeovnv1.IP, subnet *kubeovnv1.Subnet) (string, string, string, error) {	×
503	key := fmt.Sprintf("%s/%s", ip.Spec.Namespace, ip.Spec.PodName)	×
504	portName := ovs.PodNameToPortName(ip.Spec.PodName, ip.Spec.Namespace, subnet.Spec.Provider)	×
505	ipStr := util.GetStringIP(ip.Spec.V4IPAddress, ip.Spec.V6IPAddress)	×
506		×
507	var v4IP, v6IP, mac string	×
508	var err error	×
509	var macPtr *string	×
510	if isOvnSubnet(subnet) {	×
511	if ip.Spec.MacAddress != "" {	×
512	macPtr = &ip.Spec.MacAddress	×
513	}	×
514	} else {	×
515	macPtr = ptr.To("")	×
516	}	×
517
518	if ipStr == "" {	×
519	// allocate address	×
520	v4IP, v6IP, mac, err = c.acquireIPAddress(subnet.Name, ip.Name, portName)	×
521	if err == nil {	×
522	return v4IP, v6IP, mac, err	×
523	}	×
524	err = fmt.Errorf("failed to get random address for ip %s, %w", ip.Name, err)	×
525	} else {	×
526	// static address	×
527	v4IP, v6IP, mac, err = c.acquireStaticAddress(key, portName, ipStr, macPtr, subnet.Name, true)	×
528	if err == nil {	×
529	return v4IP, v6IP, mac, nil	×
530	}	×
531	err = fmt.Errorf("failed to get static address for ip %s, %w", ip.Name, err)	×
532	}
533	klog.Error(err)	×
534	return "", "", "", err	×
535	}

kubeovn / kube-ovn / 17228354799

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