kubeovn/kube-ovn | Build 22267836046 | pkg/controller/init.go | Coveralls

29

func (c *Controller) InitOVN() error {

×

30

        var err error

×

31

×

32

        // migrate vendor externalIDs to kube-ovn resources created in versions prior to v1.15.0

×

33

        // this must run before ACL cleanup to ensure existing resources are properly tagged

×

34

        if err = c.OVNNbClient.MigrateVendorExternalIDs(); err != nil {

×

35

                klog.Errorf("failed to migrate vendor externalIDs: %v", err)

×

36

                return err

×

37

×

42

        if err = c.OVNNbClient.MigrateACLTier(); err != nil {

×

43

                klog.Errorf("failed to migrate ACL tier: %v", err)

×

44

                return err

×

45

×

48

        if err = c.OVNNbClient.CleanNoParentKeyAcls(); err != nil {

×

49

                klog.Errorf("failed to clean all no parent key acls: %v", err)

×

50

                return err

×

51

×

53

        if err = c.InitDefaultVpc(); err != nil {

×

54

                klog.Errorf("init default vpc failed: %v", err)

×

55

                return err

×

56

×

58

        if err = c.initClusterRouter(); err != nil {

×

59

                klog.Errorf("init cluster router failed: %v", err)

×

60

                return err

×

61

×

63

        if c.config.EnableLb {

×

64

                if err = c.initLoadBalancer(); err != nil {

×

65

                        klog.Errorf("init load balancer failed: %v", err)

×

66

                        return err

×

67

×

70

        if err = c.initDefaultVlan(); err != nil {

×

71

                klog.Errorf("init default vlan failed: %v", err)

×

72

                return err

×

73

×

75

        if err = c.initNodeSwitch(); err != nil {

×

76

                klog.Errorf("init node switch failed: %v", err)

×

77

                return err

×

78

×

80

        if err = c.initDefaultLogicalSwitch(); err != nil {

×

81

                klog.Errorf("init default switch failed: %v", err)

×

82

                return err

×

83

×

85

        return nil

×

88

func (c *Controller) InitDefaultVpc() error {

×

89

        cachedVpc, err := c.vpcsLister.Get(c.config.ClusterRouter)

×

90

        if err != nil {

×

91

                if !k8serrors.IsNotFound(err) {

×

92

                        klog.Errorf("failed to get default vpc %q: %v", c.config.ClusterRouter, err)

×

93

                        return err

×

94

×

96

                vpc := &kubeovnv1.Vpc{

×

97

                        ObjectMeta: metav1.ObjectMeta{Name: c.config.ClusterRouter},

×

98

×

99

                cachedVpc, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Create(context.Background(), vpc, metav1.CreateOptions{})

×

100

                if err != nil {

×

101

                        klog.Errorf("failed to create default vpc %q: %v", c.config.ClusterRouter, err)

×

102

                        return err

×

103

×

107

        vpc := cachedVpc.DeepCopy()

×

108

        if !vpc.Status.Default || !vpc.Status.Standby ||

×

109

                vpc.Status.Router != c.config.ClusterRouter ||

×

110

                vpc.Status.DefaultLogicalSwitch != c.config.DefaultLogicalSwitch {

×

111

                vpc.Status.Standby = true

×

112

                vpc.Status.Default = true

×

113

                vpc.Status.Router = c.config.ClusterRouter

×

114

                vpc.Status.DefaultLogicalSwitch = c.config.DefaultLogicalSwitch

×

115

×

116

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

×

117

                        klog.Errorf("failed to update default vpc %q: %v", c.config.ClusterRouter, err)

×

118

                        return err

×

119

×

122

        return nil

×

126

func (c *Controller) initDefaultLogicalSwitch() error {

×

127

        subnet, err := c.subnetsLister.Get(c.config.DefaultLogicalSwitch)

×

128

        if err == nil {

×

129

                if subnet != nil && util.CheckProtocol(c.config.DefaultCIDR) != util.CheckProtocol(subnet.Spec.CIDRBlock) {

×

130

                        // single-stack upgrade to dual-stack

×

131

                        if util.CheckProtocol(c.config.DefaultCIDR) == kubeovnv1.ProtocolDual {

×

132

                                subnet := subnet.DeepCopy()

×

133

                                subnet.Spec.CIDRBlock = c.config.DefaultCIDR

×

134

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

×

135

                                        klog.Errorf("init format subnet %s failed: %v", c.config.DefaultLogicalSwitch, err)

×

136

                                        return err

×

137

×

140

                return nil

×

143

        if !k8serrors.IsNotFound(err) {

×

144

                klog.Errorf("get default subnet %s failed: %v", c.config.DefaultLogicalSwitch, err)

×

145

                return err

×

146

×

148

        defaultSubnet := kubeovnv1.Subnet{

×

149

                ObjectMeta: metav1.ObjectMeta{Name: c.config.DefaultLogicalSwitch},

×

150

                Spec: kubeovnv1.SubnetSpec{

×

151

                        Vpc:                 c.config.ClusterRouter,

×

152

                        Default:             true,

×

153

                        Provider:            util.OvnProvider,

×

154

                        CIDRBlock:           c.config.DefaultCIDR,

×

155

                        Gateway:             c.config.DefaultGateway,

×

156

                        GatewayNode:         "",

×

157

                        DisableGatewayCheck: !c.config.DefaultGatewayCheck,

×

158

                        ExcludeIps:          strings.Split(c.config.DefaultExcludeIps, ","),

×

159

                        NatOutgoing:         true,

×

160

                        GatewayType:         kubeovnv1.GWDistributedType,

×

161

                        Protocol:            util.CheckProtocol(c.config.DefaultCIDR),

×

162

                        EnableLb:            &c.config.EnableLb,

×

163

},

×

164

×

165

        if c.config.NetworkType == util.NetworkTypeVlan {

×

166

                defaultSubnet.Spec.Vlan = c.config.DefaultVlanName

×

167

                if c.config.DefaultLogicalGateway && c.config.DefaultU2OInterconnection {

×

168

                        err = errors.New("logicalGateway and u2oInterconnection can't be opened at the same time")

×

169

                        klog.Error(err)

×

170

                        return err

×

171

×

172

                defaultSubnet.Spec.LogicalGateway = c.config.DefaultLogicalGateway

×

173

                defaultSubnet.Spec.U2OInterconnection = c.config.DefaultU2OInterconnection

×

176

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

×

177

                klog.Errorf("failed to create default subnet %q: %v", c.config.DefaultLogicalSwitch, err)

×

178

                return err

×

179

×

180

        return nil

×

184

func (c *Controller) initNodeSwitch() error {

×

185

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

×

186

        if err == nil {

×

187

                if util.CheckProtocol(c.config.NodeSwitchCIDR) == kubeovnv1.ProtocolDual && util.CheckProtocol(subnet.Spec.CIDRBlock) != kubeovnv1.ProtocolDual {

×

188

                        // single-stack upgrade to dual-stack

×

189

                        subnet := subnet.DeepCopy()

×

190

                        subnet.Spec.CIDRBlock = c.config.NodeSwitchCIDR

×

191

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

×

192

                                klog.Errorf("init format subnet %s failed: %v", c.config.NodeSwitch, err)

×

193

                                return err

×

194

×

195

                } else {

×

196

                        c.config.NodeSwitchCIDR = subnet.Spec.CIDRBlock

×

197

×

198

                return nil

×

201

        if !k8serrors.IsNotFound(err) {

×

202

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

×

203

                return err

×

204

×

206

        nodeSubnet := kubeovnv1.Subnet{

×

207

                ObjectMeta: metav1.ObjectMeta{Name: c.config.NodeSwitch},

×

208

                Spec: kubeovnv1.SubnetSpec{

×

209

                        Vpc:                    c.config.ClusterRouter,

×

210

                        Default:                false,

×

211

                        Provider:               util.OvnProvider,

×

212

                        CIDRBlock:              c.config.NodeSwitchCIDR,

×

213

                        Gateway:                c.config.NodeSwitchGateway,

×

214

                        GatewayNode:            "",

×

215

                        ExcludeIps:             strings.Split(c.config.NodeSwitchGateway, ","),

×

216

                        Protocol:               util.CheckProtocol(c.config.NodeSwitchCIDR),

×

217

                        DisableInterConnection: true,

×

218

},

×

219

×

220

×

221

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

×

222

                klog.Errorf("failed to create node subnet %q: %v", c.config.NodeSwitch, err)

×

223

                return err

×

224

×

225

        return nil

×

229

func (c *Controller) initClusterRouter() error {

×

230

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

×

231

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

×

232

                return err

×

233

×

235

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

×

236

        if err != nil {

×

237

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

×

238

                return err

×

239

×

241

        lrOptions := make(map[string]string, len(lr.Options))

×

242

        maps.Copy(lrOptions, lr.Options)

×

243

        lrOptions["mac_binding_age_threshold"] = "300"

×

244

        lrOptions["dynamic_neigh_routers"] = "true"

×

245

        if !maps.Equal(lr.Options, lrOptions) {

×

246

                lr.Options = lrOptions

×

247

                if err = c.OVNNbClient.UpdateLogicalRouter(lr, &lr.Options); err != nil {

×

248

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

×

249

                        return err

×

250

×

253

        return nil

×

256

func (c *Controller) initLB(name, protocol string, sessionAffinity bool) error {

×

257

        protocol = strings.ToLower(protocol)

×

258

×

259

        var (

×

260

                selectFields string

×

261

                err          error

×

262

×

263

×

264

        if sessionAffinity {

×

265

                selectFields = ovnnb.LoadBalancerSelectionFieldsIPSrc

×

266

×

268

        if err = c.OVNNbClient.CreateLoadBalancer(name, protocol, selectFields); err != nil {

×

269

                klog.Errorf("create load balancer %s: %v", name, err)

×

270

                return err

×

271

×

273

        if sessionAffinity {

×

274

                if err = c.OVNNbClient.SetLoadBalancerAffinityTimeout(name, util.DefaultServiceSessionStickinessTimeout); err != nil {

×

275

                        klog.Errorf("failed to set affinity timeout of %s load balancer %s: %v", protocol, name, err)

×

276

                        return err

×

277

×

280

        err = c.OVNNbClient.SetLoadBalancerPreferLocalBackend(name, c.config.EnableOVNLBPreferLocal)

×

281

        if err != nil {

×

282

                klog.Errorf("failed to set prefer local backend for load balancer %s: %v", name, err)

×

283

                return err

×

284

×

286

        return nil

×

290

func (c *Controller) initLoadBalancer() error {

×

291

        vpcs, err := c.vpcsLister.List(labels.Everything())

×

292

        if err != nil {

×

293

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

×

294

                return err

×

295

×

297

        for _, cachedVpc := range vpcs {

×

298

                vpc := cachedVpc.DeepCopy()

×

299

                vpcLb := c.GenVpcLoadBalancer(vpc.Name)

×

300

                if err = c.initLB(vpcLb.TCPLoadBalancer, string(v1.ProtocolTCP), false); err != nil {

×

301

                        klog.Error(err)

×

302

                        return err

×

303

×

304

                if err = c.initLB(vpcLb.TCPSessLoadBalancer, string(v1.ProtocolTCP), true); err != nil {

×

305

                        klog.Error(err)

×

306

                        return err

×

307

×

308

                if err = c.initLB(vpcLb.UDPLoadBalancer, string(v1.ProtocolUDP), false); err != nil {

×

309

                        klog.Error(err)

×

310

                        return err

×

311

×

312

                if err = c.initLB(vpcLb.UDPSessLoadBalancer, string(v1.ProtocolUDP), true); err != nil {

×

313

                        klog.Error(err)

×

314

                        return err

×

315

×

316

                if err = c.initLB(vpcLb.SctpLoadBalancer, string(v1.ProtocolSCTP), false); err != nil {

×

317

                        klog.Error(err)

×

318

                        return err

×

319

×

320

                if err = c.initLB(vpcLb.SctpSessLoadBalancer, string(v1.ProtocolSCTP), true); err != nil {

×

321

                        klog.Error(err)

×

322

                        return err

×

323

×

325

                vpc.Status.TCPLoadBalancer = vpcLb.TCPLoadBalancer

×

326

                vpc.Status.TCPSessionLoadBalancer = vpcLb.TCPSessLoadBalancer

×

327

                vpc.Status.UDPLoadBalancer = vpcLb.UDPLoadBalancer

×

328

                vpc.Status.UDPSessionLoadBalancer = vpcLb.UDPSessLoadBalancer

×

329

                vpc.Status.SctpLoadBalancer = vpcLb.SctpLoadBalancer

×

330

                vpc.Status.SctpSessionLoadBalancer = vpcLb.SctpSessLoadBalancer

×

331

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

×

332

                if err != nil {

×

333

                        klog.Error(err)

×

334

                        return err

×

335

×

336

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

×

337

                        klog.Error(err)

×

338

                        return err

×

339

×

341

        return nil

×

344

func (c *Controller) InitIPAM() error {

×

345

        start := time.Now()

×

346

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

×

347

        if err != nil {

×

348

                klog.Errorf("failed to list subnet: %v", err)

×

349

                return err

×

350

×

351

        subnetProviderMaps := make(map[string]string, len(subnets))

×

352

        for _, subnet := range subnets {

×

353

                klog.Infof("Init subnet %s", subnet.Name)

×

354

                subnetProviderMaps[subnet.Name] = subnet.Spec.Provider

×

355

                if err := c.ipam.AddOrUpdateSubnet(subnet.Name, subnet.Spec.CIDRBlock, subnet.Spec.Gateway, subnet.Spec.ExcludeIps); err != nil {

×

356

                        klog.Errorf("failed to init subnet %s: %v", subnet.Name, err)

×

357

×

359

                u2oInterconnName := fmt.Sprintf(util.U2OInterconnName, subnet.Spec.Vpc, subnet.Name)

×

360

                u2oInterconnLrpName := fmt.Sprintf("%s-%s", subnet.Spec.Vpc, subnet.Name)

×

361

                if subnet.Status.U2OInterconnectionIP != "" {

×

362

                        var mac *string

×

363

                        klog.Infof("Init U2O for subnet %s", subnet.Name)

×

364

                        if subnet.Status.U2OInterconnectionMAC != "" {

×

365

                                mac = new(subnet.Status.U2OInterconnectionMAC)

×

366

                        } else {

×

367

                                lrp, err := c.OVNNbClient.GetLogicalRouterPort(u2oInterconnLrpName, true)

×

368

                                if err != nil {

×

369

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

×

370

                                        return err

×

371

×

372

                                if lrp != nil {

×

373

                                        mac = new(lrp.MAC)

×

374

×

376

                        if _, _, _, err = c.ipam.GetStaticAddress(u2oInterconnName, u2oInterconnLrpName, subnet.Status.U2OInterconnectionIP, mac, subnet.Name, true); err != nil {

×

377

                                klog.Errorf("failed to init subnet %q u2o interconnection ip to ipam %v", subnet.Name, err)

×

378

×

382

        ippools, err := c.ippoolLister.List(labels.Everything())

×

383

        if err != nil {

×

384

                klog.Errorf("failed to list ippool: %v", err)

×

385

                return err

×

386

×

387

        for _, ippool := range ippools {

×

388

                if err = c.ipam.AddOrUpdateIPPool(ippool.Spec.Subnet, ippool.Name, ippool.Spec.IPs); err != nil {

×

389

                        klog.Errorf("failed to init ippool %s: %v", ippool.Name, err)

×

390

×

393

        klog.Infof("Init IPAM from StatefulSet or VM IP CR")

×

394

        ips, err := c.ipsLister.List(labels.Everything())

×

395

        if err != nil {

×

396

                klog.Errorf("failed to list IPs: %v", err)

×

397

                return err

×

398

×

400

        for _, ip := range ips {

×

401

                if !ip.DeletionTimestamp.IsZero() {

×

402

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

×

403

                        c.updateIPQueue.Add(ip.Name)

×

404

                        continue

×

407

                if ip.Spec.PodType != util.KindStatefulSet &&

×

408

                        ip.Spec.PodType != util.KindVirtualMachine {

×

409

                        continue

×

412

                var ipamKey string

×

413

                if ip.Spec.Namespace != "" {

×

414

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

×

415

                } else {

×

416

                        ipamKey = util.NodeLspName(ip.Spec.PodName)

×

417

×

418

                if _, _, _, err = c.ipam.GetStaticAddress(ipamKey, ip.Name, ip.Spec.IPAddress, &ip.Spec.MacAddress, ip.Spec.Subnet, true); err != nil {

×

419

                        klog.Errorf("failed to init IPAM from IP CR %s: %v", ip.Name, err)

×

420

×

423

        klog.Infof("Init IPAM from pod")

×

424

        pods, err := c.podsLister.List(labels.Everything())

×

425

        if err != nil {

×

426

                klog.Errorf("failed to list pods: %v", err)

×

427

                return err

×

428

×

429

        for _, pod := range pods {

×

430

                if pod.Spec.HostNetwork {

×

431

                        continue

×

434

                isAlive := isPodAlive(pod)

×

435

                isStsPod, _, _ := isStatefulSetPod(pod)

×

436

                if !isAlive && !isStsPod {

×

437

                        continue

×

440

                podNets, err := c.getPodKubeovnNets(pod)

×

441

                if err != nil {

×

442

                        klog.Errorf("failed to get pod kubeovn nets %s.%s address %s: %v", pod.Name, pod.Namespace, pod.Annotations[util.IPAddressAnnotation], err)

×

443

                        continue

×

446

                podType := getPodType(pod)

×

447

                podName := c.getNameByPod(pod)

×

448

                key := cache.NewObjectName(pod.Namespace, podName).String()

×

449

                for _, podNet := range podNets {

×

450

                        if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "true" {

×

451

                                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)

×

452

                                ip := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]

×

453

                                mac := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]

×

454

                                if ip == "" {

×

455

                                        klog.Warningf("pod %s/%s has empty IP annotation for provider %s, skip IPAM init", pod.Namespace, podName, podNet.ProviderName)

×

456

                                        continue

×

458

                                _, _, _, err := c.ipam.GetStaticAddress(key, portName, ip, &mac, podNet.Subnet.Name, true)

×

459

                                if err != nil {

×

460

                                        klog.Errorf("failed to init pod %s.%s address %s: %v", podName, pod.Namespace, ip, err)

×

461

                                } else {

×

462

                                        err = c.createOrUpdateIPCR(portName, podName, ip, mac, podNet.Subnet.Name, pod.Namespace, pod.Spec.NodeName, podType)

×

463

                                        if err != nil {

×

464

                                                klog.Errorf("failed to create/update ips CR %s.%s with ip address %s: %v", podName, pod.Namespace, ip, err)

×

465

×

473

        klog.Infof("Init IPAM from vip CR")

×

474

        vips, err := c.virtualIpsLister.List(labels.Everything())

×

475

        if err != nil {

×

476

                klog.Errorf("failed to list vips: %v", err)

×

477

                return err

×

478

×

479

        for _, vip := range vips {

×

480

                provider, ok := subnetProviderMaps[vip.Spec.Subnet]

×

481

                if !ok {

×

482

                        klog.Errorf("failed to find subnet %s for vip %s", vip.Spec.Subnet, vip.Name)

×

483

                        continue

×

485

                portName := ovs.PodNameToPortName(vip.Name, vip.Spec.Namespace, provider)

×

486

                if _, _, _, err = c.ipam.GetStaticAddress(vip.Name, portName, vip.Status.V4ip, &vip.Status.Mac, vip.Spec.Subnet, true); err != nil {

×

487

                        klog.Errorf("failed to init ipam from vip cr %s: %v", vip.Name, err)

×

488

×

491

        klog.Infof("Init IPAM from iptables EIP CR")

×

492

        eips, err := c.iptablesEipsLister.List(labels.Everything())

×

493

        if err != nil {

×

494

                klog.Errorf("failed to list EIPs: %v", err)

×

495

                return err

×

496

×

497

        for _, eip := range eips {

×

498

                externalNetwork := util.GetExternalNetwork(eip.Spec.ExternalSubnet)

×

499

                if _, _, _, err = c.ipam.GetStaticAddress(eip.Name, eip.Name, eip.Status.IP, &eip.Spec.MacAddress, externalNetwork, true); err != nil {

×

500

                        klog.Errorf("failed to init ipam from iptables eip cr %s: %v", eip.Name, err)

×

501

×

504

        klog.Infof("Init IPAM from ovn EIP CR")

×

505

        oeips, err := c.ovnEipsLister.List(labels.Everything())

×

506

        if err != nil {

×

507

                klog.Errorf("failed to list ovn eips: %v", err)

×

508

                return err

×

509

×

510

        for _, oeip := range oeips {

×

511

                if _, _, _, err = c.ipam.GetStaticAddress(oeip.Name, oeip.Name, oeip.Status.V4Ip, &oeip.Status.MacAddress, oeip.Spec.ExternalSubnet, true); err != nil {

×

512

                        klog.Errorf("failed to init ipam from ovn eip cr %s: %v", oeip.Name, err)

×

513

×

516

        klog.Infof("Init IPAM from node annotation")

×

517

        nodes, err := c.nodesLister.List(labels.Everything())

×

518

        if err != nil {

×

519

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

×

520

                return err

×

521

×

522

        for _, node := range nodes {

×

523

                if node.Annotations[util.AllocatedAnnotation] == "true" {

×

524

                        portName := util.NodeLspName(node.Name)

×

525

                        mac := node.Annotations[util.MacAddressAnnotation]

×

526

                        v4IP, v6IP, _, err := c.ipam.GetStaticAddress(portName, portName,

×

527

                                node.Annotations[util.IPAddressAnnotation], &mac,

×

528

                                node.Annotations[util.LogicalSwitchAnnotation], true)

×

529

                        if err != nil {

×

530

                                klog.Errorf("failed to init node %s.%s address %s: %v", node.Name, node.Namespace, node.Annotations[util.IPAddressAnnotation], err)

×

531

×

532

                        if v4IP != "" && v6IP != "" {

×

533

                                node.Annotations[util.IPAddressAnnotation] = util.GetStringIP(v4IP, v6IP)

×

534

×

538

        klog.Infof("take %.2f seconds to initialize IPAM", time.Since(start).Seconds())

×

539

        return nil

×

542

func (c *Controller) initDefaultProviderNetwork() error {

×

543

        _, err := c.providerNetworksLister.Get(c.config.DefaultProviderName)

×

544

        if err == nil {

×

545

                return nil

×

546

×

547

        if !k8serrors.IsNotFound(err) {

×

548

                klog.Errorf("failed to get default provider network %s: %v", c.config.DefaultProviderName, err)

×

549

                return err

×

550

×

552

        nodes, err := c.nodesLister.List(labels.Everything())

×

553

        if err != nil {

×

554

                klog.Errorf("failed to get nodes: %v", err)

×

555

                return err

×

556

×

558

        pn := kubeovnv1.ProviderNetwork{

×

559

                ObjectMeta: metav1.ObjectMeta{

×

560

                        Name: c.config.DefaultProviderName,

×

561

},

×

562

                Spec: kubeovnv1.ProviderNetworkSpec{

×

563

                        DefaultInterface: c.config.DefaultHostInterface,

×

564

                        ExchangeLinkName: c.config.DefaultExchangeLinkName,

×

565

},

×

566

×

567

×

568

        excludeAnno := fmt.Sprintf(util.ProviderNetworkExcludeTemplate, c.config.DefaultProviderName)

×

569

        interfaceAnno := fmt.Sprintf(util.ProviderNetworkInterfaceTemplate, c.config.DefaultProviderName)

×

570

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

×

571

        for _, node := range nodes {

×

572

                if len(node.Annotations) == 0 {

×

573

                        continue

×

576

                if node.Annotations[excludeAnno] == "true" {

×

577

                        pn.Spec.ExcludeNodes = append(pn.Spec.ExcludeNodes, node.Name)

×

578

                        patchNodes = append(patchNodes, node.Name)

×

579

                } else if s := node.Annotations[interfaceAnno]; s != "" {

×

580

                        var index *int

×

581

                        for i := range pn.Spec.CustomInterfaces {

×

582

                                if pn.Spec.CustomInterfaces[i].Interface == s {

×

583

                                        index = &i

×

584

                                        break

×

587

                        if index != nil {

×

588

                                pn.Spec.CustomInterfaces[*index].Nodes = append(pn.Spec.CustomInterfaces[*index].Nodes, node.Name)

×

589

                        } else {

×

590

                                ci := kubeovnv1.CustomInterface{Interface: s, Nodes: []string{node.Name}}

×

591

                                pn.Spec.CustomInterfaces = append(pn.Spec.CustomInterfaces, ci)

×

592

×

593

                        patchNodes = append(patchNodes, node.Name)

×

597

        defer func() {

×

598

                if err != nil {

×

599

                        return

×

600

×

603

                patch := util.KVPatch{excludeAnno: nil, interfaceAnno: nil}

×

604

                for _, node := range patchNodes {

×

605

                        if err := util.PatchAnnotations(c.config.KubeClient.CoreV1().Nodes(), node, patch); err != nil {

×

606

                                klog.Errorf("failed to patch node %s: %v", node, err)

×

607

×

611

        _, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().Create(context.Background(), &pn, metav1.CreateOptions{})

×

612

        if err != nil {

×

613

                klog.Errorf("failed to create provider network %s: %v", c.config.DefaultProviderName, err)

×

614

                return err

×

615

×

616

        return nil

×

619

func (c *Controller) initDefaultVlan() error {

×

620

        if c.config.NetworkType != util.NetworkTypeVlan {

×

621

                return nil

×

622

×

624

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

×

625

                klog.Error(err)

×

626

                return err

×

627

×

629

        _, err := c.vlansLister.Get(c.config.DefaultVlanName)

×

630

        if err == nil {

×

631

                return nil

×

632

×

634

        if !k8serrors.IsNotFound(err) {

×

635

                klog.Errorf("get default vlan %s failed: %v", c.config.DefaultVlanName, err)

×

636

                return err

×

637

×

639

        if c.config.DefaultVlanID < 0 || c.config.DefaultVlanID > 4095 {

×

640

                return errors.New("the default vlan id is not between 1-4095")

×

641

×

643

        defaultVlan := kubeovnv1.Vlan{

×

644

                ObjectMeta: metav1.ObjectMeta{Name: c.config.DefaultVlanName},

×

645

                Spec: kubeovnv1.VlanSpec{

×

646

                        ID:       c.config.DefaultVlanID,

×

647

                        Provider: c.config.DefaultProviderName,

×

648

},

×

649

×

650

×

651

        _, err = c.config.KubeOvnClient.KubeovnV1().Vlans().Create(context.Background(), &defaultVlan, metav1.CreateOptions{})

×

652

        if err != nil {

×

653

                klog.Errorf("failed to create vlan %s: %v", defaultVlan.Name, err)

×

654

                return err

×

655

×

656

        return nil

×

659

func (c *Controller) syncIPCR() error {

×

660

        klog.Info("start to sync ips")

×

661

        ips, err := c.ipsLister.List(labels.Everything())

×

662

        if err != nil {

×

663

                if k8serrors.IsNotFound(err) {

×

664

                        return nil

×

665

×

666

                klog.Error(err)

×

667

                return err

×

670

        ipMap := strset.New(c.getVMLsps()...)

×

671

        for _, ip := range ips {

×

672

                if !ip.DeletionTimestamp.IsZero() {

×

673

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

×

674

                        c.updateIPQueue.Add(ip.Name)

×

675

                        continue

×

677

                changed := false

×

678

                ip = ip.DeepCopy()

×

679

                if ipMap.Has(ip.Name) && ip.Spec.PodType == "" {

×

680

                        ip.Spec.PodType = util.KindVirtualMachine

×

681

                        changed = true

×

682

×

684

                v4IP, v6IP := util.SplitStringIP(ip.Spec.IPAddress)

×

685

                if ip.Spec.V4IPAddress == v4IP && ip.Spec.V6IPAddress == v6IP && !changed {

×

686

                        continue

×

689

                ip.Spec.V4IPAddress = v4IP

×

690

                ip.Spec.V6IPAddress = v6IP

×

691

                _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Update(context.Background(), ip, metav1.UpdateOptions{})

×

692

                if err != nil {

×

693

                        klog.Errorf("failed to sync crd ip %s: %v", ip.Spec.IPAddress, err)

×

694

                        return err

×

695

×

697

        return nil

×

700

func (c *Controller) syncSubnetCR() error {

×

701

        klog.Info("start to sync subnets")

×

702

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

×

703

        if err != nil {

×

704

                if k8serrors.IsNotFound(err) {

×

705

                        return nil

×

706

×

707

                klog.Error(err)

×

708

                return err

×

710

        for _, cachedSubnet := range subnets {

×

711

                subnet := cachedSubnet.DeepCopy()

×

712

                if !subnet.Status.IsReady() {

×

713

                        klog.Warningf("subnet %s is not ready", subnet.Name)

×

714

                        continue

×

716

                subnet, err = c.calcSubnetStatusIP(subnet)

×

717

                if err != nil {

×

718

                        klog.Errorf("failed to calculate subnet %s used ip: %v", cachedSubnet.Name, err)

×

719

                        return err

×

720

×

723

                if subnet.Spec.GatewayType == kubeovnv1.GWCentralizedType && !subnet.Spec.EnableEcmp && subnet.Spec.EnableEcmp != c.config.EnableEcmp {

×

724

                        subnet, err = c.subnetsLister.Get(subnet.Name)

×

725

                        if err != nil {

×

726

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

×

727

                                return err

×

728

×

730

                        subnet.Spec.EnableEcmp = c.config.EnableEcmp

×

731

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

×

732

                                klog.Errorf("failed to sync subnet spec enableEcmp with kube-ovn-controller config enableEcmp %s: %v", subnet.Name, err)

×

733

                                return err

×

734

×

737

        return nil

×

740

func (c *Controller) syncVpcNatGatewayCR() error {

×

741

        klog.Info("start to sync crd vpc nat gw")

×

742

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

×

743

        if err != nil {

×

744

                klog.Errorf("failed to list vpc nat gateway, %v", err)

×

745

                return err

×

746

×

747

        if len(gws) == 0 {

×

748

                return nil

×

749

×

751

        cm, err := c.configMapsLister.ConfigMaps(c.config.PodNamespace).Get(util.VpcNatGatewayConfig)

×

752

        if err != nil && !k8serrors.IsNotFound(err) {

×

753

                klog.Errorf("failed to get config map %s, %v", util.VpcNatGatewayConfig, err)

×

754

                return err

×

755

×

756

        if k8serrors.IsNotFound(err) || cm.Data["enable-vpc-nat-gw"] == "false" {

×

757

                return nil

×

758

×

760

        cm, err = c.configMapsLister.ConfigMaps(c.config.PodNamespace).Get(util.VpcNatConfig)

×

761

        if err != nil {

×

762

                if k8serrors.IsNotFound(err) {

×

763

                        klog.Errorf("should set config map for vpc-nat-gateway %s, %v", util.VpcNatConfig, err)

×

764

                        return err

×

765

×

766

                klog.Errorf("failed to get config map %s, %v", util.VpcNatConfig, err)

×

767

                return err

×

770

        if cm.Data["image"] == "" {

×

771

                err = errors.New("should set image for vpc-nat-gateway pod")

×

772

                klog.Error(err)

×

773

                return err

×

774

×

776

        for _, gw := range gws {

×

777

                if err := c.updateCrdNatGwLabels(gw.Name, ""); err != nil {

×

778

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

×

779

                        return err

×

780

×

782

        return nil

×

785

func (c *Controller) syncVlanCR() error {

×

786

        klog.Info("start to sync vlans")

×

787

        vlans, err := c.vlansLister.List(labels.Everything())

×

788

        if err != nil {

×

789

                if k8serrors.IsNotFound(err) {

×

790

                        return nil

×

791

×

792

                klog.Error(err)

×

793

                return err

×

796

        for _, vlan := range vlans {

×

797

                var needUpdate bool

×

798

                newVlan := vlan.DeepCopy()

×

799

                if newVlan.Spec.VlanID != 0 && newVlan.Spec.ID == 0 {

×

800

                        newVlan.Spec.ID = newVlan.Spec.VlanID

×

801

                        newVlan.Spec.VlanID = 0

×

802

                        needUpdate = true

×

803

×

804

                if newVlan.Spec.ProviderInterfaceName != "" && newVlan.Spec.Provider == "" {

×

805

                        newVlan.Spec.Provider = newVlan.Spec.ProviderInterfaceName

×

806

                        newVlan.Spec.ProviderInterfaceName = ""

×

807

                        needUpdate = true

×

808

×

809

                if needUpdate {

×

810

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

×

811

                                klog.Errorf("failed to update spec of vlan %s: %v", newVlan.Name, err)

×

812

                                return err

×

813

×

817

        return nil

×

820

func (c *Controller) batchMigrateNodeRoute(nodes []*v1.Node) error {

×

821

        start := time.Now()

×

822

        addPolicies := make([]*kubeovnv1.PolicyRoute, 0)

×

823

        delPolicies := make([]*kubeovnv1.PolicyRoute, 0)

×

824

        staticRoutes := make([]*kubeovnv1.StaticRoute, 0)

×

825

        externalIDsMap := make(map[string]map[string]string)

×

826

        delAsNames := make([]string, 0)

×

827

        for _, node := range nodes {

×

828

                if node.Annotations[util.AllocatedAnnotation] != "true" {

×

829

                        continue

×

831

                nodeName := node.Name

×

832

                nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)

×

833

                joinAddrV4, joinAddrV6 := util.SplitStringIP(node.Annotations[util.IPAddressAnnotation])

×

834

                if nodeIPv4 != "" && joinAddrV4 != "" {

×

835

                        buildNodeRoute(4, nodeName, joinAddrV4, nodeIPv4, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)

×

836

×

837

                if nodeIPv6 != "" && joinAddrV6 != "" {

×

838

                        buildNodeRoute(6, nodeName, joinAddrV6, nodeIPv6, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)

×

839

×

842

        if err := c.batchAddPolicyRouteToVpc(c.config.ClusterRouter, addPolicies, externalIDsMap); err != nil {

×

843

                klog.Errorf("failed to batch add logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)

×

844

                return err

×

845

×

846

        if err := c.batchDeleteStaticRouteFromVpc(c.config.ClusterRouter, staticRoutes); err != nil {

×

847

                klog.Errorf("failed to batch delete  obsolete logical router static route for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)

×

848

                return err

×

849

×

850

        if err := c.batchDeletePolicyRouteFromVpc(c.config.ClusterRouter, delPolicies); err != nil {

×

851

                klog.Errorf("failed to batch delete obsolete logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)

×

852

                return err

×

853

×

854

        if err := c.OVNNbClient.BatchDeleteAddressSetByNames(delAsNames); err != nil {

×

855

                klog.Errorf("failed to batch delete obsolete address set for asNames %v nodes %d: %v", delAsNames, len(nodes), err)

×

856

                return err

×

857

×

858

        klog.V(3).Infof("take to %v batch migrate node route for router: %s priority: %d add policy len: %d externalID len: %d del policy len: %d del address set len: %d",

×

859

                time.Since(start), c.config.ClusterRouter, util.NodeRouterPolicyPriority, len(addPolicies), len(externalIDsMap), len(delPolicies), len(delAsNames))

×

860

×

861

        return nil

×

864

func buildNodeRoute(af int, nodeName, nexthop, ip string, addPolicies, delPolicies *[]*kubeovnv1.PolicyRoute, staticRoutes *[]*kubeovnv1.StaticRoute, externalIDsMap map[string]map[string]string, delAsNames *[]string) {

×

865

        var (

×

866

                match       = fmt.Sprintf("ip%d.dst == %s", af, ip)

×

867

                action      = kubeovnv1.PolicyRouteActionReroute

×

868

                externalIDs = map[string]string{

×

869

                        "vendor": util.CniTypeName,

×

870

                        "node":   nodeName,

×

871

×

872

×

873

        *addPolicies = append(*addPolicies, &kubeovnv1.PolicyRoute{

×

874

                Priority:  util.NodeRouterPolicyPriority,

×

875

                Match:     match,

×

876

                Action:    action,

×

877

                NextHopIP: nexthop,

×

878

})

×

879

        externalIDsMap[buildExternalIDsMapKey(match, string(action), util.NodeRouterPolicyPriority)] = externalIDs

×

880

        *staticRoutes = append(*staticRoutes, &kubeovnv1.StaticRoute{

×

881

                Policy:     kubeovnv1.PolicyDst,

×

882

                RouteTable: util.MainRouteTable,

×

883

                NextHopIP:  "",

×

884

                CIDR:       ip,

×

885

})

×

886

        asName := nodeUnderlayAddressSetName(nodeName, af)

×

887

        obsoleteMatch := fmt.Sprintf("ip%d.dst == %s && ip%d.src != $%s", af, ip, af, asName)

×

888

        *delPolicies = append(*delPolicies, &kubeovnv1.PolicyRoute{

×

889

                Match:    obsoleteMatch,

×

890

                Priority: util.NodeRouterPolicyPriority,

×

891

})

×

892

        *delAsNames = append(*delAsNames, asName)

×

893

×

895

func (c *Controller) syncNodeRoutes() error {

×

896

        nodes, err := c.nodesLister.List(labels.Everything())

×

897

        if err != nil {

×

898

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

×

899

                return err

×

900

×

902

        if err := c.batchMigrateNodeRoute(nodes); err != nil {

×

903

                klog.Errorf("failed to batch migrate node routes: %v", err)

×

904

                return err

×

905

×

907

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

×

908

                klog.Errorf("failed to add static route for node gateway")

×

909

                return err

×

910

×

911

        return nil

×

914

func (c *Controller) initNodeChassis() error {

×

915

        nodes, err := c.nodesLister.List(labels.Everything())

×

916

        if err != nil {

×

917

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

×

918

                return err

×

919

×

920

        chassises, err := c.OVNSbClient.GetKubeOvnChassises()

×

921

        if err != nil {

×

922

                klog.Errorf("failed to get chassis nodes: %v", err)

×

923

                return err

×

924

×

925

        chassisNodes := make(map[string]string, len(*chassises))

×

926

        for _, chassis := range *chassises {

×

927

                chassisNodes[chassis.Name] = chassis.Hostname

×

928

×

929

        for _, node := range nodes {

×

930

                if err := c.UpdateChassisTag(node); err != nil {

×

931

                        klog.Error(err)

×

932

                        if _, ok := err.(*ErrChassisNotFound); !ok {

×

933

                                return err

×

934

×

937

        return nil

×

940

func migrateFinalizers(c client.Client, list client.ObjectList, getObjectItem func(int) (client.Object, client.Object)) error {

×

941

        if err := c.List(context.Background(), list); err != nil {

×

942

                klog.Errorf("failed to list objects: %v", err)

×

943

                return err

×

944

×

946

        var i int

×

947

        var cachedObj, patchedObj client.Object

×

948

        for {

×

949

                if cachedObj, patchedObj = getObjectItem(i); cachedObj == nil {

×

950

                        break

×

952

                if !controllerutil.ContainsFinalizer(cachedObj, util.DeprecatedFinalizerName) {

×

953

i++

×

954

                        continue

×

956

                controllerutil.RemoveFinalizer(patchedObj, util.DeprecatedFinalizerName)

×

957

                if cachedObj.GetDeletionTimestamp() == nil {

×

958

                        // if the object is not being deleted, add the new finalizer

×

959

                        controllerutil.AddFinalizer(patchedObj, util.KubeOVNControllerFinalizer)

×

960

×

961

                if err := c.Patch(context.Background(), patchedObj, client.MergeFrom(cachedObj)); client.IgnoreNotFound(err) != nil {

×

962

                        klog.Errorf("failed to sync finalizers for %s %s: %v",

×

963

                                patchedObj.GetObjectKind().GroupVersionKind().Kind,

×

964

                                cache.MetaObjectToName(patchedObj), err)

×

965

                        return err

×

966

×

967

i++

×

970

        return nil

×

973

func (c *Controller) syncFinalizers() error {

×

974

        cl, err := client.New(config.GetConfigOrDie(), client.Options{})

×

975

        if err != nil {

×

976

                klog.Errorf("failed to create client: %v", err)

×

977

                return err

×

978

×

981

        klog.Info("start to sync finalizers")

×

982

        if err := c.syncIPFinalizer(cl); err != nil {

×

983

                klog.Errorf("failed to sync ip finalizer: %v", err)

×

984

                return err

×

985

×

986

        if err := c.syncIPPoolFinalizer(cl); err != nil {

×

987

                klog.Errorf("failed to sync ippool finalizer: %v", err)

×

988

                return err

×

989

×

990

        if err := c.syncOvnDnatFinalizer(cl); err != nil {

×

991

                klog.Errorf("failed to sync ovn dnat finalizer: %v", err)

×

992

                return err

×

993

×

994

        if err := c.syncOvnEipFinalizer(cl); err != nil {

×

995

                klog.Errorf("failed to sync ovn eip finalizer: %v", err)

×

996

                return err

×

997

×

998

        if err := c.syncOvnFipFinalizer(cl); err != nil {

×

999

                klog.Errorf("failed to sync ovn fip finalizer: %v", err)

×

1000

                return err

×

1001

×

1002

        if err := c.syncOvnSnatFinalizer(cl); err != nil {

×

1003

                klog.Errorf("failed to sync ovn snat finalizer: %v", err)

×

1004

                return err

×

1005

×

1006

        if err := c.syncQoSPolicyFinalizer(cl); err != nil {

×

1007

                klog.Errorf("failed to sync qos policy finalizer: %v", err)

×

1008

                return err

×

1009

×

1010

        if err := c.syncSubnetFinalizer(cl); err != nil {

×

1011

                klog.Errorf("failed to sync subnet finalizer: %v", err)

×

1012

                return err

×

1013

×

1014

        if err := c.syncVipFinalizer(cl); err != nil {

×

1015

                klog.Errorf("failed to sync vip finalizer: %v", err)

×

1016

                return err

×

1017

×

1018

        if err := c.syncIptablesEipFinalizer(cl); err != nil {

×

1019

                klog.Errorf("failed to sync iptables eip finalizer: %v", err)

×

1020

                return err

×

1021

×

1022

        if err := c.syncIptablesFipFinalizer(cl); err != nil {

×

1023

                klog.Errorf("failed to sync iptables fip finalizer: %v", err)

×

1024

                return err

×

1025

×

1026

        if err := c.syncIptablesDnatFinalizer(cl); err != nil {

×

1027

                klog.Errorf("failed to sync iptables dnat finalizer: %v", err)

×

1028

                return err

×

1029

×

1030

        if err := c.syncIptablesSnatFinalizer(cl); err != nil {

×

1031

                klog.Errorf("failed to sync iptables snat finalizer: %v", err)

×

1032

                return err

×

1033

×

1034

        klog.Info("sync finalizers done")

×

1035

        return nil

×

kubeovn / kube-ovn / 22267836046

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

kubeovn / kube-ovn / 22267836046

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