kubeovn/kube-ovn | Build 15602700125 | 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)

×

NEW

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)

×

NEW

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)

×

NEW

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)

×

NEW

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)

×

NEW

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)

×

NEW

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)

×

NEW

78

                        return

×

79

×

80

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

×

81

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

×

NEW

82

                        return

×

UNCOV

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) {

×

101

        ipObj := obj.(*kubeovnv1.IP)

×

102

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

×

103

                return

×

104

×

106

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

×

107

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

×

108

        c.delIPQueue.Add(ipObj)

×

111

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

×

112

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

×

113

        if err != nil {

×

114

                if k8serrors.IsNotFound(err) {

×

115

                        return nil

×

116

×

117

                klog.Error(err)

×

118

                return err

×

120

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

×

121

        if !ip.DeletionTimestamp.IsZero() {

×

122

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

×

123

                return nil

×

124

×

125

        if ip.Spec.Subnet == "" {

×

126

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

×

127

                klog.Error(err)

×

128

                return err

×

129

×

130

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

×

131

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

×

132

                klog.Error(err)

×

133

                return err

×

134

×

136

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

×

137

        if err != nil {

×

138

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

×

139

                klog.Error(err)

×

140

                return err

×

141

×

143

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

×

144

        if portName != ip.Name {

×

145

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

×

146

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

×

147

                return nil

×

148

×

151

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

×

152

        if err != nil {

×

153

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

×

154

                return err

×

155

×

156

        if lsp != nil {

×

157

                // port already exists means the ip already created

×

158

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

×

159

                return nil

×

160

×

163

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

×

164

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

×

165

                klog.Error(err)

×

166

                return err

×

167

×

168

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

×

169

        if err != nil {

×

170

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

×

171

                klog.Error(err)

×

172

                return err

×

173

×

174

        ipStr := util.GetStringIP(v4IP, v6IP)

×

175

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

×

176

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

×

177

                klog.Error(err)

×

178

                return err

×

179

×

180

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

×

181

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

×

182

                if err != nil {

×

183

                        if k8serrors.IsNotFound(err) {

×

184

                                return nil

×

185

×

186

                        klog.Error(err)

×

187

                        return err

×

189

                ip = cachedIP.DeepCopy()

×

190

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

×

191

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

×

192

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

×

193

                if err != nil {

×

194

                        klog.Error(err)

×

195

                        return err

×

196

×

197

                op := "replace"

×

198

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

×

199

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

×

200

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

×

201

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

×

202

                        return err

×

203

×

205

        return nil

×

208

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

×

209

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

×

210

        if err != nil {

×

211

                if k8serrors.IsNotFound(err) {

×

212

                        return nil

×

213

×

214

                klog.Error(err)

×

215

                return err

×

217

        if !cachedIP.DeletionTimestamp.IsZero() {

×

218

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

×

219

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

×

220

                if err != nil {

×

221

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

×

222

                        return err

×

223

×

224

                portName := cachedIP.Name

×

225

                if isOvnSubnet(subnet) {

×

226

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

×

227

                        if err != nil {

×

228

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

×

229

                                return err

×

230

×

231

                        if port != nil {

×

232

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

×

233

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

×

234

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

×

235

                                        return err

×

236

×

237

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

×

238

                                sgList, err := c.getPortSg(port)

×

239

                                if err != nil {

×

240

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

×

241

                                        return err

×

242

×

243

                                for _, sgName := range sgList {

×

244

                                        if sgName != "" {

×

245

                                                c.syncSgPortsQueue.Add(sgName)

×

246

×

250

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

×

251

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

×

252

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

×

253

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

×

254

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

×

255

                        return err

×

256

×

257

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

×

259

        return nil

×

262

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

×

263

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

×

264

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

×

265

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

×

266

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

×

267

                c.updateSubnetStatusQueue.Add(as)

×

268

×

269

        return nil

×

272

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

×

273

        // migrate depreciated finalizer to new finalizer

×

274

        ips := &kubeovnv1.IPList{}

×

275

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

×

276

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

×

277

                        return nil, nil

×

278

×

279

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

×

283

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

×

284

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

×

285

                return nil

×

286

×

287

        newIP := cachedIP.DeepCopy()

×

288

        controllerutil.AddFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

289

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

×

290

        if err != nil {

×

291

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

×

292

                return err

×

293

×

294

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

×

295

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

×

296

                if k8serrors.IsNotFound(err) {

×

297

                        return nil

×

298

×

299

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

×

300

                return err

×

302

        return nil

×

305

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

×

306

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

×

307

                return nil

×

308

×

309

        newIP := cachedIP.DeepCopy()

×

310

        controllerutil.RemoveFinalizer(newIP, util.DepreciatedFinalizerName)

×

311

        controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

312

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

×

313

        if err != nil {

×

314

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

×

315

                return err

×

316

×

317

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

×

318

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

×

319

                if k8serrors.IsNotFound(err) {

×

320

                        return nil

×

321

×

322

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

×

323

                return err

×

325

        return nil

×

328

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

×

329

        var skippedAddrs []string

×

330

        var v4ip, v6ip, mac string

×

331

        checkConflict := true

×

332

        var err error

×

333

        for {

×

334

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

×

335

                if err != nil {

×

336

                        klog.Error(err)

×

337

                        return "", "", "", err

×

338

×

340

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

×

341

                if err != nil {

×

342

                        klog.Error(err)

×

343

                        return "", "", "", err

×

344

×

346

                if ipv4OK && ipv6OK {

×

347

                        return v4ip, v6ip, mac, nil

×

348

×

350

                if !ipv4OK {

×

351

                        skippedAddrs = append(skippedAddrs, v4ip)

×

352

×

353

                if !ipv6OK {

×

354

                        skippedAddrs = append(skippedAddrs, v6ip)

×

355

×

359

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

×

360

        checkConflict := true

×

361

        var v4ip, v6ip, mac string

×

362

        var err error

×

363

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

×

364

                if net.ParseIP(ipStr) == nil {

×

365

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

×

366

×

369

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

×

370

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

×

371

                return "", "", "", err

×

372

×

373

        return v4ip, v6ip, mac, nil

×

376

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

×

377

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

×

378

        var key, ipName string

×

379

        if ipCRName != "" {

×

380

                // pod IP

×

381

                key = podName

×

382

                ipName = ipCRName

×

383

        } else {

×

384

                // node IP or interconn IP

×

385

                switch {

×

386

                case subnetName == c.config.NodeSwitch:

×

387

                        key = nodeName

×

388

                        ipName = util.NodeLspName(nodeName)

×

389

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

×

390

                        key = podName // interconn IP name

×

391

                        ipName = podName

×

392

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

×

393

                        key = podName // mcast querier IP name

×

394

                        ipName = podName

×

398

        var err error

×

399

        var ipCR *kubeovnv1.IP

×

400

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

×

401

        if err != nil {

×

402

                if !k8serrors.IsNotFound(err) {

×

403

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

×

404

                        klog.Error(err)

×

405

                        return err

×

406

×

408

                ipCR = nil

×

410

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

×

411

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

×

412

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

×

413

                c.updateIPQueue.Add(ipCR.Name)

×

414

                return nil

×

415

×

416

        v4IP, v6IP := util.SplitStringIP(ip)

×

417

        if ipCR == nil {

×

418

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

×

419

                        ObjectMeta: metav1.ObjectMeta{

×

420

                                Name: ipName,

×

421

                                Labels: map[string]string{

×

422

                                        util.SubnetNameLabel: subnetName,

×

423

                                        util.NodeNameLabel:   nodeName,

×

424

                                        subnetName:           "",

×

425

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

×

426

},

×

427

},

×

428

                        Spec: kubeovnv1.IPSpec{

×

429

                                PodName:       key,

×

430

                                Subnet:        subnetName,

×

431

                                NodeName:      nodeName,

×

432

                                Namespace:     ns,

×

433

                                IPAddress:     ip,

×

434

                                V4IPAddress:   v4IP,

×

435

                                V6IPAddress:   v6IP,

×

436

                                MacAddress:    mac,

×

437

                                AttachIPs:     []string{},

×

438

                                AttachMacs:    []string{},

×

439

                                AttachSubnets: []string{},

×

440

                                PodType:       podType,

×

441

},

×

442

                }, metav1.CreateOptions{})

×

443

                if err != nil {

×

444

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

×

445

                        klog.Error(errMsg)

×

446

                        return errMsg

×

447

×

448

        } else {

×

449

                newIPCR := ipCR.DeepCopy()

×

450

                if newIPCR.Labels != nil {

×

451

                        newIPCR.Labels[util.SubnetNameLabel] = subnetName

×

452

                        newIPCR.Labels[util.NodeNameLabel] = nodeName

×

453

                } else {

×

454

                        newIPCR.Labels = map[string]string{

×

455

                                util.SubnetNameLabel: subnetName,

×

456

                                util.NodeNameLabel:   nodeName,

×

457

×

458

                        // update not touch IP Reserved Label

×

459

×

460

                newIPCR.Spec.PodName = key

×

461

                newIPCR.Spec.Namespace = ns

×

462

                newIPCR.Spec.Subnet = subnetName

×

463

                newIPCR.Spec.NodeName = nodeName

×

464

                newIPCR.Spec.IPAddress = ip

×

465

                newIPCR.Spec.V4IPAddress = v4IP

×

466

                newIPCR.Spec.V6IPAddress = v6IP

×

467

                newIPCR.Spec.MacAddress = mac

×

468

                newIPCR.Spec.AttachIPs = []string{}

×

469

                newIPCR.Spec.AttachMacs = []string{}

×

470

                newIPCR.Spec.AttachSubnets = []string{}

×

471

                newIPCR.Spec.PodType = podType

×

472

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

×

473

                        return nil

×

474

×

476

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

×

477

                if err != nil {

×

478

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

×

479

                        klog.Error(err)

×

480

                        return err

×

481

×

483

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

×

484

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

×

485

                return err

×

486

×

487

        return nil

×

490

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

×

491

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

×

492

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

×

493

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

×

494

×

495

        var v4IP, v6IP, mac string

×

496

        var err error

×

497

        var macPtr *string

×

498

        if isOvnSubnet(subnet) {

×

499

                if ip.Spec.MacAddress != "" {

×

500

                        macPtr = &ip.Spec.MacAddress

×

501

×

502

        } else {

×

503

                macPtr = ptr.To("")

×

504

×

506

        if ipStr == "" {

×

507

                // allocate address

×

508

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

×

509

                if err == nil {

×

510

                        return v4IP, v6IP, mac, err

×

511

×

512

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

×

513

        } else {

×

514

                // static address

×

515

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

×

516

                if err == nil {

×

517

                        return v4IP, v6IP, mac, nil

×

518

×

519

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

×

521

        klog.Error(err)

×

522

        return "", "", "", err

×

kubeovn / kube-ovn / 15602700125

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

kubeovn / kube-ovn / 15602700125

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