kubeovn/kube-ovn | Build 13425935899 | pkg/controller/vpc.go | Coveralls

30

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

×

31

        vpc := obj.(*kubeovnv1.Vpc)

×

32

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

×

33

        if _, ok := vpc.Labels[util.VpcExternalLabel]; !ok {

×

34

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

×

35

                c.addOrUpdateVpcQueue.Add(key)

×

36

×

39

func vpcBFDPortChanged(oldObj, newObj *kubeovnv1.BFDPort) bool {

×

40

        if oldObj == nil && newObj == nil {

×

41

                return false

×

42

×

43

        if oldObj == nil || newObj == nil {

×

44

                return true

×

45

×

46

        return oldObj.Enabled != newObj.Enabled || oldObj.IP != newObj.IP || !reflect.DeepEqual(oldObj.NodeSelector, newObj.NodeSelector)

×

49

func (c *Controller) enqueueUpdateVpc(oldObj, newObj interface{}) {

×

50

        oldVpc := oldObj.(*kubeovnv1.Vpc)

×

51

        newVpc := newObj.(*kubeovnv1.Vpc)

×

52

×

53

        if !newVpc.DeletionTimestamp.IsZero() ||

×

54

                !slices.Equal(oldVpc.Spec.Namespaces, newVpc.Spec.Namespaces) ||

×

55

                !reflect.DeepEqual(oldVpc.Spec.StaticRoutes, newVpc.Spec.StaticRoutes) ||

×

56

                !reflect.DeepEqual(oldVpc.Spec.PolicyRoutes, newVpc.Spec.PolicyRoutes) ||

×

57

                !reflect.DeepEqual(oldVpc.Spec.VpcPeerings, newVpc.Spec.VpcPeerings) ||

×

58

                !maps.Equal(oldVpc.Annotations, newVpc.Annotations) ||

×

59

                !slices.Equal(oldVpc.Spec.ExtraExternalSubnets, newVpc.Spec.ExtraExternalSubnets) ||

×

60

                oldVpc.Spec.EnableExternal != newVpc.Spec.EnableExternal ||

×

61

                oldVpc.Spec.EnableBfd != newVpc.Spec.EnableBfd ||

×

62

                vpcBFDPortChanged(oldVpc.Spec.BFDPort, newVpc.Spec.BFDPort) ||

×

63

                oldVpc.Labels[util.VpcExternalLabel] != newVpc.Labels[util.VpcExternalLabel] {

×

64

                // TODO:// label VpcExternalLabel replace with spec enable external

×

65

×

66

                if newVpc.Annotations == nil {

×

67

                        newVpc.Annotations = make(map[string]string)

×

68

×

69

                newVpc.Annotations[util.VpcLastPolicies] = convertPolicies(oldVpc.Spec.PolicyRoutes)

×

70

×

71

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

×

72

                klog.Infof("enqueue update vpc %s", key)

×

73

                c.addOrUpdateVpcQueue.Add(key)

×

77

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

×

78

        vpc := obj.(*kubeovnv1.Vpc)

×

79

        if _, ok := vpc.Labels[util.VpcExternalLabel]; !vpc.Status.Default || !ok {

×

80

                klog.V(3).Infof("enqueue delete vpc %s", vpc.Name)

×

81

                c.delVpcQueue.Add(vpc)

×

82

×

85

func (c *Controller) handleDelVpc(vpc *kubeovnv1.Vpc) error {

×

86

        c.vpcKeyMutex.LockKey(vpc.Name)

×

87

        defer func() { _ = c.vpcKeyMutex.UnlockKey(vpc.Name) }()

×

88

        klog.Infof("handle delete vpc %s", vpc.Name)

×

89

×

90

        // should delete vpc subnets first

×

91

        var err error

×

92

        for _, subnet := range vpc.Status.Subnets {

×

93

                if _, err = c.subnetsLister.Get(subnet); err != nil {

×

94

                        if k8serrors.IsNotFound(err) {

×

95

                                continue

×

97

                        err = fmt.Errorf("failed to get subnet %s for vpc %s: %w", subnet, vpc.Name, err)

×

98

                } else {

×

99

                        err = fmt.Errorf("failed to delete vpc %s, please delete subnet %s first", vpc.Name, subnet)

×

100

×

101

                klog.Error(err)

×

102

                return err

×

105

        if err := c.deleteVpcLb(vpc); err != nil {

×

106

                klog.Error(err)

×

107

                return err

×

108

×

110

        if err := c.handleDelVpcExternalSubnet(vpc.Name, c.config.ExternalGatewaySwitch); err != nil {

×

111

                klog.Errorf("failed to delete external connection for vpc %s, error %v", vpc.Name, err)

×

112

                return err

×

113

×

115

        for _, subnet := range vpc.Status.ExtraExternalSubnets {

×

116

                klog.Infof("disconnect external network %s to vpc %s", subnet, vpc.Name)

×

117

                if err := c.handleDelVpcExternalSubnet(vpc.Name, subnet); err != nil {

×

118

                        klog.Error(err)

×

119

                        return err

×

120

×

123

        if err := c.deleteVpcRouter(vpc.Status.Router); err != nil {

×

124

                klog.Error(err)

×

125

                return err

×

126

×

128

        return nil

×

131

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

×

132

        c.vpcKeyMutex.LockKey(key)

×

133

        defer func() { _ = c.vpcKeyMutex.UnlockKey(key) }()

×

134

        klog.Infof("handle status update for vpc %s", key)

×

135

×

136

        cachedVpc, err := c.vpcsLister.Get(key)

×

137

        if err != nil {

×

138

                if k8serrors.IsNotFound(err) {

×

139

                        return nil

×

140

×

141

                klog.Error(err)

×

142

                return err

×

144

        vpc := cachedVpc.DeepCopy()

×

145

×

146

        subnets, defaultSubnet, err := c.getVpcSubnets(vpc)

×

147

        if err != nil {

×

148

                klog.Error(err)

×

149

                return err

×

150

×

152

        change := false

×

153

        if vpc.Status.DefaultLogicalSwitch != defaultSubnet {

×

154

                change = true

×

155

×

157

        vpc.Status.DefaultLogicalSwitch = defaultSubnet

×

158

        vpc.Status.Subnets = subnets

×

159

        if !vpc.Spec.BFDPort.IsEnabled() && !vpc.Status.BFDPort.IsEmpty() {

×

160

                vpc.Status.BFDPort.Clear()

×

161

×

162

        bytes, err := vpc.Status.Bytes()

×

163

        if err != nil {

×

164

                klog.Error(err)

×

165

                return err

×

166

×

168

        vpc, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(), vpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status")

×

169

        if err != nil {

×

170

                klog.Error(err)

×

171

                return err

×

172

×

173

        if change {

×

174

                for _, ns := range vpc.Spec.Namespaces {

×

175

                        c.addNamespaceQueue.Add(ns)

×

176

×

179

        natGws, err := c.vpcNatGatewayLister.List(labels.Everything())

×

180

        if err != nil {

×

181

                klog.Error(err)

×

182

                return err

×

183

×

184

        for _, gw := range natGws {

×

185

                if key == gw.Spec.Vpc {

×

186

                        c.updateVpcSubnetQueue.Add(gw.Name)

×

187

×

189

        return nil

×

NEW

197

func newVpcLoadBalancerConfig(name string, protocol v1.Protocol, template, sessionAffinity bool) *VpcLoadBalancerConfig {

×

NEW

198

        return &VpcLoadBalancerConfig{

×

NEW

199

                Name:            name,

×

NEW

200

                Protocol:        string(protocol),

×

NEW

201

                Template:        template,

×

NEW

202

                SessionAffinity: sessionAffinity,

×

NEW

203

×

NEW

204

×

NEW

221

func (v *VpcLoadBalancer) LBs() []*VpcLoadBalancerConfig {

×

NEW

222

        return []*VpcLoadBalancerConfig{

×

NEW

223

                v.TCPLoadBalancer,

×

NEW

224

                v.TCPSessLoadBalancer,

×

NEW

225

                v.UDPLoadBalancer,

×

NEW

226

                v.UDPSessLoadBalancer,

×

NEW

227

                v.SCTPLoadBalancer,

×

NEW

228

                v.SCTPSessLoadBalancer,

×

NEW

229

                v.LocalTCPLoadBalancer,

×

NEW

230

                v.LocalTCPSessLoadBalancer,

×

NEW

231

                v.LocalUDPLoadBalancer,

×

NEW

232

                v.LocalUDPSessLoadBalancer,

×

NEW

233

                v.LocalSCTPLoadBalancer,

×

NEW

234

                v.LocalSCTPSessLoadBalancer,

×

NEW

235

×

UNCOV

236

×

238

func (c *Controller) GenVpcLoadBalancer(vpcKey string) *VpcLoadBalancer {

×

239

        if vpcKey == c.config.ClusterRouter || vpcKey == "" {

×

240

                return &VpcLoadBalancer{

×

NEW

241

                        TCPLoadBalancer:           newVpcLoadBalancerConfig(c.config.ClusterTCPLoadBalancer, v1.ProtocolTCP, false, false),

×

NEW

242

                        TCPSessLoadBalancer:       newVpcLoadBalancerConfig(c.config.ClusterTCPSessionLoadBalancer, v1.ProtocolTCP, false, true),

×

NEW

243

                        UDPLoadBalancer:           newVpcLoadBalancerConfig(c.config.ClusterUDPLoadBalancer, v1.ProtocolUDP, false, false),

×

NEW

244

                        UDPSessLoadBalancer:       newVpcLoadBalancerConfig(c.config.ClusterUDPSessionLoadBalancer, v1.ProtocolUDP, false, true),

×

NEW

245

                        SCTPLoadBalancer:          newVpcLoadBalancerConfig(c.config.ClusterSCTPLoadBalancer, v1.ProtocolSCTP, false, false),

×

NEW

246

                        SCTPSessLoadBalancer:      newVpcLoadBalancerConfig(c.config.ClusterSCTPSessionLoadBalancer, v1.ProtocolSCTP, false, true),

×

NEW

247

                        LocalTCPLoadBalancer:      newVpcLoadBalancerConfig(c.config.LocalTCPLoadBalancer, v1.ProtocolTCP, true, false),

×

NEW

248

                        LocalTCPSessLoadBalancer:  newVpcLoadBalancerConfig(c.config.LocalTCPSessionLoadBalancer, v1.ProtocolTCP, true, true),

×

NEW

249

                        LocalUDPLoadBalancer:      newVpcLoadBalancerConfig(c.config.LocalUDPLoadBalancer, v1.ProtocolUDP, true, false),

×

NEW

250

                        LocalUDPSessLoadBalancer:  newVpcLoadBalancerConfig(c.config.LocalUDPSessionLoadBalancer, v1.ProtocolUDP, true, true),

×

NEW

251

                        LocalSCTPLoadBalancer:     newVpcLoadBalancerConfig(c.config.LocalSCTPLoadBalancer, v1.ProtocolSCTP, true, false),

×

NEW

252

                        LocalSCTPSessLoadBalancer: newVpcLoadBalancerConfig(c.config.LocalSCTPSessionLoadBalancer, v1.ProtocolSCTP, true, true),

×

253

×

254

×

255

        return &VpcLoadBalancer{

×

NEW

256

                TCPLoadBalancer:           newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-tcp-load", vpcKey), v1.ProtocolTCP, false, false),

×

NEW

257

                TCPSessLoadBalancer:       newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-tcp-sess-load", vpcKey), v1.ProtocolTCP, false, true),

×

NEW

258

                UDPLoadBalancer:           newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-udp-load", vpcKey), v1.ProtocolUDP, false, false),

×

NEW

259

                UDPSessLoadBalancer:       newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-udp-sess-load", vpcKey), v1.ProtocolUDP, false, true),

×

NEW

260

                SCTPLoadBalancer:          newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-sctp-load", vpcKey), v1.ProtocolSCTP, false, false),

×

NEW

261

                SCTPSessLoadBalancer:      newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-sctp-sess-load", vpcKey), v1.ProtocolSCTP, false, true),

×

NEW

262

                LocalTCPLoadBalancer:      newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-tcp-load", vpcKey), v1.ProtocolTCP, true, false),

×

NEW

263

                LocalTCPSessLoadBalancer:  newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-tcp-sess-load", vpcKey), v1.ProtocolTCP, true, true),

×

NEW

264

                LocalUDPLoadBalancer:      newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-udp-load", vpcKey), v1.ProtocolUDP, true, false),

×

NEW

265

                LocalUDPSessLoadBalancer:  newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-udp-sess-load", vpcKey), v1.ProtocolUDP, true, true),

×

NEW

266

                LocalSCTPLoadBalancer:     newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-sctp-load", vpcKey), v1.ProtocolSCTP, true, false),

×

NEW

267

                LocalSCTPSessLoadBalancer: newVpcLoadBalancerConfig(fmt.Sprintf("vpc-%s-local-sctp-sess-load", vpcKey), v1.ProtocolSCTP, true, true),

×

UNCOV

268

×

271

func (c *Controller) addLoadBalancer(vpc string) (*VpcLoadBalancer, error) {

×

272

        vpcLbConfig := c.GenVpcLoadBalancer(vpc)

×

NEW

273

        for _, lb := range vpcLbConfig.LBs() {

×

NEW

274

                if err := c.initLB(lb.Name, lb.Protocol, lb.Template, lb.SessionAffinity); err != nil {

×

NEW

275

                        klog.Error(err)

×

NEW

276

                        return nil, err

×

NEW

277

×

UNCOV

279

        return vpcLbConfig, nil

×

282

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

×

283

        c.vpcKeyMutex.LockKey(key)

×

284

        defer func() { _ = c.vpcKeyMutex.UnlockKey(key) }()

×

285

        klog.Infof("handle add/update vpc %s", key)

×

286

×

287

        cachedVpc, err := c.vpcsLister.Get(key)

×

288

        if err != nil {

×

289

                if k8serrors.IsNotFound(err) {

×

290

                        return nil

×

291

×

292

                klog.Error(err)

×

293

                return err

×

296

        vpc, err := c.formatVpc(cachedVpc.DeepCopy())

×

297

        if err != nil {

×

298

                klog.Errorf("failed to format vpc %s: %v", key, err)

×

299

                return err

×

300

×

301

        if err = c.createVpcRouter(key); err != nil {

×

302

                klog.Errorf("failed to create vpc router for vpc %s: %v", key, err)

×

303

                return err

×

304

×

306

        var newPeers []string

×

307

        for _, peering := range vpc.Spec.VpcPeerings {

×

308

                if err = util.CheckCidrs(peering.LocalConnectIP); err != nil {

×

309

                        klog.Errorf("invalid cidr %s", peering.LocalConnectIP)

×

310

                        return err

×

311

×

313

                newPeers = append(newPeers, peering.RemoteVpc)

×

314

                if err := c.OVNNbClient.CreatePeerRouterPort(vpc.Name, peering.RemoteVpc, peering.LocalConnectIP); err != nil {

×

315

                        klog.Errorf("create peer router port for vpc %s, %v", vpc.Name, err)

×

316

                        return err

×

317

×

319

        for _, oldPeer := range vpc.Status.VpcPeerings {

×

320

                if !slices.Contains(newPeers, oldPeer) {

×

321

                        if err = c.OVNNbClient.DeleteLogicalRouterPort(fmt.Sprintf("%s-%s", vpc.Name, oldPeer)); err != nil {

×

322

                                klog.Errorf("delete peer router port for vpc %s, %v", vpc.Name, err)

×

323

                                return err

×

324

×

329

        var (

×

330

                staticExistedRoutes []*ovnnb.LogicalRouterStaticRoute

×

331

                staticTargetRoutes  []*kubeovnv1.StaticRoute

×

332

                staticRouteMapping  map[string][]*kubeovnv1.StaticRoute

×

333

×

334

×

335

        staticExistedRoutes, err = c.OVNNbClient.ListLogicalRouterStaticRoutes(vpc.Name, nil, nil, "", nil)

×

336

        if err != nil {

×

337

                klog.Errorf("failed to get vpc %s static route list, %v", vpc.Name, err)

×

338

                return err

×

339

×

341

        var externalSubnet *kubeovnv1.Subnet

×

342

        externalSubnetExist := false

×

343

        if c.config.EnableEipSnat {

×

344

                externalSubnet, err = c.subnetsLister.Get(c.config.ExternalGatewaySwitch)

×

345

                if err != nil {

×

346

                        klog.Warningf("enable-eip-snat need external subnet %s to be exist: %v", c.config.ExternalGatewaySwitch, err)

×

347

                } else {

×

348

                        externalSubnetExist = true

×

349

×

352

        staticRouteMapping = c.getRouteTablesByVpc(vpc)

×

353

        staticTargetRoutes = vpc.Spec.StaticRoutes

×

354

        if vpc.Name == c.config.ClusterRouter {

×

355

                if _, ok := staticRouteMapping[util.MainRouteTable]; !ok {

×

356

                        staticRouteMapping[util.MainRouteTable] = nil

×

357

×

359

                joinSubnet, err := c.subnetsLister.Get(c.config.NodeSwitch)

×

360

                if err != nil {

×

361

                        if !k8serrors.IsNotFound(err) {

×

362

                                klog.Errorf("failed to get node switch subnet %s: %v", c.config.NodeSwitch, err)

×

363

                                return err

×

364

×

365

                        c.addOrUpdateVpcQueue.Add(vpc.Name)

×

366

                        return nil

×

368

                gatewayV4, gatewayV6 := util.SplitStringIP(joinSubnet.Spec.Gateway)

×

369

                if gatewayV4 != "" {

×

370

                        for table := range staticRouteMapping {

×

371

                                staticTargetRoutes = append(

×

372

                                        staticTargetRoutes,

×

373

                                        &kubeovnv1.StaticRoute{

×

374

                                                Policy:     kubeovnv1.PolicyDst,

×

375

                                                CIDR:       "0.0.0.0/0",

×

376

                                                NextHopIP:  gatewayV4,

×

377

                                                RouteTable: table,

×

378

},

×

379

×

380

×

382

                if gatewayV6 != "" {

×

383

                        for table := range staticRouteMapping {

×

384

                                staticTargetRoutes = append(

×

385

                                        staticTargetRoutes,

×

386

                                        &kubeovnv1.StaticRoute{

×

387

                                                Policy:     kubeovnv1.PolicyDst,

×

388

                                                CIDR:       "::/0",

×

389

                                                NextHopIP:  gatewayV6,

×

390

                                                RouteTable: table,

×

391

},

×

392

×

393

×

395

                if c.config.EnableEipSnat {

×

396

                        cm, err := c.configMapsLister.ConfigMaps(c.config.ExternalGatewayConfigNS).Get(util.ExternalGatewayConfig)

×

397

                        if err == nil {

×

398

                                nextHop := cm.Data["external-gw-addr"]

×

399

                                if nextHop == "" {

×

400

                                        if !externalSubnetExist {

×

401

                                                err = fmt.Errorf("failed to get external subnet %s", c.config.ExternalGatewaySwitch)

×

402

                                                klog.Error(err)

×

403

                                                return err

×

404

×

405

                                        nextHop = externalSubnet.Spec.Gateway

×

406

                                        if nextHop == "" {

×

407

                                                klog.Errorf("no available gateway address")

×

408

                                                return errors.New("no available gateway address")

×

409

×

411

                                if strings.Contains(nextHop, "/") {

×

412

                                        nextHop = strings.Split(nextHop, "/")[0]

×

413

×

415

                                lr, err := c.OVNNbClient.GetLogicalRouter(vpc.Name, false)

×

416

                                if err != nil {

×

417

                                        klog.Errorf("failed to get logical router %s: %v", vpc.Name, err)

×

418

                                        return err

×

419

×

421

                                for _, nat := range lr.Nat {

×

422

                                        info, err := c.OVNNbClient.GetNATByUUID(nat)

×

423

                                        if err != nil {

×

424

                                                klog.Errorf("failed to get nat ip info for vpc %s, %v", vpc.Name, err)

×

425

                                                return err

×

426

×

427

                                        if info.LogicalIP != "" {

×

428

                                                for table := range staticRouteMapping {

×

429

                                                        staticTargetRoutes = append(

×

430

                                                                staticTargetRoutes,

×

431

                                                                &kubeovnv1.StaticRoute{

×

432

                                                                        Policy:     kubeovnv1.PolicySrc,

×

433

                                                                        CIDR:       info.LogicalIP,

×

434

                                                                        NextHopIP:  nextHop,

×

435

                                                                        RouteTable: table,

×

436

},

×

437

×

438

×

445

        routeNeedDel, routeNeedAdd, err := diffStaticRoute(staticExistedRoutes, staticTargetRoutes)

×

446

        if err != nil {

×

447

                klog.Errorf("failed to diff vpc %s static route, %v", vpc.Name, err)

×

448

                return err

×

449

×

451

        for _, item := range routeNeedDel {

×

452

                klog.Infof("vpc %s del static route: %+v", vpc.Name, item)

×

453

                policy := convertPolicy(item.Policy)

×

454

                if err = c.OVNNbClient.DeleteLogicalRouterStaticRoute(vpc.Name, &item.RouteTable, &policy, item.CIDR, item.NextHopIP); err != nil {

×

455

                        klog.Errorf("del vpc %s static route failed, %v", vpc.Name, err)

×

456

                        return err

×

457

×

460

        for _, item := range routeNeedAdd {

×

461

                if item.BfdID != "" {

×

462

                        klog.Infof("vpc %s add static ecmp route: %+v", vpc.Name, item)

×

463

                        if err = c.OVNNbClient.AddLogicalRouterStaticRoute(

×

464

                                vpc.Name, item.RouteTable, convertPolicy(item.Policy), item.CIDR, &item.BfdID, nil, item.NextHopIP,

×

465

                        ); err != nil {

×

466

                                klog.Errorf("failed to add bfd static route to vpc %s , %v", vpc.Name, err)

×

467

                                return err

×

468

×

469

                } else {

×

470

                        klog.Infof("vpc %s add static route: %+v", vpc.Name, item)

×

471

                        if err = c.OVNNbClient.AddLogicalRouterStaticRoute(

×

472

                                vpc.Name, item.RouteTable, convertPolicy(item.Policy), item.CIDR, nil, nil, item.NextHopIP,

×

473

                        ); err != nil {

×

474

                                klog.Errorf("failed to add normal static route to vpc %s , %v", vpc.Name, err)

×

475

                                return err

×

476

×

481

        var (

×

482

                policyRouteExisted, policyRouteNeedDel, policyRouteNeedAdd []*kubeovnv1.PolicyRoute

×

483

                policyRouteLogical                                         []*ovnnb.LogicalRouterPolicy

×

484

                externalIDs                                                = map[string]string{"vendor": util.CniTypeName}

×

485

×

486

×

487

        if vpc.Name == c.config.ClusterRouter {

×

488

                policyRouteExisted = reversePolicies(vpc.Annotations[util.VpcLastPolicies])

×

489

                // diff list

×

490

                policyRouteNeedDel, policyRouteNeedAdd = diffPolicyRouteWithExisted(policyRouteExisted, vpc.Spec.PolicyRoutes)

×

491

        } else {

×

492

                if vpc.Spec.PolicyRoutes == nil {

×

493

                        // do not clean default vpc policy routes

×

494

                        if err = c.OVNNbClient.ClearLogicalRouterPolicy(vpc.Name); err != nil {

×

495

                                klog.Errorf("clean all vpc %s policy route failed, %v", vpc.Name, err)

×

496

                                return err

×

497

×

498

                } else {

×

499

                        policyRouteLogical, err = c.OVNNbClient.ListLogicalRouterPolicies(vpc.Name, -1, nil, true)

×

500

                        if err != nil {

×

501

                                klog.Errorf("failed to get vpc %s policy route list, %v", vpc.Name, err)

×

502

                                return err

×

503

×

505

                        policyRouteNeedDel, policyRouteNeedAdd = diffPolicyRouteWithLogical(policyRouteLogical, vpc.Spec.PolicyRoutes)

×

509

        for _, item := range policyRouteNeedDel {

×

510

                klog.Infof("delete policy route for router: %s, priority: %d, match %s", vpc.Name, item.Priority, item.Match)

×

511

                if err = c.OVNNbClient.DeleteLogicalRouterPolicy(vpc.Name, item.Priority, item.Match); err != nil {

×

512

                        klog.Errorf("del vpc %s policy route failed, %v", vpc.Name, err)

×

513

                        return err

×

514

×

517

        for _, item := range policyRouteNeedAdd {

×

518

                klog.Infof("add policy route for router: %s, match %s, action %s, nexthop %s, externalID %v", c.config.ClusterRouter, item.Match, string(item.Action), item.NextHopIP, externalIDs)

×

519

                if err = c.OVNNbClient.AddLogicalRouterPolicy(vpc.Name, item.Priority, item.Match, string(item.Action), []string{item.NextHopIP}, nil, externalIDs); err != nil {

×

520

                        klog.Errorf("add policy route to vpc %s failed, %v", vpc.Name, err)

×

521

                        return err

×

522

×

525

        vpc.Status.Router = key

×

526

        vpc.Status.Standby = true

×

527

        vpc.Status.VpcPeerings = newPeers

×

528

        if c.config.EnableLb {

×

529

                vpcLb, err := c.addLoadBalancer(key)

×

530

                if err != nil {

×

531

                        klog.Error(err)

×

532

                        return err

×

533

×

NEW

534

                vpc.Status.TCPLoadBalancer = vpcLb.TCPLoadBalancer.Name

×

NEW

535

                vpc.Status.TCPSessionLoadBalancer = vpcLb.TCPSessLoadBalancer.Name

×

NEW

536

                vpc.Status.UDPLoadBalancer = vpcLb.UDPLoadBalancer.Name

×

NEW

537

                vpc.Status.UDPSessionLoadBalancer = vpcLb.UDPSessLoadBalancer.Name

×

NEW

538

                vpc.Status.SCTPLoadBalancer = vpcLb.SCTPLoadBalancer.Name

×

NEW

539

                vpc.Status.SCTPSessionLoadBalancer = vpcLb.SCTPSessLoadBalancer.Name

×

NEW

540

                vpc.Status.LocalTCPLoadBalancer = vpcLb.LocalTCPLoadBalancer.Name

×

NEW

541

                vpc.Status.LocalTCPSessionLoadBalancer = vpcLb.LocalTCPSessLoadBalancer.Name

×

NEW

542

                vpc.Status.LocalUDPLoadBalancer = vpcLb.LocalUDPLoadBalancer.Name

×

NEW

543

                vpc.Status.LocalUDPSessionLoadBalancer = vpcLb.LocalUDPSessLoadBalancer.Name

×

NEW

544

                vpc.Status.LocalSCTPLoadBalancer = vpcLb.LocalSCTPLoadBalancer.Name

×

NEW

545

                vpc.Status.LocalSCTPSessionLoadBalancer = vpcLb.LocalSCTPSessLoadBalancer.Name

×

547

        bytes, err := vpc.Status.Bytes()

×

548

        if err != nil {

×

549

                klog.Error(err)

×

550

                return err

×

551

×

552

        vpc, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(), vpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status")

×

553

        if err != nil {

×

554

                klog.Error(err)

×

555

                return err

×

556

×

558

        if len(vpc.Annotations) != 0 && strings.ToLower(vpc.Annotations[util.VpcLbAnnotation]) == "on" {

×

559

                if err = c.createVpcLb(vpc); err != nil {

×

560

                        klog.Error(err)

×

561

                        return err

×

562

×

563

        } else if err = c.deleteVpcLb(vpc); err != nil {

×

564

                klog.Error(err)

×

565

                return err

×

566

×

568

        subnets, err := c.subnetsLister.List(labels.Everything())

×

569

        if err != nil {

×

570

                klog.Error(err)

×

571

                return err

×

572

×

573

        custVpcEnableExternalMultiHopEcmp := false

×

574

        for _, subnet := range subnets {

×

575

                if subnet.Spec.Vpc == key {

×

576

                        c.addOrUpdateSubnetQueue.Add(subnet.Name)

×

577

                        if vpc.Name != util.DefaultVpc && vpc.Spec.EnableBfd && subnet.Spec.EnableEcmp {

×

578

                                custVpcEnableExternalMultiHopEcmp = true

×

579

×

583

        if vpc.Name != util.DefaultVpc {

×

584

                if cachedVpc.Spec.EnableExternal {

×

585

                        if !externalSubnetExist {

×

586

                                err = fmt.Errorf("failed to get external subnet %s", c.config.ExternalGatewaySwitch)

×

587

                                klog.Error(err)

×

588

                                return err

×

589

×

590

                        if externalSubnet.Spec.LogicalGateway {

×

591

                                klog.Infof("no need to handle external connection for logical gw external subnet %s", c.config.ExternalGatewaySwitch)

×

592

                                return nil

×

593

×

594

                        if !cachedVpc.Status.EnableExternal {

×

595

                                // connect vpc to default external

×

596

                                klog.Infof("connect external network with vpc %s", vpc.Name)

×

597

                                if err := c.handleAddVpcExternalSubnet(key, c.config.ExternalGatewaySwitch); err != nil {

×

598

                                        klog.Errorf("failed to add default external connection for vpc %s, error %v", key, err)

×

599

                                        return err

×

600

×

602

                        if vpc.Spec.EnableBfd {

×

603

                                // create bfd between lrp and physical switch gw

×

604

                                // bfd status down means current lrp binding chassis node external nic lost external network connectivity

×

605

                                // should switch lrp to another node

×

606

                                lrpEipName := fmt.Sprintf("%s-%s", key, c.config.ExternalGatewaySwitch)

×

607

                                v4ExtGw, _ := util.SplitStringIP(externalSubnet.Spec.Gateway)

×

608

                                // TODO: dualstack

×

609

                                if _, err := c.OVNNbClient.CreateBFD(lrpEipName, v4ExtGw, c.config.BfdMinRx, c.config.BfdMinTx, c.config.BfdDetectMult, nil); err != nil {

×

610

                                        klog.Error(err)

×

611

                                        return err

×

612

×

614

                                if custVpcEnableExternalMultiHopEcmp {

×

615

                                        klog.Infof("remove normal static ecmp route for vpc %s", vpc.Name)

×

616

                                        // auto remove normal type static route, if using ecmp based bfd

×

617

                                        if err := c.reconcileCustomVpcDelNormalStaticRoute(vpc.Name); err != nil {

×

618

                                                klog.Errorf("failed to reconcile del vpc %q normal static route", vpc.Name)

×

619

                                                return err

×

620

×

623

                        if cachedVpc.Spec.ExtraExternalSubnets != nil {

×

624

                                sort.Strings(vpc.Spec.ExtraExternalSubnets)

×

625

×

627

                        if !slices.Equal(vpc.Spec.ExtraExternalSubnets, vpc.Status.ExtraExternalSubnets) {

×

628

                                for _, subnetStatus := range cachedVpc.Status.ExtraExternalSubnets {

×

629

                                        if !slices.Contains(cachedVpc.Spec.ExtraExternalSubnets, subnetStatus) {

×

630

                                                klog.Infof("delete external subnet %s connection for vpc %s", subnetStatus, vpc.Name)

×

631

                                                if err := c.handleDelVpcExternalSubnet(vpc.Name, subnetStatus); err != nil {

×

632

                                                        klog.Errorf("failed to delete external subnet %s connection for vpc %s, error %v", subnetStatus, vpc.Name, err)

×

633

                                                        return err

×

634

×

637

                                for _, subnetSpec := range cachedVpc.Spec.ExtraExternalSubnets {

×

638

                                        if !slices.Contains(cachedVpc.Status.ExtraExternalSubnets, subnetSpec) {

×

639

                                                klog.Infof("connect external subnet %s with vpc %s", subnetSpec, vpc.Name)

×

640

                                                if err := c.handleAddVpcExternalSubnet(key, subnetSpec); err != nil {

×

641

                                                        klog.Errorf("failed to add external subnet %s connection for vpc %s, error %v", subnetSpec, key, err)

×

642

                                                        return err

×

643

×

646

                                if err := c.updateVpcAddExternalStatus(key, true); err != nil {

×

647

                                        klog.Errorf("failed to update additional external subnets status, %v", err)

×

648

                                        return err

×

649

×

653

                if !cachedVpc.Spec.EnableBfd && cachedVpc.Status.EnableBfd {

×

654

                        lrpEipName := fmt.Sprintf("%s-%s", key, c.config.ExternalGatewaySwitch)

×

655

                        if err := c.OVNNbClient.DeleteBFDByDstIP(lrpEipName, ""); err != nil {

×

656

                                klog.Error(err)

×

657

                                return err

×

658

×

659

                        if err := c.handleDeleteVpcStaticRoute(key); err != nil {

×

660

                                klog.Errorf("failed to delete bfd route for vpc %s, error %v", key, err)

×

661

                                return err

×

662

×

665

                if !cachedVpc.Spec.EnableExternal && cachedVpc.Status.EnableExternal {

×

666

                        // disconnect vpc to default external

×

667

                        if err := c.handleDelVpcExternalSubnet(key, c.config.ExternalGatewaySwitch); err != nil {

×

668

                                klog.Errorf("failed to delete external connection for vpc %s, error %v", key, err)

×

669

                                return err

×

670

×

673

                if cachedVpc.Status.ExtraExternalSubnets != nil && !cachedVpc.Spec.EnableExternal {

×

674

                        // disconnect vpc to extra external subnets

×

675

                        for _, subnet := range cachedVpc.Status.ExtraExternalSubnets {

×

676

                                klog.Infof("disconnect external network %s to vpc %s", subnet, vpc.Name)

×

677

                                if err := c.handleDelVpcExternalSubnet(key, subnet); err != nil {

×

678

                                        klog.Error(err)

×

679

                                        return err

×

680

×

682

                        if err := c.updateVpcAddExternalStatus(key, false); err != nil {

×

683

                                klog.Errorf("failed to update additional external subnets status, %v", err)

×

684

                                return err

×

685

×

689

        bfdPortName, bfdPortNodes, err := c.reconcileVpcBfdLRP(vpc)

×

690

        if err != nil {

×

691

                klog.Error(err)

×

692

                return err

×

693

×

694

        if vpc.Spec.BFDPort == nil || !vpc.Spec.BFDPort.Enabled {

×

695

                vpc.Status.BFDPort = kubeovnv1.BFDPortStatus{}

×

696

        } else {

×

697

                vpc.Status.BFDPort = kubeovnv1.BFDPortStatus{

×

698

                        Name:  bfdPortName,

×

699

                        IP:    vpc.Spec.BFDPort.IP,

×

700

                        Nodes: bfdPortNodes,

×

701

×

702

×

703

        if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().

×

704

                UpdateStatus(context.Background(), vpc, metav1.UpdateOptions{}); err != nil {

×

705

                klog.Error(err)

×

706

                return err

×

707

×

709

        return nil

×

712

func (c *Controller) reconcileVpcBfdLRP(vpc *kubeovnv1.Vpc) (string, []string, error) {

×

713

        portName := "bfd@" + vpc.Name

×

714

        if vpc.Spec.BFDPort == nil || !vpc.Spec.BFDPort.Enabled {

×

715

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

×

716

                        err = fmt.Errorf("failed to delete BFD LRP %s: %w", portName, err)

×

717

                        klog.Error(err)

×

718

                        return portName, nil, err

×

719

×

720

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

×

721

                        err = fmt.Errorf("failed to delete HA chassis group %s: %w", portName, err)

×

722

                        klog.Error(err)

×

723

                        return portName, nil, err

×

724

×

725

                return portName, nil, nil

×

728

        var err error

×

729

        chassisCount := 3

×

730

        selector := labels.Everything()

×

731

        if vpc.Spec.BFDPort.NodeSelector != nil {

×

732

                chassisCount = math.MaxInt

×

733

                if selector, err = metav1.LabelSelectorAsSelector(vpc.Spec.BFDPort.NodeSelector); err != nil {

×

734

                        err = fmt.Errorf("failed to parse node selector %q: %w", vpc.Spec.BFDPort.NodeSelector.String(), err)

×

735

                        klog.Error(err)

×

736

                        return portName, nil, err

×

737

×

740

        nodes, err := c.nodesLister.List(selector)

×

741

        if err != nil {

×

742

                err = fmt.Errorf("failed to list nodes with selector %q: %w", vpc.Spec.BFDPort.NodeSelector, err)

×

743

                klog.Error(err)

×

744

                return portName, nil, err

×

745

×

746

        if len(nodes) == 0 {

×

747

                err = fmt.Errorf("no nodes found by selector %q", selector.String())

×

748

                klog.Error(err)

×

749

                return portName, nil, err

×

750

×

752

        nodeNames := make([]string, 0, len(nodes))

×

753

        chassisCount = min(chassisCount, len(nodes))

×

754

        chassisNames := make([]string, 0, chassisCount)

×

755

        for _, nodes := range nodes[:chassisCount] {

×

756

                chassis, err := c.OVNSbClient.GetChassisByHost(nodes.Name)

×

757

                if err != nil {

×

758

                        err = fmt.Errorf("failed to get chassis of node %s: %w", nodes.Name, err)

×

759

                        klog.Error(err)

×

760

                        return portName, nil, err

×

761

×

762

                chassisNames = append(chassisNames, chassis.Name)

×

763

                nodeNames = append(nodeNames, nodes.Name)

×

766

        networks := strings.Split(vpc.Spec.BFDPort.IP, ",")

×

767

        if err = c.OVNNbClient.CreateLogicalRouterPort(vpc.Name, portName, "", networks); err != nil {

×

768

                klog.Error(err)

×

769

                return portName, nil, err

×

770

×

771

        if err = c.OVNNbClient.UpdateLogicalRouterPortNetworks(portName, networks); err != nil {

×

772

                klog.Error(err)

×

773

                return portName, nil, err

×

774

×

775

        if err = c.OVNNbClient.UpdateLogicalRouterPortOptions(portName, map[string]string{"bfd-only": "true"}); err != nil {

×

776

                klog.Error(err)

×

777

                return portName, nil, err

×

778

×

779

        if err = c.OVNNbClient.CreateHAChassisGroup(portName, chassisNames, map[string]string{"lrp": portName}); err != nil {

×

780

                klog.Error(err)

×

781

                return portName, nil, err

×

782

×

783

        if err = c.OVNNbClient.SetLogicalRouterPortHAChassisGroup(portName, portName); err != nil {

×

784

                klog.Error(err)

×

785

                return portName, nil, err

×

786

×

788

        return portName, nodeNames, nil

×

791

func (c *Controller) addPolicyRouteToVpc(vpcName string, policy *kubeovnv1.PolicyRoute, externalIDs map[string]string) error {

×

792

        var (

×

793

                nextHops []string

×

794

                err      error

×

795

×

796

×

797

        if policy.NextHopIP != "" {

×

798

                nextHops = strings.Split(policy.NextHopIP, ",")

×

799

×

801

        if err = c.OVNNbClient.AddLogicalRouterPolicy(vpcName, policy.Priority, policy.Match, string(policy.Action), nextHops, nil, externalIDs); err != nil {

×

802

                klog.Errorf("add policy route to vpc %s failed, %v", vpcName, err)

×

803

                return err

×

804

×

805

        return nil

×

808

func buildExternalIDsMapKey(match, action string, priority int) string {

×

809

        return fmt.Sprintf("%s-%s-%d", match, action, priority)

×

810

×

812

func (c *Controller) batchAddPolicyRouteToVpc(name string, policies []*kubeovnv1.PolicyRoute, externalIDs map[string]map[string]string) error {

×

813

        if len(policies) == 0 {

×

814

                return nil

×

815

×

816

        start := time.Now()

×

817

        routerPolicies := make([]*ovnnb.LogicalRouterPolicy, 0, len(policies))

×

818

        for _, policy := range policies {

×

819

                var nextHops []string

×

820

                if policy.NextHopIP != "" {

×

821

                        nextHops = strings.Split(policy.NextHopIP, ",")

×

822

×

823

                routerPolicies = append(routerPolicies, &ovnnb.LogicalRouterPolicy{

×

824

                        Priority:    policy.Priority,

×

825

                        Nexthops:    nextHops,

×

826

                        Action:      string(policy.Action),

×

827

                        Match:       policy.Match,

×

828

                        ExternalIDs: externalIDs[buildExternalIDsMapKey(policy.Match, string(policy.Action), policy.Priority)],

×

829

})

×

832

        if err := c.OVNNbClient.BatchAddLogicalRouterPolicy(name, routerPolicies...); err != nil {

×

833

                klog.Errorf("batch add policy route to vpc %s failed, %v", name, err)

×

834

                return err

×

835

×

836

        klog.Infof("take to %v batch add policy route to vpc %s policies %d", time.Since(start), name, len(policies))

×

837

        return nil

×

840

func (c *Controller) deletePolicyRouteFromVpc(vpcName string, priority int, match string) error {

×

841

        var (

×

842

                vpc, cachedVpc *kubeovnv1.Vpc

×

843

                err            error

×

844

×

845

×

846

        if err = c.OVNNbClient.DeleteLogicalRouterPolicy(vpcName, priority, match); err != nil {

×

847

                klog.Error(err)

×

848

                return err

×

849

×

851

        cachedVpc, err = c.vpcsLister.Get(vpcName)

×

852

        if err != nil {

×

853

                if k8serrors.IsNotFound(err) {

×

854

                        return nil

×

855

×

856

                klog.Error(err)

×

857

                return err

×

859

        vpc = cachedVpc.DeepCopy()

×

860

        // make sure custom policies not be deleted

×

861

        _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Update(context.Background(), vpc, metav1.UpdateOptions{})

×

862

        if err != nil {

×

863

                klog.Error(err)

×

864

                return err

×

865

×

866

        return nil

×

869

func (c *Controller) batchDeletePolicyRouteFromVpc(name string, policies []*kubeovnv1.PolicyRoute) error {

×

870

        var (

×

871

                vpc, cachedVpc *kubeovnv1.Vpc

×

872

                err            error

×

873

×

874

×

875

        start := time.Now()

×

876

        routerPolicies := make([]*ovnnb.LogicalRouterPolicy, 0, len(policies))

×

877

        for _, policy := range policies {

×

878

                routerPolicies = append(routerPolicies, &ovnnb.LogicalRouterPolicy{

×

879

                        Priority: policy.Priority,

×

880

                        Match:    policy.Match,

×

881

})

×

882

×

884

        if err = c.OVNNbClient.BatchDeleteLogicalRouterPolicy(name, routerPolicies); err != nil {

×

885

                return err

×

886

×

887

        klog.V(3).Infof("take to %v batch delete policy route from vpc %s policies %d", time.Since(start), name, len(policies))

×

888

×

889

        cachedVpc, err = c.vpcsLister.Get(name)

×

890

        if err != nil {

×

891

                if k8serrors.IsNotFound(err) {

×

892

                        return nil

×

893

×

894

                klog.Error(err)

×

895

                return err

×

897

        vpc = cachedVpc.DeepCopy()

×

898

        // make sure custom policies not be deleted

×

899

        _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Update(context.Background(), vpc, metav1.UpdateOptions{})

×

900

        if err != nil {

×

901

                klog.Error(err)

×

902

                return err

×

903

×

904

        return nil

×

907

func (c *Controller) addStaticRouteToVpc(name string, route *kubeovnv1.StaticRoute) error {

×

908

        if route.BfdID != "" {

×

909

                klog.Infof("vpc %s add static ecmp route: %+v", name, route)

×

910

                if err := c.OVNNbClient.AddLogicalRouterStaticRoute(

×

911

                        name, route.RouteTable, convertPolicy(route.Policy), route.CIDR, &route.BfdID, nil, route.NextHopIP,

×

912

                ); err != nil {

×

913

                        klog.Errorf("failed to add bfd static route to vpc %s , %v", name, err)

×

914

                        return err

×

915

×

916

        } else {

×

917

                klog.Infof("vpc %s add static route: %+v", name, route)

×

918

                if err := c.OVNNbClient.AddLogicalRouterStaticRoute(

×

919

                        name, route.RouteTable, convertPolicy(route.Policy), route.CIDR, nil, nil, route.NextHopIP,

×

920

                ); err != nil {

×

921

                        klog.Errorf("failed to add normal static route to vpc %s , %v", name, err)

×

922

                        return err

×

923

×

925

        return nil

×

928

func (c *Controller) deleteStaticRouteFromVpc(name, table, cidr, nextHop string, policy kubeovnv1.RoutePolicy) error {

×

929

        var (

×

930

                policyStr string

×

931

                err       error

×

932

×

933

×

934

        policyStr = convertPolicy(policy)

×

935

        if err = c.OVNNbClient.DeleteLogicalRouterStaticRoute(name, &table, &policyStr, cidr, nextHop); err != nil {

×

936

                klog.Errorf("del vpc %s static route failed, %v", name, err)

×

937

                return err

×

938

×

940

        return nil

×

943

func (c *Controller) batchDeleteStaticRouteFromVpc(name string, staticRoutes []*kubeovnv1.StaticRoute) error {

×

944

        var (

×

945

                vpc, cachedVpc *kubeovnv1.Vpc

×

946

                err            error

×

947

×

948

        start := time.Now()

×

949

        routeCount := len(staticRoutes)

×

950

        delRoutes := make([]*ovnnb.LogicalRouterStaticRoute, 0, routeCount)

×

951

        for _, sr := range staticRoutes {

×

952

                policyStr := convertPolicy(sr.Policy)

×

953

                newRoute := &ovnnb.LogicalRouterStaticRoute{

×

954

                        RouteTable: sr.RouteTable,

×

955

                        Nexthop:    sr.NextHopIP,

×

956

                        Policy:     &policyStr,

×

957

                        IPPrefix:   sr.CIDR,

×

958

×

959

                delRoutes = append(delRoutes, newRoute)

×

960

×

961

        if err = c.OVNNbClient.BatchDeleteLogicalRouterStaticRoute(name, delRoutes); err != nil {

×

962

                klog.Errorf("batch del vpc %s static route %d failed, %v", name, routeCount, err)

×

963

                return err

×

964

×

965

        klog.V(3).Infof("take to %v batch delete static route from vpc %s static routes %d", time.Since(start), name, len(delRoutes))

×

966

×

967

        cachedVpc, err = c.vpcsLister.Get(name)

×

968

        if err != nil {

×

969

                if k8serrors.IsNotFound(err) {

×

970

                        return nil

×

971

×

972

                klog.Error(err)

×

973

                return err

×

975

        vpc = cachedVpc.DeepCopy()

×

976

        // make sure custom policies not be deleted

×

977

        _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Update(context.Background(), vpc, metav1.UpdateOptions{})

×

978

        if err != nil {

×

979

                klog.Error(err)

×

980

                return err

×

981

×

982

        return nil

×

985

func diffPolicyRouteWithExisted(exists, target []*kubeovnv1.PolicyRoute) ([]*kubeovnv1.PolicyRoute, []*kubeovnv1.PolicyRoute) {

×

986

        var (

×

987

                dels, adds []*kubeovnv1.PolicyRoute

×

988

                existsMap  map[string]*kubeovnv1.PolicyRoute

×

989

                key        string

×

990

                ok         bool

×

991

×

992

×

993

        existsMap = make(map[string]*kubeovnv1.PolicyRoute, len(exists))

×

994

        for _, item := range exists {

×

995

                existsMap[getPolicyRouteItemKey(item)] = item

×

996

×

998

        for _, item := range target {

×

999

                key = getPolicyRouteItemKey(item)

×

1000

×

1001

                if _, ok = existsMap[key]; ok {

×

1002

                        delete(existsMap, key)

×

1003

                } else {

×

1004

                        adds = append(adds, item)

×

1005

×

1008

        for _, item := range existsMap {

×

1009

                dels = append(dels, item)

×

1010

×

1011

        return dels, adds

×

1014

func diffPolicyRouteWithLogical(exists []*ovnnb.LogicalRouterPolicy, target []*kubeovnv1.PolicyRoute) ([]*kubeovnv1.PolicyRoute, []*kubeovnv1.PolicyRoute) {

×

1015

        var (

×

1016

                dels, adds []*kubeovnv1.PolicyRoute

×

1017

                existsMap  map[string]*kubeovnv1.PolicyRoute

×

1018

                key        string

×

1019

                ok         bool

×

1020

×

1021

        existsMap = make(map[string]*kubeovnv1.PolicyRoute, len(exists))

×

1022

×

1023

        for _, item := range exists {

×

1024

                policy := &kubeovnv1.PolicyRoute{

×

1025

                        Priority: item.Priority,

×

1026

                        Match:    item.Match,

×

1027

                        Action:   kubeovnv1.PolicyRouteAction(item.Action),

×

1028

×

1029

                existsMap[getPolicyRouteItemKey(policy)] = policy

×

1030

×

1032

        for _, item := range target {

×

1033

                key = getPolicyRouteItemKey(item)

×

1034

×

1035

                if _, ok = existsMap[key]; ok {

×

1036

                        delete(existsMap, key)

×

1037

                } else {

×

1038

                        adds = append(adds, item)

×

1039

×

1042

        for _, item := range existsMap {

×

1043

                dels = append(dels, item)

×

1044

×

1045

        return dels, adds

×

1048

func getPolicyRouteItemKey(item *kubeovnv1.PolicyRoute) (key string) {

×

1049

        return fmt.Sprintf("%d:%s:%s:%s", item.Priority, item.Match, item.Action, item.NextHopIP)

×

1050

×

1052

func diffStaticRoute(exist []*ovnnb.LogicalRouterStaticRoute, target []*kubeovnv1.StaticRoute) (routeNeedDel, routeNeedAdd []*kubeovnv1.StaticRoute, err error) {

×

1053

        existRouteMap := make(map[string]*kubeovnv1.StaticRoute, len(exist))

×

1054

        for _, item := range exist {

×

1055

                policy := kubeovnv1.PolicyDst

×

1056

                if item.Policy != nil && *item.Policy == ovnnb.LogicalRouterStaticRoutePolicySrcIP {

×

1057

                        policy = kubeovnv1.PolicySrc

×

1058

×

1059

                route := &kubeovnv1.StaticRoute{

×

1060

                        Policy:     policy,

×

1061

                        CIDR:       item.IPPrefix,

×

1062

                        NextHopIP:  item.Nexthop,

×

1063

                        RouteTable: item.RouteTable,

×

1064

                        ECMPMode:   util.StaticRouteBfdEcmp,

×

1065

×

1066

                if item.BFD != nil {

×

1067

                        route.BfdID = *item.BFD

×

1068

×

1069

                existRouteMap[getStaticRouteItemKey(route)] = route

×

1072

        for _, item := range target {

×

1073

                key := getStaticRouteItemKey(item)

×

1074

                if _, ok := existRouteMap[key]; ok {

×

1075

                        delete(existRouteMap, key)

×

1076

                } else {

×

1077

                        routeNeedAdd = append(routeNeedAdd, item)

×

1078

×

1080

        for _, item := range existRouteMap {

×

1081

                routeNeedDel = append(routeNeedDel, item)

×

1082

×

1083

        return

×

1086

func getStaticRouteItemKey(item *kubeovnv1.StaticRoute) string {

×

1087

        var key string

×

1088

        if item.Policy == kubeovnv1.PolicyDst {

×

1089

                key = fmt.Sprintf("%s:dst:%s=>%s", item.RouteTable, item.CIDR, item.NextHopIP)

×

1090

        } else {

×

1091

                key = fmt.Sprintf("%s:src:%s=>%s", item.RouteTable, item.CIDR, item.NextHopIP)

×

1092

×

1093

        return key

×

1096

func (c *Controller) formatVpc(vpc *kubeovnv1.Vpc) (*kubeovnv1.Vpc, error) {

×

1097

        var changed bool

×

1098

        for _, item := range vpc.Spec.StaticRoutes {

×

1099

                // check policy

×

1100

                if item.Policy == "" {

×

1101

                        item.Policy = kubeovnv1.PolicyDst

×

1102

                        changed = true

×

1103

×

1104

                if item.Policy != kubeovnv1.PolicyDst && item.Policy != kubeovnv1.PolicySrc {

×

1105

                        return nil, fmt.Errorf("unknown policy type: %q", item.Policy)

×

1106

×

1108

                if strings.Contains(item.CIDR, "/") {

×

1109

                        if _, _, err := net.ParseCIDR(item.CIDR); err != nil {

×

1110

                                return nil, fmt.Errorf("invalid cidr %q: %w", item.CIDR, err)

×

1111

×

1112

                } else if ip := net.ParseIP(item.CIDR); ip == nil {

×

1113

                        return nil, fmt.Errorf("invalid ip %q", item.CIDR)

×

1114

×

1116

                if ip := net.ParseIP(item.NextHopIP); ip == nil {

×

1117

                        return nil, fmt.Errorf("invalid next hop ip %q", item.NextHopIP)

×

1118

×

1121

        for _, route := range vpc.Spec.PolicyRoutes {

×

1122

                if route.Action != kubeovnv1.PolicyRouteActionReroute {

×

1123

                        if route.NextHopIP != "" {

×

1124

                                route.NextHopIP = ""

×

1125

                                changed = true

×

1126

×

1127

                } else {

×

1128

                        // ecmp policy route may reroute to multiple next hop ips

×

1129

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

×

1130

                                if ip := net.ParseIP(ipStr); ip == nil {

×

1131

                                        err := fmt.Errorf("invalid next hop ips: %s", route.NextHopIP)

×

1132

                                        klog.Error(err)

×

1133

                                        return nil, err

×

1134

×

1139

        if changed {

×

1140

                newVpc, err := c.config.KubeOvnClient.KubeovnV1().Vpcs().Update(context.Background(), vpc, metav1.UpdateOptions{})

×

1141

                if err != nil {

×

1142

                        klog.Errorf("failed to update vpc %s: %v", vpc.Name, err)

×

1143

                        return nil, err

×

1144

×

1145

                return newVpc, nil

×

1148

        return vpc, nil

×

1151

func convertPolicies(list []*kubeovnv1.PolicyRoute) string {

×

1152

        if list == nil {

×

1153

                return ""

×

1154

×

1156

        var (

×

1157

                res []byte

×

1158

                err error

×

1159

×

1160

×

1161

        if res, err = json.Marshal(list); err != nil {

×

1162

                klog.Errorf("failed to serialize policy routes %v , reason : %v", list, err)

×

1163

                return ""

×

1164

×

1165

        return string(res)

×

1168

func reversePolicies(origin string) []*kubeovnv1.PolicyRoute {

×

1169

        if origin == "" {

×

1170

                return nil

×

1171

×

1173

        var (

×

1174

                list []*kubeovnv1.PolicyRoute

×

1175

                err  error

×

1176

×

1177

×

1178

        if err = json.Unmarshal([]byte(origin), &list); err != nil {

×

1179

                klog.Errorf("failed to deserialize policy routes %v , reason : %v", list, err)

×

1180

                return nil

×

1181

×

1182

        return list

×

1185

func convertPolicy(origin kubeovnv1.RoutePolicy) string {

×

1186

        if origin == kubeovnv1.PolicyDst {

×

1187

                return ovnnb.LogicalRouterStaticRoutePolicyDstIP

×

1188

×

1189

        return ovnnb.LogicalRouterStaticRoutePolicySrcIP

×

1192

func reversePolicy(origin ovnnb.LogicalRouterStaticRoutePolicy) kubeovnv1.RoutePolicy {

×

1193

        if origin == ovnnb.LogicalRouterStaticRoutePolicyDstIP {

×

1194

                return kubeovnv1.PolicyDst

×

1195

×

1196

        return kubeovnv1.PolicySrc

×

1199

func (c *Controller) getVpcSubnets(vpc *kubeovnv1.Vpc) (subnets []string, defaultSubnet string, err error) {

×

1200

        subnets = []string{}

×

1201

        allSubnets, err := c.subnetsLister.List(labels.Everything())

×

1202

        if err != nil {

×

1203

                klog.Error(err)

×

1204

                return nil, "", err

×

1205

×

1207

        for _, subnet := range allSubnets {

×

1208

                if subnet.Spec.Vpc != vpc.Name || !subnet.DeletionTimestamp.IsZero() || !isOvnSubnet(subnet) {

×

1209

                        continue

×

1212

                subnets = append(subnets, subnet.Name)

×

1213

                if subnet.Spec.Default {

×

1214

                        defaultSubnet = subnet.Name

×

1215

×

1217

                if vpc.Name != util.DefaultVpc && vpc.Spec.DefaultSubnet != "" && vpc.Spec.DefaultSubnet == subnet.Name {

×

1218

                        defaultSubnet = vpc.Spec.DefaultSubnet

×

1219

×

1221

        return

×

1225

func (c *Controller) createVpcRouter(lr string) error {

×

1226

        if err := c.OVNNbClient.CreateLogicalRouter(lr); err != nil {

×

1227

                klog.Errorf("create logical router %s failed: %v", lr, err)

×

1228

                return err

×

1229

×

1231

        vpcRouter, err := c.OVNNbClient.GetLogicalRouter(lr, false)

×

1232

        if err != nil {

×

1233

                klog.Errorf("get logical router %s failed: %v", lr, err)

×

1234

                return err

×

1235

×

1237

        vpcRouter.Options = map[string]string{"always_learn_from_arp_request": "false", "dynamic_neigh_routers": "true", "mac_binding_age_threshold": "300"}

×

1238

        err = c.OVNNbClient.UpdateLogicalRouter(vpcRouter, &vpcRouter.Options)

×

1239

        if err != nil {

×

1240

                klog.Errorf("update logical router %s failed: %v", lr, err)

×

1241

                return err

×

1242

×

1243

        return nil

×

1247

func (c *Controller) deleteVpcRouter(lr string) error {

×

1248

        return c.OVNNbClient.DeleteLogicalRouter(lr)

×

1249

×

1251

func (c *Controller) handleAddVpcExternalSubnet(key, subnet string) error {

×

1252

        cachedSubnet, err := c.subnetsLister.Get(subnet)

×

1253

        if err != nil {

×

1254

                klog.Error(err)

×

1255

                return err

×

1256

×

1257

        lrpEipName := fmt.Sprintf("%s-%s", key, subnet)

×

1258

        cachedEip, err := c.ovnEipsLister.Get(lrpEipName)

×

1259

        var needCreateEip bool

×

1260

        if err != nil {

×

1261

                if !k8serrors.IsNotFound(err) {

×

1262

                        klog.Error(err)

×

1263

                        return err

×

1264

×

1265

                needCreateEip = true

×

1267

        var v4ip, v6ip, mac string

×

1268

        klog.V(3).Infof("create vpc lrp eip %s", lrpEipName)

×

1269

        if needCreateEip {

×

1270

                if v4ip, v6ip, mac, err = c.acquireIPAddress(subnet, lrpEipName, lrpEipName); err != nil {

×

1271

                        klog.Errorf("failed to acquire ip address for lrp eip %s, %v", lrpEipName, err)

×

1272

                        return err

×

1273

×

1274

                if err := c.createOrUpdateOvnEipCR(lrpEipName, subnet, v4ip, v6ip, mac, util.OvnEipTypeLRP); err != nil {

×

1275

                        klog.Errorf("failed to create ovn eip for lrp %s: %v", lrpEipName, err)

×

1276

                        return err

×

1277

×

1278

        } else {

×

1279

                v4ip = cachedEip.Spec.V4Ip

×

1280

                mac = cachedEip.Spec.MacAddress

×

1281

×

1282

        if v4ip == "" || mac == "" {

×

1283

                err := fmt.Errorf("lrp '%s' ip or mac should not be empty", lrpEipName)

×

1284

                klog.Error(err)

×

1285

                return err

×

1286

×

1288

        chassises := []string{}

×

1289

        sel, _ := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{MatchLabels: map[string]string{util.ExGatewayLabel: "true"}})

×

1290

        gwNodes, err := c.nodesLister.List(sel)

×

1291

        if err != nil {

×

1292

                klog.Errorf("failed to list external gw nodes, %v", err)

×

1293

                return err

×

1294

×

1295

        for _, gwNode := range gwNodes {

×

1296

                annoChassisName := gwNode.Annotations[util.ChassisAnnotation]

×

1297

                if annoChassisName == "" {

×

1298

                        err := fmt.Errorf("node %s has no chassis annotation, kube-ovn-cni not ready", gwNode.Name)

×

1299

                        klog.Error(err)

×

1300

                        return err

×

1301

×

1302

                klog.Infof("get node %s chassis: %s", gwNode.Name, annoChassisName)

×

1303

                chassis, err := c.OVNSbClient.GetChassis(annoChassisName, false)

×

1304

                if err != nil {

×

1305

                        klog.Errorf("failed to get node %s chassis: %s, %v", gwNode.Name, annoChassisName, err)

×

1306

                        return err

×

1307

×

1308

                chassises = append(chassises, chassis.Name)

×

1311

        if len(chassises) == 0 {

×

1312

                err := errors.New("no external gw nodes")

×

1313

                klog.Error(err)

×

1314

                return err

×

1315

×

1317

        v4ipCidr, err := util.GetIPAddrWithMask(v4ip, cachedSubnet.Spec.CIDRBlock)

×

1318

        if err != nil {

×

1319

                klog.Error(err)

×

1320

                return err

×

1321

×

1322

        lspName := fmt.Sprintf("%s-%s", subnet, key)

×

1323

        lrpName := fmt.Sprintf("%s-%s", key, subnet)

×

1324

×

1325

        if err := c.OVNNbClient.CreateLogicalPatchPort(subnet, key, lspName, lrpName, v4ipCidr, mac, chassises...); err != nil {

×

1326

                klog.Errorf("failed to connect router '%s' to external: %v", key, err)

×

1327

                return err

×

1328

×

1330

        cachedVpc, err := c.vpcsLister.Get(key)

×

1331

        if err != nil {

×

1332

                if k8serrors.IsNotFound(err) {

×

1333

                        return nil

×

1334

×

1335

                klog.Errorf("failed to get vpc %s, %v", key, err)

×

1336

                return err

×

1338

        if subnet == c.config.ExternalGatewaySwitch {

×

1339

                vpc := cachedVpc.DeepCopy()

×

1340

                vpc.Status.EnableExternal = cachedVpc.Spec.EnableExternal

×

1341

                bytes, err := vpc.Status.Bytes()

×

1342

                if err != nil {

×

1343

                        klog.Errorf("failed to marshal vpc status: %v", err)

×

1344

                        return err

×

1345

×

1346

                if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(),

×

1347

                        vpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {

×

1348

                        err := fmt.Errorf("failed to patch vpc %s status, %w", vpc.Name, err)

×

1349

                        klog.Error(err)

×

1350

                        return err

×

1351

×

1353

        if _, err = c.ovnEipsLister.Get(lrpEipName); err != nil {

×

1354

                err := fmt.Errorf("failed to get ovn eip %s, %w", lrpEipName, err)

×

1355

                klog.Error(err)

×

1356

                return err

×

1357

×

1358

        return nil

×

1361

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

×

1362

        vpc, err := c.vpcsLister.Get(key)

×

1363

        if err != nil {

×

1364

                if k8serrors.IsNotFound(err) {

×

1365

                        return nil

×

1366

×

1367

                klog.Errorf("failed to get vpc %s, %v", key, err)

×

1368

                return err

×

1370

        needUpdate := false

×

1371

        newStaticRoutes := make([]*kubeovnv1.StaticRoute, 0, len(vpc.Spec.StaticRoutes))

×

1372

        for _, route := range vpc.Spec.StaticRoutes {

×

1373

                if route.ECMPMode != util.StaticRouteBfdEcmp {

×

1374

                        newStaticRoutes = append(newStaticRoutes, route)

×

1375

                        needUpdate = true

×

1376

×

1379

        vpc.Spec.StaticRoutes = newStaticRoutes

×

1380

        if needUpdate {

×

1381

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

×

1382

                        klog.Errorf("failed to update vpc spec static route %s, %v", vpc.Name, err)

×

1383

                        return err

×

1384

×

1386

        if err = c.patchVpcBfdStatus(vpc.Name); err != nil {

×

1387

                klog.Errorf("failed to patch vpc %s, %v", vpc.Name, err)

×

1388

                return err

×

1389

×

1390

        return nil

×

1393

func (c *Controller) handleDelVpcExternalSubnet(key, subnet string) error {

×

1394

        lspName := fmt.Sprintf("%s-%s", subnet, key)

×

1395

        lrpName := fmt.Sprintf("%s-%s", key, subnet)

×

1396

        klog.V(3).Infof("delete vpc lrp %s", lrpName)

×

1397

        if err := c.OVNNbClient.RemoveLogicalPatchPort(lspName, lrpName); err != nil {

×

1398

                klog.Errorf("failed to disconnect router '%s' to external, %v", key, err)

×

1399

                return err

×

1400

×

1402

        if err := c.config.KubeOvnClient.KubeovnV1().OvnEips().Delete(context.Background(), lrpName, metav1.DeleteOptions{}); err != nil {

×

1403

                if !k8serrors.IsNotFound(err) {

×

1404

                        klog.Errorf("failed to delete ovn eip %s, %v", lrpName, err)

×

1405

                        return err

×

1406

×

1408

        if err := c.OVNNbClient.DeleteBFDByDstIP(lrpName, ""); err != nil {

×

1409

                klog.Error(err)

×

1410

                return err

×

1411

×

1412

        cachedVpc, err := c.vpcsLister.Get(key)

×

1413

        if err != nil {

×

1414

                if k8serrors.IsNotFound(err) {

×

1415

                        return nil

×

1416

×

1417

                klog.Errorf("failed to get vpc %s, %v", key, err)

×

1418

                return err

×

1420

        if subnet == c.config.ExternalGatewaySwitch {

×

1421

                vpc := cachedVpc.DeepCopy()

×

1422

                vpc.Status.EnableExternal = cachedVpc.Spec.EnableExternal

×

1423

                vpc.Status.EnableBfd = cachedVpc.Spec.EnableBfd

×

1424

                bytes, err := vpc.Status.Bytes()

×

1425

                if err != nil {

×

1426

                        klog.Errorf("failed to marshal vpc status: %v", err)

×

1427

                        return err

×

1428

×

1429

                if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(),

×

1430

                        vpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {

×

1431

                        if k8serrors.IsNotFound(err) {

×

1432

                                return nil

×

1433

×

1434

                        klog.Errorf("failed to patch vpc %s, %v", key, err)

×

1435

                        return err

×

1438

        return nil

×

1441

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

×

1442

        cachedVpc, err := c.vpcsLister.Get(key)

×

1443

        if err != nil {

×

1444

                if k8serrors.IsNotFound(err) {

×

1445

                        return nil

×

1446

×

1447

                klog.Errorf("failed to get vpc %s, %v", key, err)

×

1448

                return err

×

1451

        if cachedVpc.Status.EnableBfd != cachedVpc.Spec.EnableBfd {

×

1452

                status := cachedVpc.Status.DeepCopy()

×

1453

                status.EnableExternal = cachedVpc.Spec.EnableExternal

×

1454

                status.EnableBfd = cachedVpc.Spec.EnableBfd

×

1455

                bytes, err := status.Bytes()

×

1456

                if err != nil {

×

1457

                        klog.Errorf("failed to marshal vpc status: %v", err)

×

1458

                        return err

×

1459

×

1460

                if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(),

×

1461

                        cachedVpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {

×

1462

                        klog.Error(err)

×

1463

                        return err

×

1464

×

1466

        return nil

×

1469

func (c *Controller) getRouteTablesByVpc(vpc *kubeovnv1.Vpc) map[string][]*kubeovnv1.StaticRoute {

×

1470

        rtbs := make(map[string][]*kubeovnv1.StaticRoute)

×

1471

        for _, route := range vpc.Spec.StaticRoutes {

×

1472

                rtbs[route.RouteTable] = append(rtbs[route.RouteTable], route)

×

1473

×

1474

        return rtbs

×

1477

func (c *Controller) updateVpcAddExternalStatus(key string, addExternalStatus bool) error {

×

1478

        cachedVpc, err := c.vpcsLister.Get(key)

×

1479

        if err != nil {

×

1480

                klog.Errorf("failed to get vpc %s, %v", key, err)

×

1481

                return err

×

1482

×

1483

        vpc := cachedVpc.DeepCopy()

×

1484

        if addExternalStatus && vpc.Spec.ExtraExternalSubnets != nil {

×

1485

                sort.Strings(vpc.Spec.ExtraExternalSubnets)

×

1486

                vpc.Status.ExtraExternalSubnets = vpc.Spec.ExtraExternalSubnets

×

1487

        } else {

×

1488

                vpc.Status.ExtraExternalSubnets = nil

×

1489

×

1490

        bytes, err := vpc.Status.Bytes()

×

1491

        if err != nil {

×

1492

                klog.Errorf("failed to get vpc bytes, %v", err)

×

1493

                return err

×

1494

×

1495

        if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(),

×

1496

                vpc.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {

×

1497

                klog.Errorf("failed to patch vpc %s, %v", key, err)

×

1498

                return err

×

1499

×

1501

        return nil

×

kubeovn / kube-ovn / 13425935899

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

kubeovn / kube-ovn / 13425935899

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