27017056817

Committed 05 Jun 2026 01:15PM UTC coverage: 25.681% (+0.09%) from 25.587%

Build # 27017056817

Build Type

push

github

Committed by

web-flow

Commit Message

perf(daemon): list subnets and pods once per gateway setter (#6827)

* perf(daemon): list subnets and pods once per gateway setter

runGateway runs every 3s, and each of setIPSet/setPolicyRouting/setIptables
re-listed the whole subnet cache up to 15 times per dual-stack tick (and pods
2-4 times). List subnets/pods once per setter and thread the slice through the
helpers, dropping the now-redundant error returns whose only source was List.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>

* perf(daemon): log getCidrByProtocol errors instead of dropping them

Address review feedback: getSubnetsNeedNAT, getSubnetsDistributedGateway and
getDefaultVpcSubnetsCIDR silently skipped subnets when getCidrByProtocol
failed, and StartTProxyTCPPortProbe swallowed getTProxyConditionPod errors.
Log these errors to match the existing handling in reconcileNatOutGoingPolicyIPset
and generateNatOutgoingPolicyChainRules; the error path only triggers on
malformed CIDRs, not on normal single-stack subnets.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>

---------

Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Coverage Stats

0 of 80 new or added lines in 3 files covered. (0.0%)

433 existing lines in 5 files now uncovered.

14867 of 57891 relevant lines covered (25.68%)

0.3 hits per line

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

5.45

/pkg/controller/init.go

package controller

import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "maps"
        "strings"
        "time"

        "github.com/scylladb/go-set/strset"
        v1 "k8s.io/api/core/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/labels"
        "k8s.io/apimachinery/pkg/types"
        "k8s.io/client-go/tools/cache"
        "k8s.io/klog/v2"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/client/config"
        "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

        kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
        "github.com/kubeovn/kube-ovn/pkg/ovs"
        "github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
        "github.com/kubeovn/kube-ovn/pkg/util"
)

func (c *Controller) InitOVN() error {
        var err error

        // migrate vendor externalIDs to kube-ovn resources created in versions prior to v1.15.0
        // this must run before ACL cleanup to ensure existing resources are properly tagged
        if err = c.OVNNbClient.MigrateVendorExternalIDs(); err != nil {
                klog.Errorf("failed to migrate vendor externalIDs: %v", err)
                return err
        }

        // migrate tier field of ACL rules created in versions prior to v1.13.0
        // after upgrading, the tier field has a default value of zero, which is not the value used in versions >= v1.13.0
        // we need to migrate the tier field to the correct value
        if err = c.OVNNbClient.MigrateACLTier(); err != nil {
                klog.Errorf("failed to migrate ACL tier: %v", err)
                return err
        }

        // clean all no parent key acls
        if err = c.OVNNbClient.CleanNoParentKeyAcls(); err != nil {
                klog.Errorf("failed to clean all no parent key acls: %v", err)
                return err
        }

        if err = c.InitDefaultVpc(); err != nil {
                klog.Errorf("init default vpc failed: %v", err)
                return err
        }

        if err = c.initClusterRouter(); err != nil {
                klog.Errorf("init cluster router failed: %v", err)
                return err
        }

        if c.config.EnableLb {
                if err = c.initLoadBalancer(); err != nil {
                        klog.Errorf("init load balancer failed: %v", err)
                        return err
                }
        }

        if err = c.initDefaultVlan(); err != nil {
                klog.Errorf("init default vlan failed: %v", err)
                return err
        }

        if err = c.initNodeSwitch(); err != nil {
                klog.Errorf("init node switch failed: %v", err)
                return err
        }

        if err = c.initDefaultLogicalSwitch(); err != nil {
                klog.Errorf("init default switch failed: %v", err)
                return err
        }

        return nil
}

func (c *Controller) InitDefaultVpc() error {
        cachedVpc, err := c.vpcsLister.Get(c.config.ClusterRouter)
        if err != nil {
                if !k8serrors.IsNotFound(err) {
                        klog.Errorf("failed to get default vpc %q: %v", c.config.ClusterRouter, err)
                        return err
                }
                // create default vpc
                vpc := &kubeovnv1.Vpc{
                        ObjectMeta: metav1.ObjectMeta{Name: c.config.ClusterRouter},
                }
                cachedVpc, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Create(context.Background(), vpc, metav1.CreateOptions{})
                if err != nil {
                        klog.Errorf("failed to create default vpc %q: %v", c.config.ClusterRouter, err)
                        return err
                }
        }

        // update default vpc status
        vpc := cachedVpc.DeepCopy()
        if !vpc.Status.Default || !vpc.Status.Standby ||
                vpc.Status.Router != c.config.ClusterRouter ||
                vpc.Status.DefaultLogicalSwitch != c.config.DefaultLogicalSwitch {
                vpc.Status.Standby = true
                vpc.Status.Default = true
                vpc.Status.Router = c.config.ClusterRouter
                vpc.Status.DefaultLogicalSwitch = c.config.DefaultLogicalSwitch

                if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().UpdateStatus(context.Background(), vpc, metav1.UpdateOptions{}); err != nil {
                        klog.Errorf("failed to update default vpc %q: %v", c.config.ClusterRouter, err)
                        return err
                }
        }

        return nil
}

