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

28

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

×

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

×

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

×

52

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

×

53

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

×

54

×

55

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

×

56

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

×

57

×

58

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

×

59

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

×

60

×

61

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

×

62

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

×

63

×

64

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

×

65

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

×

66

×

67

        if oldIP.Spec.V6IPAddress != "" {

×

68

                // v6 ip address can not use upper case

×

69

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

×

70

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

×

71

                        klog.Error(err)

×

72

×

73

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

×

74

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

×

75

×

77

        if !newIP.DeletionTimestamp.IsZero() {

×

78

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

×

79

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

×

80

                c.updateIPQueue.Add(key)

×

81

                return

×

82

×

83

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

×

84

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

×

85

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

×

86

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

×

87

                        c.updateSubnetStatusQueue.Add(as)

×

88

×

92

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

×

93

        ipObj := obj.(*kubeovnv1.IP)

×

94

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

×

95

                return

×

96

×

98

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

×

99

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

×

100

        c.delIPQueue.Add(ipObj)

×

103

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

×

104

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

×

105

        if err != nil {

×

106

                if k8serrors.IsNotFound(err) {

×

107

                        return nil

×

108

×

109

                klog.Error(err)

×

110

                return err

×

112

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

×

113

        if ip.Spec.Subnet == "" {

×

114

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

×

115

                klog.Error(err)

×

116

                return err

×

117

×

118

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

×

119

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

×

120

                klog.Error(err)

×

121

                return err

×

122

×

124

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

×

125

        if err != nil {

×

126

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

×

127

                klog.Error(err)

×

128

                return err

×

129

×

131

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

×

132

        if portName != ip.Name {

×

133

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

×

134

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

×

135

                return nil

×

136

×

139

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

×

140

        if err != nil {

×

141

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

×

142

                return err

×

143

×

144

        if lsp != nil {

×

145

                // port already exists means the ip already created

×

146

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

×

147

                return nil

×

148

×

151

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

×

152

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

×

153

                klog.Error(err)

×

154

                return err

×

155

×

156

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

×

157

        if err != nil {

×

158

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

×

159

                klog.Error(err)

×

160

                return err

×

161

×

162

        ipStr := util.GetStringIP(v4IP, v6IP)

×

163

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

×

164

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

×

165

                klog.Error(err)

×

166

                return err

×

167

×

168

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

×

169

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

×

170

                if err != nil {

×

171

                        if k8serrors.IsNotFound(err) {

×

172

                                return nil

×

173

×

174

                        klog.Error(err)

×

175

                        return err

×

177

                ip = cachedIP.DeepCopy()

×

178

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

×

179

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

×

180

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

×

181

                if err != nil {

×

182

                        klog.Error(err)

×

183

                        return err

×

184

×

185

                op := "replace"

×

186

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

×

187

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

×

188

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

×

189

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

×

190

                        return err

×

191

×

193

        return nil

×

196

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

×

197

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

×

198

        if err != nil {

×

199

                if k8serrors.IsNotFound(err) {

×

200

                        return nil

×

201

×

202

                klog.Error(err)

×

203

                return err

×

205

        if !cachedIP.DeletionTimestamp.IsZero() {

×

206

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

×

207

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

×

208

                if err != nil {

×

209

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

×

210

                        return err

×

211

×

212

                if isOvnSubnet(subnet) {

×

213

                        portName := cachedIP.Name

×

214

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

×

215

                        if err != nil {

×

216

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

×

217

                                return err

×

218

×

219

                        if port != nil {

×

220

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

×

221

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

×

222

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

×

223

                                        return err

×

224

×

225

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

×

226

                                sgList, err := c.getPortSg(port)

×

227

                                if err != nil {

×

228

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

×

229

                                        return err

×

230

×

231

                                for _, sgName := range sgList {

×

232

                                        if sgName != "" {

×

233

                                                c.syncSgPortsQueue.Add(sgName)

×

234

×

238

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

×

239

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

×

240

                c.ipam.ReleaseAddressByPod(podKey, cachedIP.Spec.Subnet)

×

241

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

×

242

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

×

243

                        return err

×

244

×

245

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

×

247

        return nil

×

250

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

×

251

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

×

252

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

×

253

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

×

254

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

×

255

                c.updateSubnetStatusQueue.Add(as)

×

256

×

257

        return nil

×

260

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

×

261

        // migrate depreciated finalizer to new finalizer

×

262

        ips := &kubeovnv1.IPList{}

×

263

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

×

264

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

×

265

                        return nil, nil

×

266

×

267

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

×

271

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

×

272

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

×

273

                return nil

×

274

×

275

        newIP := cachedIP.DeepCopy()

×

276

        controllerutil.AddFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

277

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

×

278

        if err != nil {

×

279

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

×

280

                return err

×

281

×

282

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

×

283

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

×

284

                if k8serrors.IsNotFound(err) {

×

285

                        return nil

×

286

×

287

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

×

288

                return err

×

290

        return nil

×

293

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

×

294

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

×

295

                return nil

×

296

×

297

        newIP := cachedIP.DeepCopy()

×

298

        controllerutil.RemoveFinalizer(newIP, util.DepreciatedFinalizerName)

×

299

        controllerutil.RemoveFinalizer(newIP, util.KubeOVNControllerFinalizer)

×

300

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

×

301

        if err != nil {

×

302

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

×

303

                return err

×

304

×

305

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

×

306

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

×

307

                if k8serrors.IsNotFound(err) {

×

308

                        return nil

×

309

×

310

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

×

311

                return err

×

313

        return nil

×

316

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

×

317

        var skippedAddrs []string

×

318

        var v4ip, v6ip, mac string

×

319

        checkConflict := true

×

320

        var err error

×

321

        for {

×

322

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

×

323

                if err != nil {

×

324

                        klog.Error(err)

×

325

                        return "", "", "", err

×

326

×

328

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

×

329

                if err != nil {

×

330

                        klog.Error(err)

×

331

                        return "", "", "", err

×

332

×

334

                if ipv4OK && ipv6OK {

×

335

                        return v4ip, v6ip, mac, nil

×

336

×

338

                if !ipv4OK {

×

339

                        skippedAddrs = append(skippedAddrs, v4ip)

×

340

×

341

                if !ipv6OK {

×

342

                        skippedAddrs = append(skippedAddrs, v6ip)

×

343

×

NEW

347

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

×

348

        checkConflict := true

×

349

        var v4ip, v6ip, mac string

×

350

        var err error

×

351

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

×

352

                if net.ParseIP(ipStr) == nil {

×

353

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

×

354

×

NEW

357

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

×

358

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

×

359

                return "", "", "", err

×

360

×

361

        return v4ip, v6ip, mac, nil

×

364

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

×

365

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

×

366

        var key, ipName string

×

367

        if ipCRName != "" {

×

368

                // pod IP

×

369

                key = podName

×

370

                ipName = ipCRName

×

371

        } else {

×

372

                // node IP or interconn IP

×

373

                switch {

×

374

                case subnetName == c.config.NodeSwitch:

×

375

                        key = nodeName

×

376

                        ipName = util.NodeLspName(nodeName)

×

377

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

×

378

                        key = podName // interconn IP name

×

379

                        ipName = podName

×

380

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

×

381

                        key = podName // mcast querier IP name

×

382

                        ipName = podName

×

386

        var err error

×

387

        var ipCR *kubeovnv1.IP

×

388

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

×

389

        if err != nil {

×

390

                if !k8serrors.IsNotFound(err) {

×

391

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

×

392

                        klog.Error(err)

×

393

                        return err

×

394

×

396

                ipCR = nil

×

399

        v4IP, v6IP := util.SplitStringIP(ip)

×

400

        if ipCR == nil {

×

401

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

×

402

                        ObjectMeta: metav1.ObjectMeta{

×

403

                                Name: ipName,

×

404

                                Labels: map[string]string{

×

405

                                        util.SubnetNameLabel: subnetName,

×

406

                                        util.NodeNameLabel:   nodeName,

×

407

                                        subnetName:           "",

×

408

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

×

409

},

×

410

},

×

411

                        Spec: kubeovnv1.IPSpec{

×

412

                                PodName:       key,

×

413

                                Subnet:        subnetName,

×

414

                                NodeName:      nodeName,

×

415

                                Namespace:     ns,

×

416

                                IPAddress:     ip,

×

417

                                V4IPAddress:   v4IP,

×

418

                                V6IPAddress:   v6IP,

×

419

                                MacAddress:    mac,

×

420

                                AttachIPs:     []string{},

×

421

                                AttachMacs:    []string{},

×

422

                                AttachSubnets: []string{},

×

423

                                PodType:       podType,

×

424

},

×

425

                }, metav1.CreateOptions{})

×

426

                if err != nil {

×

427

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

×

428

                        klog.Error(errMsg)

×

429

                        return errMsg

×

430

×

431

        } else {

×

432

                newIPCR := ipCR.DeepCopy()

×

433

                if newIPCR.Labels != nil {

×

434

                        newIPCR.Labels[util.SubnetNameLabel] = subnetName

×

435

                        newIPCR.Labels[util.NodeNameLabel] = nodeName

×

436

                } else {

×

437

                        newIPCR.Labels = map[string]string{

×

438

                                util.SubnetNameLabel: subnetName,

×

439

                                util.NodeNameLabel:   nodeName,

×

440

×

441

                        // update not touch IP Reserved Label

×

442

×

443

                newIPCR.Spec.PodName = key

×

444

                newIPCR.Spec.Namespace = ns

×

445

                newIPCR.Spec.Subnet = subnetName

×

446

                newIPCR.Spec.NodeName = nodeName

×

447

                newIPCR.Spec.IPAddress = ip

×

448

                newIPCR.Spec.V4IPAddress = v4IP

×

449

                newIPCR.Spec.V6IPAddress = v6IP

×

450

                newIPCR.Spec.MacAddress = mac

×

451

                newIPCR.Spec.AttachIPs = []string{}

×

452

                newIPCR.Spec.AttachMacs = []string{}

×

453

                newIPCR.Spec.AttachSubnets = []string{}

×

454

                newIPCR.Spec.PodType = podType

×

455

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

×

456

                        return nil

×

457

×

459

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

×

460

                if err != nil {

×

461

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

×

462

                        klog.Error(err)

×

463

                        return err

×

464

×

466

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

×

467

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

×

468

                return err

×

469

×

470

        return nil

×

473

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

×

474

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

×

475

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

×

476

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

×

477

×

478

        var v4IP, v6IP, mac string

×

479

        var err error

×

480

        var macPtr *string

×

481

        if isOvnSubnet(subnet) {

×

482

                if ip.Spec.MacAddress != "" {

×

483

                        macPtr = &ip.Spec.MacAddress

×

484

×

485

        } else {

×

486

                macPtr = ptr.To("")

×

487

×

489

        if ipStr == "" {

×

490

                // allocate address

×

491

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

×

492

                if err == nil {

×

493

                        return v4IP, v6IP, mac, err

×

494

×

495

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

×

496

        } else {

×

497

                // static address

×

498

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

×

499

                if err == nil {

×

500

                        return v4IP, v6IP, mac, nil

×

501

×

502

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

×

504

        klog.Error(err)

×

505

        return "", "", "", err

×

kubeovn / kube-ovn / 14505803448

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

kubeovn / kube-ovn / 14505803448

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