kubeovn/kube-ovn | Build 21623048013 | 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 == "" {

×

147

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

×

148

                klog.Error(err)

×

149

                return err

×

150

×

152

        if ip.Spec.PodType != "" &&

×

153

                ip.Spec.PodType != util.KindVirtualMachine &&

×

154

                ip.Spec.PodType != util.KindStatefulSet {

×

155

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

×

156

                klog.Error(err)

×

157

                return err

×

158

×

160

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

×

161

        if err != nil {

×

162

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

×

163

                klog.Error(err)

×

164

                return err

×

165

×

167

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

×

168

        if portName != ip.Name {

×

169

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

×

170

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

×

171

                return nil

×

172

×

175

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

×

176

        if err != nil {

×

177

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

×

178

                return err

×

179

×

180

        if lsp != nil {

×

181

                // port already exists means the ip already created

×

182

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

×

183

                return nil

×

184

×

187

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

×

188

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

×

189

                klog.Error(err)

×

190

                return err

×

191

×

192

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

×

193

        if err != nil {

×

194

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

×

195

                klog.Error(err)

×

196

                return err

×

197

×

198

        ipStr := util.GetStringIP(v4IP, v6IP)

×

199

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

×

200

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

×

201

                klog.Error(err)

×

202

                return err

×

203

×

205

        ip = ip.DeepCopy()

×

206

        if ip.Labels == nil {

×

207

                ip.Labels = map[string]string{}

×

208

×

209

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

×

210

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

×

211

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

×

212

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

×

213

                if err != nil {

×

214

                        klog.Error(err)

×

215

                        return err

×

216

×

217

                op := "replace"

×

218

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

×

219

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

×

220

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

×

221

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

×

222

                        return err

×

223

×

225

        return nil

×

228

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

×

229

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

×

230

        if err != nil {

×

231

                if k8serrors.IsNotFound(err) {

×

232

                        return nil

×

233

×

234

                klog.Error(err)

×

235

                return err

×

237

        if !cachedIP.DeletionTimestamp.IsZero() {

×

238

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

×

239

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

×

240

                if err != nil {

×

241

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

×

242

                        return err

×

243

×

244

                portName := cachedIP.Name

×

245

                if isOvnSubnet(subnet) {

×

246

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

×

247

                        if err != nil {

×

248

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

×

249

                                return err

×

250

×

251

                        if port != nil {

×

252

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

×

253

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

×

254

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

×

255

                                        return err

×

256

×

257

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

×

258

                                sgList, err := c.getPortSg(port)

×

259

                                if err != nil {

×

260

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

×

261

                                        return err

×

262

×

263

                                for _, sgName := range sgList {

×

264

                                        if sgName != "" {

×

265

                                                c.syncSgPortsQueue.Add(sgName)

×

266

×

270

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

×

271

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

×

272

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

×

273

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

×

274

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

×

275

                        return err

×

276

×

277

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

×

279

        return nil

×

282

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

×

283

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

×

284

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

×

285

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

×

286

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

×

287

                c.updateSubnetStatusQueue.Add(as)

×

288

×

289

        return nil

×

292

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

×

NEW

293

        // migrate deprecated finalizer to new finalizer

×

294

        ips := &kubeovnv1.IPList{}

×

295

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

×

296

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

×

297

                        return nil, nil

×

298

×

299

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

×

303

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

×

304

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

×

305

                return nil

×

306

×

307

        newIP := cachedIP.DeepCopy()

×

NEW

308

        controllerutil.RemoveFinalizer(newIP, util.DeprecatedFinalizerName)

×

309

        controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

310

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

×

311

        if err != nil {

×

312

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

×

313

                return err

×

314

×

315

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

×

316

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

×

317

                if k8serrors.IsNotFound(err) {

×

318

                        return nil

×

319

×

320

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

×

321

                return err

×

323

        return nil

×

326

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

×

327

        var skippedAddrs []string

×

328

        var v4ip, v6ip, mac string

×

329

        checkConflict := true

×

330

        var err error

×

331

        for {

×

332

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

×

333

                if err != nil {

×

334

                        klog.Error(err)

×

335

                        return "", "", "", err

×

336

×

338

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

×

339

                if err != nil {

×

340

                        klog.Error(err)

×

341

                        return "", "", "", err

×

342

×

344

                if ipv4OK && ipv6OK {

×

345

                        return v4ip, v6ip, mac, nil

×

346

×

348

                if !ipv4OK {

×

349

                        skippedAddrs = append(skippedAddrs, v4ip)

×

350

×

351

                if !ipv6OK {

×

352

                        skippedAddrs = append(skippedAddrs, v6ip)

×

353

×

357

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

×

358

        checkConflict := true

×

359

        var v4ip, v6ip, mac string

×

360

        var err error

×

361

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

×

362

                if net.ParseIP(ipStr) == nil {

×

363

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

×

364

×

367

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

×

368

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

×

369

                return "", "", "", err

×

370

×

371

        return v4ip, v6ip, mac, nil

×

374

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

×

375

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

×

376

        var key, ipName string

×

377

        if ipCRName != "" {

×

378

                // pod IP

×

379

                key = podName

×

380

                ipName = ipCRName

×

381

        } else {

×

382

                // node IP or interconn IP

×

383

                switch {

×

384

                case subnetName == c.config.NodeSwitch:

×

385

                        key = nodeName

×

386

                        ipName = util.NodeLspName(nodeName)

×

387

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

×

388

                        key = podName // interconn IP name

×

389

                        ipName = podName

×

390

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

×

391

                        key = podName // mcast querier IP name

×

392

                        ipName = podName

×

396

        var err error

×

397

        var ipCR *kubeovnv1.IP

×

398

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

×

399

        if err != nil {

×

400

                if !k8serrors.IsNotFound(err) {

×

401

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

×

402

                        klog.Error(err)

×

403

                        return err

×

404

×

406

                ipCR = nil

×

408

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

×

409

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

×

410

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

×

411

                c.updateIPQueue.Add(ipCR.Name)

×

412

                return nil

×

413

×

414

        v4IP, v6IP := util.SplitStringIP(ip)

×

415

        if ipCR == nil {

×

416

                ipCR = &kubeovnv1.IP{

×

417

                        ObjectMeta: metav1.ObjectMeta{

×

418

                                Name: ipName,

×

419

                                Labels: map[string]string{

×

420

                                        util.SubnetNameLabel: subnetName,

×

421

                                        util.NodeNameLabel:   nodeName,

×

422

                                        subnetName:           "",

×

423

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

×

424

},

×

425

                                Finalizers: []string{util.KubeOVNControllerFinalizer},

×

426

},

×

427

                        Spec: kubeovnv1.IPSpec{

×

428

                                PodName:       key,

×

429

                                Subnet:        subnetName,

×

430

                                NodeName:      nodeName,

×

431

                                Namespace:     ns,

×

432

                                IPAddress:     ip,

×

433

                                V4IPAddress:   v4IP,

×

434

                                V6IPAddress:   v6IP,

×

435

                                MacAddress:    mac,

×

436

                                AttachIPs:     []string{},

×

437

                                AttachMacs:    []string{},

×

438

                                AttachSubnets: []string{},

×

439

                                PodType:       podType,

×

440

},

×

441

×

442

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

×

443

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

×

444

                        klog.Error(errMsg)

×

445

                        return errMsg

×

446

×

447

        } else {

×

448

                newIPCR := ipCR.DeepCopy()

×

449

                if newIPCR.Labels != nil {

×

450

                        newIPCR.Labels[util.SubnetNameLabel] = subnetName

×

451

                        newIPCR.Labels[util.NodeNameLabel] = nodeName

×

452

                } else {

×

453

                        newIPCR.Labels = map[string]string{

×

454

                                util.SubnetNameLabel: subnetName,

×

455

                                util.NodeNameLabel:   nodeName,

×

456

×

457

                        // update not touch IP Reserved Label

×

458

×

459

                newIPCR.Spec.PodName = key

×

460

                newIPCR.Spec.Namespace = ns

×

461

                newIPCR.Spec.Subnet = subnetName

×

462

                newIPCR.Spec.NodeName = nodeName

×

463

                newIPCR.Spec.IPAddress = ip

×

464

                newIPCR.Spec.V4IPAddress = v4IP

×

465

                newIPCR.Spec.V6IPAddress = v6IP

×

466

                newIPCR.Spec.MacAddress = mac

×

467

                newIPCR.Spec.AttachIPs = []string{}

×

468

                newIPCR.Spec.AttachMacs = []string{}

×

469

                newIPCR.Spec.AttachSubnets = []string{}

×

470

                newIPCR.Spec.PodType = podType

×

471

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

×

472

                        return nil

×

473

×

474

                controllerutil.AddFinalizer(newIPCR, util.KubeOVNControllerFinalizer)

×

475

×

476

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

×

477

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

×

478

                        klog.Error(err)

×

479

                        return err

×

480

×

482

        return nil

×

485

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

×

486

        key := cache.NewObjectName(ip.Spec.Namespace, ip.Spec.PodName).String()

×

487

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

×

488

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

×

489

×

490

        var v4IP, v6IP, mac string

×

491

        var err error

×

492

        var macPtr *string

×

493

        if isOvnSubnet(subnet) {

×

494

                if ip.Spec.MacAddress != "" {

×

495

                        macPtr = &ip.Spec.MacAddress

×

496

×

497

        } else {

×

498

                macPtr = ptr.To("")

×

499

×

501

        if ipStr == "" {

×

502

                // allocate address

×

503

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

×

504

                if err == nil {

×

505

                        return v4IP, v6IP, mac, err

×

506

×

507

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

×

508

        } else {

×

509

                // static address

×

510

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

×

511

                if err == nil {

×

512

                        return v4IP, v6IP, mac, nil

×

513

×

514

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

×

516

        klog.Error(err)

×

517

        return "", "", "", err

×

kubeovn / kube-ovn / 21623048013

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

kubeovn / kube-ovn / 21623048013

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