// InitDefaultLogicalSwitch init the default logical switch for ovn network
func (c *Controller) initDefaultLogicalSwitch() error {
        subnet, err := c.subnetsLister.Get(c.config.DefaultLogicalSwitch)
        if err == nil {
                if subnet != nil && util.CheckProtocol(c.config.DefaultCIDR) != util.CheckProtocol(subnet.Spec.CIDRBlock) {
                        // single-stack upgrade to dual-stack
                        if util.CheckProtocol(c.config.DefaultCIDR) == kubeovnv1.ProtocolDual {
                                subnet := subnet.DeepCopy()
                                subnet.Spec.CIDRBlock = c.config.DefaultCIDR
                                if _, err = c.formatSubnet(subnet); err != nil {
                                        klog.Errorf("init format subnet %s failed: %v", c.config.DefaultLogicalSwitch, err)
                                        return err
                                }
                        }
                }
                return nil
        }

        if !k8serrors.IsNotFound(err) {
                klog.Errorf("get default subnet %s failed: %v", c.config.DefaultLogicalSwitch, err)
                return err
        }

        defaultSubnet := kubeovnv1.Subnet{
                ObjectMeta: metav1.ObjectMeta{Name: c.config.DefaultLogicalSwitch},
                Spec: kubeovnv1.SubnetSpec{
                        Vpc:                 c.config.ClusterRouter,
                        Default:             true,
                        Provider:            util.OvnProvider,
                        CIDRBlock:           c.config.DefaultCIDR,
                        Gateway:             c.config.DefaultGateway,
                        GatewayNode:         "",
                        DisableGatewayCheck: !c.config.DefaultGatewayCheck,
                        ExcludeIps:          strings.Split(c.config.DefaultExcludeIps, ","),
                        NatOutgoing:         true,
                        GatewayType:         kubeovnv1.GWDistributedType,
                        Protocol:            util.CheckProtocol(c.config.DefaultCIDR),
                        EnableLb:            &c.config.EnableLb,
                },
        }
        if c.config.NetworkType == util.NetworkTypeVlan {
                defaultSubnet.Spec.Vlan = c.config.DefaultVlanName
                if c.config.DefaultLogicalGateway && c.config.DefaultU2OInterconnection {
                        err = errors.New("logicalGateway and u2oInterconnection can't be opened at the same time")
                        klog.Error(err)
                        return err
                }
                defaultSubnet.Spec.LogicalGateway = c.config.DefaultLogicalGateway
                defaultSubnet.Spec.U2OInterconnection = c.config.DefaultU2OInterconnection
        }

        if _, err = c.config.KubeOvnClient.KubeovnV1().Subnets().Create(context.Background(), &defaultSubnet, metav1.CreateOptions{}); err != nil {
                klog.Errorf("failed to create default subnet %q: %v", c.config.DefaultLogicalSwitch, err)
                return err
        }
        return nil
}

// InitNodeSwitch init node switch to connect host and pod
func (c *Controller) initNodeSwitch() error {
        subnet, err := c.subnetsLister.Get(c.config.NodeSwitch)
        if err == nil {
                if util.CheckProtocol(c.config.NodeSwitchCIDR) == kubeovnv1.ProtocolDual && util.CheckProtocol(subnet.Spec.CIDRBlock) != kubeovnv1.ProtocolDual {
                        // single-stack upgrade to dual-stack
                        subnet := subnet.DeepCopy()
                        subnet.Spec.CIDRBlock = c.config.NodeSwitchCIDR
                        if _, err = c.formatSubnet(subnet); err != nil {
                                klog.Errorf("init format subnet %s failed: %v", c.config.NodeSwitch, err)
                                return err
                        }
                } else {
                        c.config.NodeSwitchCIDR = subnet.Spec.CIDRBlock
                }
                return nil
        }

        if !k8serrors.IsNotFound(err) {
                klog.Errorf("get node subnet %s failed: %v", c.config.NodeSwitch, err)
                return err
        }

        nodeSubnet := kubeovnv1.Subnet{
                ObjectMeta: metav1.ObjectMeta{Name: c.config.NodeSwitch},
                Spec: kubeovnv1.SubnetSpec{
                        Vpc:                    c.config.ClusterRouter,
                        Default:                false,
                        Provider:               util.OvnProvider,
                        CIDRBlock:              c.config.NodeSwitchCIDR,
                        Gateway:                c.config.NodeSwitchGateway,
                        GatewayNode:            "",
                        ExcludeIps:             strings.Split(c.config.NodeSwitchGateway, ","),
                        Protocol:               util.CheckProtocol(c.config.NodeSwitchCIDR),
                        DisableInterConnection: true,
                },
        }

        if _, err = c.config.KubeOvnClient.KubeovnV1().Subnets().Create(context.Background(), &nodeSubnet, metav1.CreateOptions{}); err != nil {
                klog.Errorf("failed to create node subnet %q: %v", c.config.NodeSwitch, err)
                return err
        }
        return nil
}

// InitClusterRouter init cluster router to connect different logical switches
func (c *Controller) initClusterRouter() error {
        if err := c.OVNNbClient.CreateLogicalRouter(c.config.ClusterRouter); err != nil {
                klog.Errorf("create logical router %s failed: %v", c.config.ClusterRouter, err)
                return err
        }

        lr, err := c.OVNNbClient.GetLogicalRouter(c.config.ClusterRouter, false)
        if err != nil {
                klog.Errorf("get logical router %s failed: %v", c.config.ClusterRouter, err)
                return err
        }

        lrOptions := make(map[string]string, len(lr.Options))
        maps.Copy(lrOptions, lr.Options)
        lrOptions["mac_binding_age_threshold"] = "300"
        lrOptions["dynamic_neigh_routers"] = "true"
        if !maps.Equal(lr.Options, lrOptions) {
                lr.Options = lrOptions
                if err = c.OVNNbClient.UpdateLogicalRouter(lr, &lr.Options); err != nil {
                        klog.Errorf("update logical router %s failed: %v", c.config.ClusterRouter, err)
                        return err
                }
        }

        return nil
}

