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

×

101

        var ipObj *kubeovnv1.IP

×

102

        switch t := obj.(type) {

×

103

        case *kubeovnv1.IP:

×

104

                ipObj = t

×

105

        case cache.DeletedFinalStateUnknown:

×

106

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

×

107

                if !ok {

×

108

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

×

109

                        return

×

110

×

111

                ipObj = ip

×

112

        default:

×

113

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

×

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

        if !ip.DeletionTimestamp.IsZero() {

×

136

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

×

137

                return nil

×

138

×

139

        if len(ip.Finalizers) != 0 {

×

140

                // finalizer already added, no need to handle it again

×

141

                return nil

×

142

×

144

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

×

145

×

146

        if ip.Spec.Subnet == "" {

×

UNCOV

147

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

×

148

                klog.Error(err)

×

149

                return err

×

150

×

152

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

×

UNCOV

153

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

×

154

                klog.Error(err)

×

155

                return err

×

156

×

158

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

×

159

        if err != nil {

×

UNCOV

160

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

×

161

                klog.Error(err)

×

162

                return err

×

163

×

165

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

×

166

        if portName != ip.Name {

×

UNCOV

167

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

×

UNCOV

168

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

×

169

                return nil

×

170

×

173

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

×

174

        if err != nil {

×

175

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

×

176

                return err

×

177

×

178

        if lsp != nil {

×

UNCOV

179

                // port already exists means the ip already created

×

UNCOV

180

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

×

181

                return nil

×

182

×

185

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

×

186

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

×

187

                klog.Error(err)

×

188

                return err

×

189

×

190

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

×

191

        if err != nil {

×

192

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

×

193

                klog.Error(err)

×

194

                return err

×

195

×

196

        ipStr := util.GetStringIP(v4IP, v6IP)

×

197

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

×

198

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

×

199

                klog.Error(err)

×

200

                return err

×

201

×

203

        ip = ip.DeepCopy()

×

204

        if ip.Labels == nil {

×

205

                ip.Labels = map[string]string{}

×

206

×

207

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

×

208

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

×

209

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

×

210

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

×

211

                if err != nil {

×

212

                        klog.Error(err)

×

213

                        return err

×

214

×

215

                op := "replace"

×

UNCOV

216

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

×

217

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

×

UNCOV

218

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

×

UNCOV

219

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

×

220

                        return err

×

221

×

223

        return nil

×

226

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

×

227

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

×

UNCOV

228

        if err != nil {

×

229

                if k8serrors.IsNotFound(err) {

×

230

                        return nil

×

231

×

232

                klog.Error(err)

×

233

                return err

×

235

        if !cachedIP.DeletionTimestamp.IsZero() {

×

236

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

×

237

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

×

238

                if err != nil {

×

239

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

×

240

                        return err

×

241

×

242

                portName := cachedIP.Name

×

243

                if isOvnSubnet(subnet) {

×

244

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

×

245

                        if err != nil {

×

246

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

×

247

                                return err

×

248

×

249

                        if port != nil {

×

250

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

×

251

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

×

252

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

×

253

                                        return err

×

254

×

255

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

×

256

                                sgList, err := c.getPortSg(port)

×

257

                                if err != nil {

×

258

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

×

UNCOV

259

                                        return err

×

UNCOV

260

×

UNCOV

261

                                for _, sgName := range sgList {

×

262

                                        if sgName != "" {

×

263

                                                c.syncSgPortsQueue.Add(sgName)

×

264

×

268

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

×

269

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

×

UNCOV

270

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

×

271

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

×

UNCOV

272

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

×

UNCOV

273

                        return err

×

274

×

275

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

×

277

        return nil

×

280

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

×

281

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

×

UNCOV

282

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

×

UNCOV

283

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

×

284

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

×

285

                c.updateSubnetStatusQueue.Add(as)

×

286

×

287

        return nil

×

290

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

×

291

        // migrate depreciated finalizer to new finalizer

×

UNCOV

292

        ips := &kubeovnv1.IPList{}

×

UNCOV

293

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

×

UNCOV

294

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

×

295

                        return nil, nil

×

296

×

297

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

×

301

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

×

302

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

×

303

                return nil

×

304

×

305

        newIP := cachedIP.DeepCopy()

×

306

        controllerutil.RemoveFinalizer(newIP, util.DepreciatedFinalizerName)

×

307

        controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

308

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

×

309

        if err != nil {

×

310

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

×

311

                return err

×

312

×

UNCOV

313

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

×

314

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

×

UNCOV

315

                if k8serrors.IsNotFound(err) {

×

UNCOV

316

                        return nil

×

317

×

318

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

×

319

                return err

×

321

        return nil

×

324

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

×

325

        var skippedAddrs []string

×

326

        var v4ip, v6ip, mac string

×

327

        checkConflict := true

×

328

        var err error

×

329

        for {

×

330

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

×

331

                if err != nil {

×

332

                        klog.Error(err)

×

333

                        return "", "", "", err

×

334

×

UNCOV

336

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

×

337

                if err != nil {

×

UNCOV

338

                        klog.Error(err)

×

UNCOV

339

                        return "", "", "", err

×

340

×

342

                if ipv4OK && ipv6OK {

×

343

                        return v4ip, v6ip, mac, nil

×

344

×

346

                if !ipv4OK {

×

347

                        skippedAddrs = append(skippedAddrs, v4ip)

×

348

×

349

                if !ipv6OK {

×

350

                        skippedAddrs = append(skippedAddrs, v6ip)

×

UNCOV

351

×

355

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

×

356

        checkConflict := true

×

UNCOV

357

        var v4ip, v6ip, mac string

×

358

        var err error

×

359

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

×

360

                if net.ParseIP(ipStr) == nil {

×

UNCOV

361

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

×

362

×

365

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

×

366

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

×

367

                return "", "", "", err

×

UNCOV

368

×

UNCOV

369

        return v4ip, v6ip, mac, nil

×

372

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

×

373

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

×

374

        var key, ipName string

×

375

        if ipCRName != "" {

×

376

                // pod IP

×

377

                key = podName

×

378

                ipName = ipCRName

×

UNCOV

379

        } else {

×

UNCOV

380

                // node IP or interconn IP

×

381

                switch {

×

382

                case subnetName == c.config.NodeSwitch:

×

383

                        key = nodeName

×

384

                        ipName = util.NodeLspName(nodeName)

×

385

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

×

UNCOV

386

                        key = podName // interconn IP name

×

UNCOV

387

                        ipName = podName

×

388

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

×

389

                        key = podName // mcast querier IP name

×

390

                        ipName = podName

×

394

        var err error

×

395

        var ipCR *kubeovnv1.IP

×

396

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

×

397

        if err != nil {

×

398

                if !k8serrors.IsNotFound(err) {

×

399

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

×

400

                        klog.Error(err)

×

401

                        return err

×

402

×

404

                ipCR = nil

×

406

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

×

UNCOV

407

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

×

UNCOV

408

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

×

UNCOV

409

                c.updateIPQueue.Add(ipCR.Name)

×

410

                return nil

×

411

×

412

        v4IP, v6IP := util.SplitStringIP(ip)

×

413

        if ipCR == nil {

×

414

                ipCR = &kubeovnv1.IP{

×

415

                        ObjectMeta: metav1.ObjectMeta{

×

416

                                Name: ipName,

×

417

                                Labels: map[string]string{

×

418

                                        util.SubnetNameLabel: subnetName,

×

UNCOV

419

                                        util.NodeNameLabel:   nodeName,

×

420

                                        subnetName:           "",

×

UNCOV

421

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

×

422

},

×

423

                                Finalizers: []string{util.KubeOVNControllerFinalizer},

×

424

},

×

425

                        Spec: kubeovnv1.IPSpec{

×

426

                                PodName:       key,

×

427

                                Subnet:        subnetName,

×

428

                                NodeName:      nodeName,

×

429

                                Namespace:     ns,

×

430

                                IPAddress:     ip,

×

431

                                V4IPAddress:   v4IP,

×

432

                                V6IPAddress:   v6IP,

×

433

                                MacAddress:    mac,

×

434

                                AttachIPs:     []string{},

×

435

                                AttachMacs:    []string{},

×

436

                                AttachSubnets: []string{},

×

437

                                PodType:       podType,

×

438

},

×

439

×

440

                if _, err = c.config.KubeOvnClient.KubeovnV1().IPs().Create(context.Background(), ipCR, metav1.CreateOptions{}); err != nil {

×

441

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

×

442

                        klog.Error(errMsg)

×

443

                        return errMsg

×

444

×

445

        } else {

×

446

                newIPCR := ipCR.DeepCopy()

×

447

                if newIPCR.Labels != nil {

×

448

                        newIPCR.Labels[util.SubnetNameLabel] = subnetName

×

449

                        newIPCR.Labels[util.NodeNameLabel] = nodeName

×

450

                } else {

×

451

                        newIPCR.Labels = map[string]string{

×

452

                                util.SubnetNameLabel: subnetName,

×

453

                                util.NodeNameLabel:   nodeName,

×

454

×

455

                        // update not touch IP Reserved Label

×

456

×

457

                newIPCR.Spec.PodName = key

×

458

                newIPCR.Spec.Namespace = ns

×

459

                newIPCR.Spec.Subnet = subnetName

×

460

                newIPCR.Spec.NodeName = nodeName

×

461

                newIPCR.Spec.IPAddress = ip

×

462

                newIPCR.Spec.V4IPAddress = v4IP

×

463

                newIPCR.Spec.V6IPAddress = v6IP

×

464

                newIPCR.Spec.MacAddress = mac

×

465

                newIPCR.Spec.AttachIPs = []string{}

×

466

                newIPCR.Spec.AttachMacs = []string{}

×

467

                newIPCR.Spec.AttachSubnets = []string{}

×

468

                newIPCR.Spec.PodType = podType

×

469

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

×

470

                        return nil

×

471

×

472

                controllerutil.AddFinalizer(newIPCR, util.KubeOVNControllerFinalizer)

×

473

×

474

                if _, err = c.config.KubeOvnClient.KubeovnV1().IPs().Update(context.Background(), newIPCR, metav1.UpdateOptions{}); err != nil {

×

475

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

×

476

                        klog.Error(err)

×

477

                        return err

×

478

×

480

        return nil

×

483

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

×

484

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

×

485

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

×

486

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

×

UNCOV

487

×

488

        var v4IP, v6IP, mac string

×

489

        var err error

×

490

        var macPtr *string

×

491

        if isOvnSubnet(subnet) {

×

492

                if ip.Spec.MacAddress != "" {

×

493

                        macPtr = &ip.Spec.MacAddress

×

UNCOV

494

×

495

        } else {

×

496

                macPtr = ptr.To("")

×

497

×

499

        if ipStr == "" {

×

UNCOV

500

                // allocate address

×

UNCOV

501

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

×

502

                if err == nil {

×

503

                        return v4IP, v6IP, mac, err

×

504

×

505

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

×

506

        } else {

×

507

                // static address

×

508

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

×

509

                if err == nil {

×

510

                        return v4IP, v6IP, mac, nil

×

511

×

512

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

×

514

        klog.Error(err)

×

515

        return "", "", "", err

×

kubeovn / kube-ovn / 17568569437

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

kubeovn / kube-ovn / 17568569437

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