func (c *Controller) initLB(name, protocol string, sessionAffinity bool) error {
        protocol = strings.ToLower(protocol)

        var (
                selectFields []string
                err          error
        )

        if sessionAffinity {
                selectFields = []string{
                        ovnnb.LoadBalancerSelectionFieldsIPSrc,
                        ovnnb.LoadBalancerSelectionFieldsIpv6Src,
                }
        }

        if err = c.OVNNbClient.CreateLoadBalancer(name, protocol, selectFields...); err != nil {
                klog.Errorf("create load balancer %s: %v", name, err)
                return err
        }

        if sessionAffinity {
                if err = c.OVNNbClient.SetLoadBalancerAffinityTimeout(name, util.DefaultServiceSessionStickinessTimeout); err != nil {
                        klog.Errorf("failed to set affinity timeout of %s load balancer %s: %v", protocol, name, err)
                        return err
                }
        }

        err = c.OVNNbClient.SetLoadBalancerPreferLocalBackend(name, c.config.EnableOVNLBPreferLocal)
        if err != nil {
                klog.Errorf("failed to set prefer local backend for load balancer %s: %v", name, err)
                return err
        }

        // ct_flush wipes all conntrack entries on the LB's datapath whenever a vip
        // is mutated. Session-affinity LBs are shared across services, and their
        // per-client affinity binding is carried in conntrack; enabling ct_flush on
        // those LBs lets an unrelated service's backend change invalidate another
        // service's active affinity. Only enable ct_flush on non-session UDP LBs.
        if protocol == "udp" && !sessionAffinity {
                if err = c.OVNNbClient.SetLoadBalancerCtFlush(name, true); err != nil {
                        klog.Errorf("failed to set ct_flush for load balancer %s: %v", name, err)
                        return err
                }
        }

        return nil
}

// InitLoadBalancer creates the default TCP/UDP/SCTP cluster load balancers in
// OVN for every existing VPC and records their names in each VPC's status so
// the subnet worker can attach them to its logical switch on its first
// reconcile.
//
// The status write uses a targeted merge patch that contains only the six
// LB-name fields. An earlier version serialized the whole VpcStatus via
// vpc.Status.Bytes() and raced InitDefaultVpc: if the VPC lister cache still
// held the pre-UpdateStatus copy (Standby=false) the whole-status merge patch
// would silently overwrite the Standby/Default/Router/DefaultLogicalSwitch
// fields that InitDefaultVpc had just written, deadlocking the subnet worker.
// A field-scoped patch avoids that class of overwrite entirely.
func (c *Controller) initLoadBalancer() error {
        vpcs, err := c.vpcsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list vpc: %v", err)
                return err
        }

        for _, cachedVpc := range vpcs {
                vpcLb := c.GenVpcLoadBalancer(cachedVpc.Name)
                if err = c.initLB(vpcLb.TCPLoadBalancer, string(v1.ProtocolTCP), false); err != nil {
                        klog.Error(err)
                        return err
                }
                if err = c.initLB(vpcLb.TCPSessLoadBalancer, string(v1.ProtocolTCP), true); err != nil {
                        klog.Error(err)
                        return err
                }
                if err = c.initLB(vpcLb.UDPLoadBalancer, string(v1.ProtocolUDP), false); err != nil {
                        klog.Error(err)
                        return err
                }
                if err = c.initLB(vpcLb.UDPSessLoadBalancer, string(v1.ProtocolUDP), true); err != nil {
                        klog.Error(err)
                        return err
                }
                if err = c.initLB(vpcLb.SctpLoadBalancer, string(v1.ProtocolSCTP), false); err != nil {
                        klog.Error(err)
                        return err
                }
                if err = c.initLB(vpcLb.SctpSessLoadBalancer, string(v1.ProtocolSCTP), true); err != nil {
                        klog.Error(err)
                        return err
                }

                body, err := buildVpcLBStatusPatch(vpcLb)
                if err != nil {
                        klog.Error(err)
                        return err
                }
                if _, err = c.config.KubeOvnClient.KubeovnV1().Vpcs().Patch(context.Background(), cachedVpc.Name, types.MergePatchType, body, metav1.PatchOptions{}, "status"); err != nil {
                        klog.Error(err)
                        return err
                }
        }
        return nil
}

// buildVpcLBStatusPatch builds a merge-patch body that updates only the six
// LB-name fields of VpcStatus. It deliberately excludes every other field so
// the merge patch cannot overwrite state owned by InitDefaultVpc (Standby,
// Default, Router, DefaultLogicalSwitch) when the caller reads from a stale
// lister cache.
func buildVpcLBStatusPatch(vpcLb *VpcLoadBalancer) ([]byte, error) {
        patch := struct {
                Status struct {
                        TCPLoadBalancer         string `json:"tcpLoadBalancer"`
                        TCPSessionLoadBalancer  string `json:"tcpSessionLoadBalancer"`
                        UDPLoadBalancer         string `json:"udpLoadBalancer"`
                        UDPSessionLoadBalancer  string `json:"udpSessionLoadBalancer"`
                        SctpLoadBalancer        string `json:"sctpLoadBalancer"`
                        SctpSessionLoadBalancer string `json:"sctpSessionLoadBalancer"`
                } `json:"status"`
        }{}
        patch.Status.TCPLoadBalancer = vpcLb.TCPLoadBalancer
        patch.Status.TCPSessionLoadBalancer = vpcLb.TCPSessLoadBalancer
        patch.Status.UDPLoadBalancer = vpcLb.UDPLoadBalancer
        patch.Status.UDPSessionLoadBalancer = vpcLb.UDPSessLoadBalancer
        patch.Status.SctpLoadBalancer = vpcLb.SctpLoadBalancer
        patch.Status.SctpSessionLoadBalancer = vpcLb.SctpSessLoadBalancer
        return json.Marshal(patch)
}

func (c *Controller) InitIPAM() error {
        start := time.Now()
        subnets, err := c.subnetsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list subnet: %v", err)
                return err
        }
        subnetProviderMaps := make(map[string]string, len(subnets))
        for _, subnet := range subnets {
                klog.Infof("Init subnet %s", subnet.Name)
                subnetProviderMaps[subnet.Name] = subnet.Spec.Provider
                if err := c.ipam.AddOrUpdateSubnet(subnet.Name, subnet.Spec.CIDRBlock, subnet.Spec.Gateway, subnet.Spec.ExcludeIps); err != nil {
                        klog.Errorf("failed to init subnet %s: %v", subnet.Name, err)
                }

                u2oInterconnName := fmt.Sprintf(util.U2OInterconnName, subnet.Spec.Vpc, subnet.Name)
                u2oInterconnLrpName := fmt.Sprintf("%s-%s", subnet.Spec.Vpc, subnet.Name)
                if subnet.Status.U2OInterconnectionIP != "" {
                        var mac *string
                        klog.Infof("Init U2O for subnet %s", subnet.Name)
                        if subnet.Status.U2OInterconnectionMAC != "" {
                                mac = new(subnet.Status.U2OInterconnectionMAC)
                        } else {
                                lrp, err := c.OVNNbClient.GetLogicalRouterPort(u2oInterconnLrpName, true)
                                if err != nil {
                                        klog.Errorf("failed to get logical router port %s: %v", u2oInterconnLrpName, err)
                                        return err
                                }
                                if lrp != nil {
                                        mac = new(lrp.MAC)
                                }
                        }
                        if _, _, _, err = c.ipam.GetStaticAddress(u2oInterconnName, u2oInterconnLrpName, subnet.Status.U2OInterconnectionIP, mac, subnet.Name, true); err != nil {
                                klog.Errorf("failed to init subnet %q u2o interconnection ip to ipam %v", subnet.Name, err)
                        }
                }
        }

        ippools, err := c.ippoolLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list ippool: %v", err)
                return err
        }
        for _, ippool := range ippools {
                if err = c.ipam.AddOrUpdateIPPool(ippool.Spec.Subnet, ippool.Name, ippool.Spec.IPs); err != nil {
                        klog.Errorf("failed to init ippool %s: %v", ippool.Name, err)
                }
        }

        klog.Infof("Init IPAM from StatefulSet or VM IP CR")
        ips, err := c.ipsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list IPs: %v", err)
                return err
        }

        for _, ip := range ips {
                if !ip.DeletionTimestamp.IsZero() {
                        klog.Infof("enqueue update for removing finalizer to delete ip %s", ip.Name)
                        c.updateIPQueue.Add(ip.Name)
                        continue
                }
                // recover sts and kubevirt vm ip, other ip recover in later pod loop
                if ip.Spec.PodType != util.KindStatefulSet &&
                        ip.Spec.PodType != util.KindVirtualMachine {
                        continue
                }

                var ipamKey string
                if ip.Spec.Namespace != "" {
                        ipamKey = fmt.Sprintf("%s/%s", ip.Spec.Namespace, ip.Spec.PodName)
                } else {
                        ipamKey = util.NodeLspName(ip.Spec.PodName)
                }
                if _, _, _, err = c.ipam.GetStaticAddress(ipamKey, ip.Name, ip.Spec.IPAddress, &ip.Spec.MacAddress, ip.Spec.Subnet, true); err != nil {
                        klog.Errorf("failed to init IPAM from IP CR %s: %v", ip.Name, err)
                }
        }

        klog.Infof("Init IPAM from pod")
        pods, err := c.podsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list pods: %v", err)
                return err
        }
        for _, pod := range pods {
                if pod.Spec.HostNetwork {
                        continue
                }

                isAlive := isPodAlive(pod)
                isStsPod, _, _ := isStatefulSetPod(pod)
                if !isAlive && !isStsPod {
                        continue
                }

                if !hasAllocatedAnnotation(pod) {
                        continue
                }

                podNets, err := c.getPodKubeovnNets(pod)
                if err != nil {
                        klog.Errorf("failed to get pod kubeovn nets %s.%s address %s: %v", pod.Name, pod.Namespace, pod.Annotations[util.IPAddressAnnotation], err)
                        continue
                }

                podType := getPodType(pod)
                podName := c.getNameByPod(pod)
                key := cache.NewObjectName(pod.Namespace, podName).String()
                for _, podNet := range podNets {
                        if pod.Annotations[fmt.Sprintf(util.AllocatedAnnotationTemplate, podNet.ProviderName)] == "true" {
                                portName := ovs.PodNameToPortName(podName, pod.Namespace, podNet.ProviderName)
                                ip := pod.Annotations[fmt.Sprintf(util.IPAddressAnnotationTemplate, podNet.ProviderName)]
                                mac := pod.Annotations[fmt.Sprintf(util.MacAddressAnnotationTemplate, podNet.ProviderName)]
                                if ip == "" {
                                        klog.Warningf("pod %s/%s has empty IP annotation for provider %s, skip IPAM init", pod.Namespace, podName, podNet.ProviderName)
                                        continue
                                }
                                _, _, _, err := c.ipam.GetStaticAddress(key, portName, ip, &mac, podNet.Subnet.Name, true)
                                if err != nil {
                                        klog.Errorf("failed to init pod %s.%s address %s: %v", podName, pod.Namespace, ip, err)
                                } else {
                                        err = c.createOrUpdateIPCR(portName, podName, ip, mac, podNet.Subnet.Name, pod.Namespace, pod.Spec.NodeName, podType)
                                        if err != nil {
                                                klog.Errorf("failed to create/update ips CR %s.%s with ip address %s: %v", podName, pod.Namespace, ip, err)
                                        }
                                }

                                // Append ExternalIds is added in v1.7, used for upgrading from v1.6.3. It should be deleted now since v1.7 is not used anymore.
                        }
                }
        }

        klog.Infof("Init IPAM from vip CR")
        vips, err := c.virtualIpsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list vips: %v", err)
                return err
        }
        for _, vip := range vips {
                provider, ok := subnetProviderMaps[vip.Spec.Subnet]
                if !ok {
                        klog.Errorf("failed to find subnet %s for vip %s", vip.Spec.Subnet, vip.Name)
                        continue
                }
                portName := ovs.PodNameToPortName(vip.Name, vip.Spec.Namespace, provider)
                if _, _, _, err = c.ipam.GetStaticAddress(vip.Name, portName, vip.Status.V4ip, &vip.Status.Mac, vip.Spec.Subnet, true); err != nil {
                        klog.Errorf("failed to init ipam from vip cr %s: %v", vip.Name, err)
                }
        }

        klog.Infof("Init IPAM from iptables EIP CR")
        eips, err := c.iptablesEipsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list EIPs: %v", err)
                return err
        }
        for _, eip := range eips {
                externalNetwork := util.GetExternalNetwork(eip.Spec.ExternalSubnet)
                if _, _, _, err = c.ipam.GetStaticAddress(eip.Name, eip.Name, eip.Status.IP, &eip.Spec.MacAddress, externalNetwork, true); err != nil {
                        klog.Errorf("failed to init ipam from iptables eip cr %s: %v", eip.Name, err)
                }
        }

        klog.Infof("Init IPAM from ovn EIP CR")
        oeips, err := c.ovnEipsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list ovn eips: %v", err)
                return err
        }
        for _, oeip := range oeips {
                if _, _, _, err = c.ipam.GetStaticAddress(oeip.Name, oeip.Name, oeip.Status.V4Ip, &oeip.Status.MacAddress, oeip.Spec.ExternalSubnet, true); err != nil {
                        klog.Errorf("failed to init ipam from ovn eip cr %s: %v", oeip.Name, err)
                }
        }

        klog.Infof("Init IPAM from node annotation")
        nodes, err := c.nodesLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list nodes: %v", err)
                return err
        }
        for _, node := range nodes {
                if node.Annotations[util.AllocatedAnnotation] == "true" {
                        portName := util.NodeLspName(node.Name)
                        mac := node.Annotations[util.MacAddressAnnotation]
                        v4IP, v6IP, _, err := c.ipam.GetStaticAddress(portName, portName,
                                node.Annotations[util.IPAddressAnnotation], &mac,
                                node.Annotations[util.LogicalSwitchAnnotation], true)
                        if err != nil {
                                klog.Errorf("failed to init node %s.%s address %s: %v", node.Name, node.Namespace, node.Annotations[util.IPAddressAnnotation], err)
                        }
                        if v4IP != "" && v6IP != "" {
                                ipStr := util.GetStringIP(v4IP, v6IP)
                                if ipStr != node.Annotations[util.IPAddressAnnotation] {
                                        patch := util.KVPatch{util.IPAddressAnnotation: ipStr}
                                        if err = util.PatchAnnotations(c.config.KubeClient.CoreV1().Nodes(), node.Name, patch); err != nil {
                                                klog.Errorf("failed to patch node %s IP annotation: %v", node.Name, err)
                                        }
                                }
                        }
                }
        }

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

func (c *Controller) initDefaultProviderNetwork() error {
        _, err := c.providerNetworksLister.Get(c.config.DefaultProviderName)
        if err == nil {
                return nil
        }
        if !k8serrors.IsNotFound(err) {
                klog.Errorf("failed to get default provider network %s: %v", c.config.DefaultProviderName, err)
                return err
        }

        nodes, err := c.nodesLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to get nodes: %v", err)
                return err
        }

        pn := kubeovnv1.ProviderNetwork{
                ObjectMeta: metav1.ObjectMeta{
                        Name: c.config.DefaultProviderName,
                },
                Spec: kubeovnv1.ProviderNetworkSpec{
                        DefaultInterface: c.config.DefaultHostInterface,
                        ExchangeLinkName: c.config.DefaultExchangeLinkName,
                },
        }

        excludeAnno := fmt.Sprintf(util.ProviderNetworkExcludeTemplate, c.config.DefaultProviderName)
        interfaceAnno := fmt.Sprintf(util.ProviderNetworkInterfaceTemplate, c.config.DefaultProviderName)
        patchNodes := make([]string, 0, len(nodes))
        for _, node := range nodes {
                if len(node.Annotations) == 0 {
                        continue
                }

                if node.Annotations[excludeAnno] == "true" {
                        pn.Spec.ExcludeNodes = append(pn.Spec.ExcludeNodes, node.Name)
                        patchNodes = append(patchNodes, node.Name)
                } else if s := node.Annotations[interfaceAnno]; s != "" {
                        var index *int
                        for i := range pn.Spec.CustomInterfaces {
                                if pn.Spec.CustomInterfaces[i].Interface == s {
                                        index = &i
                                        break
                                }
                        }
                        if index != nil {
                                pn.Spec.CustomInterfaces[*index].Nodes = append(pn.Spec.CustomInterfaces[*index].Nodes, node.Name)
                        } else {
                                ci := kubeovnv1.CustomInterface{Interface: s, Nodes: []string{node.Name}}
                                pn.Spec.CustomInterfaces = append(pn.Spec.CustomInterfaces, ci)
                        }
                        patchNodes = append(patchNodes, node.Name)
                }
        }

        defer func() {
                if err != nil {
                        return
                }

                // update nodes only when provider network has been created successfully
                patch := util.KVPatch{excludeAnno: nil, interfaceAnno: nil}
                for _, node := range patchNodes {
                        if err := util.PatchAnnotations(c.config.KubeClient.CoreV1().Nodes(), node, patch); err != nil {
                                klog.Errorf("failed to patch node %s: %v", node, err)
                        }
                }
        }()

        _, err = c.config.KubeOvnClient.KubeovnV1().ProviderNetworks().Create(context.Background(), &pn, metav1.CreateOptions{})
        if err != nil {
                klog.Errorf("failed to create provider network %s: %v", c.config.DefaultProviderName, err)
                return err
        }
        return nil
}

func (c *Controller) initDefaultVlan() error {
        if c.config.NetworkType != util.NetworkTypeVlan {
                return nil
        }

        if err := c.initDefaultProviderNetwork(); err != nil {
                klog.Error(err)
                return err
        }

        _, err := c.vlansLister.Get(c.config.DefaultVlanName)
        if err == nil {
                return nil
        }

        if !k8serrors.IsNotFound(err) {
                klog.Errorf("get default vlan %s failed: %v", c.config.DefaultVlanName, err)
                return err
        }

        if c.config.DefaultVlanID < 0 || c.config.DefaultVlanID > 4095 {
                return errors.New("the default vlan id is not between 1-4095")
        }

        defaultVlan := kubeovnv1.Vlan{
                ObjectMeta: metav1.ObjectMeta{Name: c.config.DefaultVlanName},
                Spec: kubeovnv1.VlanSpec{
                        ID:       c.config.DefaultVlanID,
                        Provider: c.config.DefaultProviderName,
                },
        }

        _, err = c.config.KubeOvnClient.KubeovnV1().Vlans().Create(context.Background(), &defaultVlan, metav1.CreateOptions{})
        if err != nil {
                klog.Errorf("failed to create vlan %s: %v", defaultVlan.Name, err)
                return err
        }
        return nil
}

func (c *Controller) syncIPCR() error {
        klog.Info("start to sync ips")
        ips, err := c.ipsLister.List(labels.Everything())
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }

        vmLsps, err := c.getVMLsps()
        if err != nil {
                klog.Errorf("failed to get vm lsps, %v", err)
                return err
        }
        ipMap := strset.New(vmLsps...)
        for _, ip := range ips {
                if !ip.DeletionTimestamp.IsZero() {
                        klog.Infof("enqueue update for removing finalizer to delete ip %s", ip.Name)
                        c.updateIPQueue.Add(ip.Name)
                        continue
                }
                changed := false
                ip = ip.DeepCopy()
                if ipMap.Has(ip.Name) && ip.Spec.PodType == "" {
                        ip.Spec.PodType = util.KindVirtualMachine
                        changed = true
                }

                v4IP, v6IP := util.SplitStringIP(ip.Spec.IPAddress)
                if ip.Spec.V4IPAddress == v4IP && ip.Spec.V6IPAddress == v6IP && !changed {
                        continue
                }

                ip.Spec.V4IPAddress = v4IP
                ip.Spec.V6IPAddress = v6IP
                _, err := c.config.KubeOvnClient.KubeovnV1().IPs().Update(context.Background(), ip, metav1.UpdateOptions{})
                if err != nil {
                        klog.Errorf("failed to sync crd ip %s: %v", ip.Spec.IPAddress, err)
                        return err
                }
        }
        return nil
}

func (c *Controller) syncSubnetCR() error {
        klog.Info("start to sync subnets")
        subnets, err := c.subnetsLister.List(labels.Everything())
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        for _, cachedSubnet := range subnets {
                subnet := cachedSubnet.DeepCopy()
                if !subnet.Status.IsReady() {
                        klog.Warningf("subnet %s is not ready", subnet.Name)
                        continue
                }
                subnet, err = c.calcSubnetStatusIP(subnet)
                if err != nil {
                        klog.Errorf("failed to calculate subnet %s used ip: %v", cachedSubnet.Name, err)
                        return err
                }

                // only sync subnet spec enableEcmp when subnet.Spec.EnableEcmp is false and c.config.EnableEcmp is true
                if subnet.Spec.GatewayType == kubeovnv1.GWCentralizedType && !subnet.Spec.EnableEcmp && subnet.Spec.EnableEcmp != c.config.EnableEcmp {
                        subnet, err = c.subnetsLister.Get(subnet.Name)
                        if err != nil {
                                klog.Errorf("failed to get subnet %s: %v", subnet.Name, err)
                                return err
                        }

                        subnet.Spec.EnableEcmp = c.config.EnableEcmp
                        if _, err := c.config.KubeOvnClient.KubeovnV1().Subnets().Update(context.Background(), subnet, metav1.UpdateOptions{}); err != nil {
                                klog.Errorf("failed to sync subnet spec enableEcmp with kube-ovn-controller config enableEcmp %s: %v", subnet.Name, err)
                                return err
                        }
                }
        }
        return nil
}

func (c *Controller) syncVpcNatGatewayCR() error {
        klog.Info("start to sync crd vpc nat gw")
        gws, err := c.vpcNatGatewayLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list vpc nat gateway, %v", err)
                return err
        }
        if len(gws) == 0 {
                return nil
        }
        // get vpc nat gateway enable state
        cm, err := c.configMapsLister.ConfigMaps(c.config.PodNamespace).Get(util.VpcNatGatewayConfig)
        if err != nil && !k8serrors.IsNotFound(err) {
                klog.Errorf("failed to get config map %s, %v", util.VpcNatGatewayConfig, err)
                return err
        }
        if k8serrors.IsNotFound(err) || cm.Data["enable-vpc-nat-gw"] == "false" {
                return nil
        }
        // get vpc nat gateway image
        cm, err = c.configMapsLister.ConfigMaps(c.config.PodNamespace).Get(util.VpcNatConfig)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        klog.Errorf("should set config map for vpc-nat-gateway %s, %v", util.VpcNatConfig, err)
                        return err
                }
                klog.Errorf("failed to get config map %s, %v", util.VpcNatConfig, err)
                return err
        }

        if cm.Data["image"] == "" {
                err = errors.New("should set image for vpc-nat-gateway pod")
                klog.Error(err)
                return err
        }

        for _, gw := range gws {
                if err := c.updateCrdNatGwLabels(gw.Name, ""); err != nil {
                        klog.Errorf("failed to update nat gw %s: %v", gw.Name, err)
                        return err
                }
        }
        return nil
}

func (c *Controller) syncVlanCR() error {
        klog.Info("start to sync vlans")
        vlans, err := c.vlansLister.List(labels.Everything())
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }

        for _, vlan := range vlans {
                var needUpdate bool
                newVlan := vlan.DeepCopy()
                if newVlan.Spec.VlanID != 0 && newVlan.Spec.ID == 0 {
                        newVlan.Spec.ID = newVlan.Spec.VlanID
                        newVlan.Spec.VlanID = 0
                        needUpdate = true
                }
                //nolint:staticcheck // Ignore SA1019 for backward compatibility of deprecated field ProviderInterfaceName
                if newVlan.Spec.ProviderInterfaceName != "" && newVlan.Spec.Provider == "" {
                        //nolint:staticcheck // Ignore SA1019
                        newVlan.Spec.Provider = newVlan.Spec.ProviderInterfaceName
                        //nolint:staticcheck // Ignore SA1019
                        newVlan.Spec.ProviderInterfaceName = ""
                        needUpdate = true
                }
                if needUpdate {
                        if _, err = c.config.KubeOvnClient.KubeovnV1().Vlans().Update(context.Background(), newVlan, metav1.UpdateOptions{}); err != nil {
                                klog.Errorf("failed to update spec of vlan %s: %v", newVlan.Name, err)
                                return err
                        }
                }
        }

        return nil
}

func (c *Controller) batchMigrateNodeRoute(nodes []*v1.Node) error {
        start := time.Now()
        addPolicies := make([]*kubeovnv1.PolicyRoute, 0)
        delPolicies := make([]*kubeovnv1.PolicyRoute, 0)
        staticRoutes := make([]*kubeovnv1.StaticRoute, 0)
        externalIDsMap := make(map[string]map[string]string)
        delAsNames := make([]string, 0)
        for _, node := range nodes {
                if node.Annotations[util.AllocatedAnnotation] != "true" {
                        continue
                }
                nodeName := node.Name
                nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
                joinAddrV4, joinAddrV6 := util.SplitStringIP(node.Annotations[util.IPAddressAnnotation])
                if nodeIPv4 != "" && joinAddrV4 != "" {
                        buildNodeRoute(4, nodeName, joinAddrV4, nodeIPv4, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)
                }
                if nodeIPv6 != "" && joinAddrV6 != "" {
                        buildNodeRoute(6, nodeName, joinAddrV6, nodeIPv6, &addPolicies, &delPolicies, &staticRoutes, externalIDsMap, &delAsNames)
                }
        }

        if err := c.batchAddPolicyRouteToVpc(c.config.ClusterRouter, addPolicies, externalIDsMap); err != nil {
                klog.Errorf("failed to batch add logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
                return err
        }
        if err := c.batchDeleteStaticRouteFromVpc(c.config.ClusterRouter, staticRoutes); err != nil {
                klog.Errorf("failed to batch delete  obsolete logical router static route for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
                return err
        }
        if err := c.batchDeletePolicyRouteFromVpc(c.config.ClusterRouter, delPolicies); err != nil {
                klog.Errorf("failed to batch delete obsolete logical router policy for lr %s nodes %d: %v", c.config.ClusterRouter, len(nodes), err)
                return err
        }
        if err := c.OVNNbClient.BatchDeleteAddressSetByNames(delAsNames); err != nil {
                klog.Errorf("failed to batch delete obsolete address set for asNames %v nodes %d: %v", delAsNames, len(nodes), err)
                return err
        }
        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",
                time.Since(start), c.config.ClusterRouter, util.NodeRouterPolicyPriority, len(addPolicies), len(externalIDsMap), len(delPolicies), len(delAsNames))

        return nil
}

func buildNodeRoute(af int, nodeName, nexthop, ip string, addPolicies, delPolicies *[]*kubeovnv1.PolicyRoute, staticRoutes *[]*kubeovnv1.StaticRoute, externalIDsMap map[string]map[string]string, delAsNames *[]string) {
        var (
                match       = fmt.Sprintf("ip%d.dst == %s", af, ip)
                action      = kubeovnv1.PolicyRouteActionReroute
                externalIDs = map[string]string{
                        "vendor": util.CniTypeName,
                        "node":   nodeName,
                }
        )
        *addPolicies = append(*addPolicies, &kubeovnv1.PolicyRoute{
                Priority:  util.NodeRouterPolicyPriority,
                Match:     match,
                Action:    action,
                NextHopIP: nexthop,
        })
        externalIDsMap[buildExternalIDsMapKey(match, string(action), util.NodeRouterPolicyPriority)] = externalIDs
        *staticRoutes = append(*staticRoutes, &kubeovnv1.StaticRoute{
                Policy:     kubeovnv1.PolicyDst,
                RouteTable: util.MainRouteTable,
                NextHopIP:  "",
                CIDR:       ip,
        })
        asName := nodeUnderlayAddressSetName(nodeName, af)
        obsoleteMatch := fmt.Sprintf("ip%d.dst == %s && ip%d.src != $%s", af, ip, af, asName)
        *delPolicies = append(*delPolicies, &kubeovnv1.PolicyRoute{
                Match:    obsoleteMatch,
                Priority: util.NodeRouterPolicyPriority,
        })
        *delAsNames = append(*delAsNames, asName)
}

func (c *Controller) syncNodeRoutes() error {
        nodes, err := c.nodesLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list nodes: %v", err)
                return err
        }

        if err := c.batchMigrateNodeRoute(nodes); err != nil {
                klog.Errorf("failed to batch migrate node routes: %v", err)
                return err
        }

        return nil
}

func (c *Controller) initNodeChassis() error {
        nodes, err := c.nodesLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list nodes: %v", err)
                return err
        }
        chassises, err := c.OVNSbClient.GetKubeOvnChassises()
        if err != nil {
                klog.Errorf("failed to get chassis nodes: %v", err)
                return err
        }
        chassisNodes := make(map[string]string, len(*chassises))
        for _, chassis := range *chassises {
                chassisNodes[chassis.Name] = chassis.Hostname
        }
        for _, node := range nodes {
                if err := c.UpdateChassisTag(node); err != nil {
                        klog.Error(err)
                        if _, ok := err.(*ErrChassisNotFound); !ok {
                                return err
                        }
                }
        }
        return nil
}

func migrateFinalizers(c client.Client, list client.ObjectList, getObjectItem func(int) (client.Object, client.Object)) error {
        if err := c.List(context.Background(), list); err != nil {
                klog.Errorf("failed to list objects: %v", err)
                return err
        }

        var i int
        var cachedObj, patchedObj client.Object
        for {
                if cachedObj, patchedObj = getObjectItem(i); cachedObj == nil {
                        break
                }
                if !controllerutil.ContainsFinalizer(cachedObj, util.DeprecatedFinalizerName) {
                        i++
                        continue
                }
                controllerutil.RemoveFinalizer(patchedObj, util.DeprecatedFinalizerName)
                if cachedObj.GetDeletionTimestamp() == nil {
                        // if the object is not being deleted, add the new finalizer
                        controllerutil.AddFinalizer(patchedObj, util.KubeOVNControllerFinalizer)
                }
                if err := c.Patch(context.Background(), patchedObj, client.MergeFrom(cachedObj)); client.IgnoreNotFound(err) != nil {
                        klog.Errorf("failed to sync finalizers for %s %s: %v",
                                patchedObj.GetObjectKind().GroupVersionKind().Kind,
                                cache.MetaObjectToName(patchedObj), err)
                        return err
                }
                i++
        }

        return nil
}

func (c *Controller) syncFinalizers() error {
        cl, err := client.New(config.GetConfigOrDie(), client.Options{})
        if err != nil {
                klog.Errorf("failed to create client: %v", err)
                return err
        }

        // migrate deprecated finalizer to new finalizer
        klog.Info("start to sync finalizers")
        if err := c.syncIPFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ip finalizer: %v", err)
                return err
        }
        if err := c.syncIPPoolFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ippool finalizer: %v", err)
                return err
        }
        if err := c.syncOvnDnatFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ovn dnat finalizer: %v", err)
                return err
        }
        if err := c.syncOvnEipFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ovn eip finalizer: %v", err)
                return err
        }
        if err := c.syncOvnFipFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ovn fip finalizer: %v", err)
                return err
        }
        if err := c.syncOvnSnatFinalizer(cl); err != nil {
                klog.Errorf("failed to sync ovn snat finalizer: %v", err)
                return err
        }
        if err := c.syncQoSPolicyFinalizer(cl); err != nil {
                klog.Errorf("failed to sync qos policy finalizer: %v", err)
                return err
        }
        if err := c.syncSubnetFinalizer(cl); err != nil {
                klog.Errorf("failed to sync subnet finalizer: %v", err)
                return err
        }
        if err := c.syncVipFinalizer(cl); err != nil {
                klog.Errorf("failed to sync vip finalizer: %v", err)
                return err
        }
        if err := c.syncIptablesEipFinalizer(cl); err != nil {
                klog.Errorf("failed to sync iptables eip finalizer: %v", err)
                return err
        }
        if err := c.syncIptablesFipFinalizer(cl); err != nil {
                klog.Errorf("failed to sync iptables fip finalizer: %v", err)
                return err
        }
        if err := c.syncIptablesDnatFinalizer(cl); err != nil {
                klog.Errorf("failed to sync iptables dnat finalizer: %v", err)
                return err
        }
        if err := c.syncIptablesSnatFinalizer(cl); err != nil {
                klog.Errorf("failed to sync iptables snat finalizer: %v", err)
                return err
        }
        klog.Info("sync finalizers done")
        return nil
}

func hasAllocatedAnnotation(pod *v1.Pod) bool {
        for key, value := range pod.Annotations {
                if value == "true" && strings.HasSuffix(key, util.AllocatedAnnotationSuffix) {
                        return true
                }
        }
        return false
}

kubeovn / kube-ovn / 27017056817

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