10013062981

Committed 19 Jul 2024 07:04PM UTC coverage: 52.665% (-0.2%) from 52.818%

Build # 10013062981

Build Type

Pull #4551

github

Committed by

maiqueb

Commit Message

udn, config: increase masquerade subnet netmask

We need to increase the masquerade subnet netmask size to accomodate for
more masquerade IPs, since each UDN requires a unique masquerade IP.

Thus, we need the masquerade subnet netmask to be large enough to
accomodate the desired number of networks.

Signed-off-by: Miguel Duarte Barroso <mdbarroso@redhat.com>

Pull Request Pull Request #4551: UDN egress integration: create logical infra for primary UDNs to egress for layer3 networks

Run Details

231 of 519 new or added lines in 12 files covered. (44.51%)

948 existing lines in 12 files now uncovered.

29707 of 56408 relevant lines covered (52.66%)

97.36 hits per line

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

53.76

/go-controller/pkg/ovn/secondary_layer2_network_controller.go

package ovn

import (
        "context"
        "fmt"
        "net"
        "reflect"
        "sync"
        "time"

        mnpapi "github.com/k8snetworkplumbingwg/multi-networkpolicy/pkg/apis/k8s.cni.cncf.io/v1beta1"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/allocator/pod"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/config"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/metrics"
        addressset "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/address_set"
        lsm "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/logical_switch_manager"
        zoneinterconnect "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/zone_interconnect"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/persistentips"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/retry"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/syncmap"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
        "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
        utilerrors "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util/errors"

        corev1 "k8s.io/api/core/v1"
        kapi "k8s.io/api/core/v1"
        "k8s.io/klog/v2"
)

// method/structure shared by all layer 2 network controller, including localnet and layer2 network controllres.

type secondaryLayer2NetworkControllerEventHandler struct {
        baseHandler  baseNetworkControllerEventHandler
        watchFactory *factory.WatchFactory
        objType      reflect.Type
        oc           *SecondaryLayer2NetworkController
        syncFunc     func([]interface{}) error
}

// AreResourcesEqual returns true if, given two objects of a known resource type, the update logic for this resource
// type considers them equal and therefore no update is needed. It returns false when the two objects are not considered
// equal and an update needs be executed. This is regardless of how the update is carried out (whether with a dedicated update
// function or with a delete on the old obj followed by an add on the new obj).
func (h *secondaryLayer2NetworkControllerEventHandler) AreResourcesEqual(obj1, obj2 interface{}) (bool, error) {
        return h.baseHandler.areResourcesEqual(h.objType, obj1, obj2)
}

// GetInternalCacheEntry returns the internal cache entry for this object, given an object and its type.
// This is now used only for pods, which will get their the logical port cache entry.
func (h *secondaryLayer2NetworkControllerEventHandler) GetInternalCacheEntry(obj interface{}) interface{} {
        return h.oc.GetInternalCacheEntryForSecondaryNetwork(h.objType, obj)
}

// GetResourceFromInformerCache returns the latest state of the object, given an object key and its type.
// from the informers cache.
func (h *secondaryLayer2NetworkControllerEventHandler) GetResourceFromInformerCache(key string) (interface{}, error) {
        return h.baseHandler.getResourceFromInformerCache(h.objType, h.watchFactory, key)
}

// RecordAddEvent records the add event on this given object.
func (h *secondaryLayer2NetworkControllerEventHandler) RecordAddEvent(obj interface{}) {
        switch h.objType {
        case factory.MultiNetworkPolicyType:
                mnp := obj.(*mnpapi.MultiNetworkPolicy)
                klog.V(5).Infof("Recording add event on multinetwork policy %s/%s", mnp.Namespace, mnp.Name)
                metrics.GetConfigDurationRecorder().Start("multinetworkpolicy", mnp.Namespace, mnp.Name)
        }
}

// RecordUpdateEvent records the udpate event on this given object.
func (h *secondaryLayer2NetworkControllerEventHandler) RecordUpdateEvent(obj interface{}) {
        h.baseHandler.recordAddEvent(h.objType, obj)
}

// RecordDeleteEvent records the delete event on this given object.
func (h *secondaryLayer2NetworkControllerEventHandler) RecordDeleteEvent(obj interface{}) {
        h.baseHandler.recordAddEvent(h.objType, obj)
}

// RecordSuccessEvent records the success event on this given object.
func (h *secondaryLayer2NetworkControllerEventHandler) RecordSuccessEvent(obj interface{}) {
        h.baseHandler.recordAddEvent(h.objType, obj)
}

// RecordErrorEvent records the error event on this given object.
func (h *secondaryLayer2NetworkControllerEventHandler) RecordErrorEvent(obj interface{}, reason string, err error) {
}

// IsResourceScheduled returns true if the given object has been scheduled.
// Only applied to pods for now. Returns true for all other types.
func (h *secondaryLayer2NetworkControllerEventHandler) IsResourceScheduled(obj interface{}) bool {
        return h.baseHandler.isResourceScheduled(h.objType, obj)
}

// AddResource adds the specified object to the cluster according to its type and returns the error,
// if any, yielded during object creation.
// Given an object to add and a boolean specifying if the function was executed from iterateRetryResources
func (h *secondaryLayer2NetworkControllerEventHandler) AddResource(obj interface{}, fromRetryLoop bool) error {
        switch h.objType {
        case factory.NodeType:
                node, ok := obj.(*corev1.Node)
                if !ok {
                        return fmt.Errorf("could not cast %T object to Node", obj)
                }
                if h.oc.isLocalZoneNode(node) {
                        var nodeParams *nodeSyncs
                        if fromRetryLoop {
                                _, syncMgmtPort := h.oc.mgmtPortFailed.Load(node.Name)
                                nodeParams = &nodeSyncs{syncMgmtPort: syncMgmtPort}
                        } else {
                                nodeParams = &nodeSyncs{syncMgmtPort: true}
                        }
                        return h.oc.addUpdateLocalNodeEvent(node, nodeParams)
                }
                return h.oc.addUpdateRemoteNodeEvent(node)
        default:
                return h.oc.AddSecondaryNetworkResourceCommon(h.objType, obj)
        }
}

// DeleteResource deletes the object from the cluster according to the delete logic of its resource type.
// Given an object and optionally a cachedObj; cachedObj is the internal cache entry for this object,
// used for now for pods and network policies.
func (h *secondaryLayer2NetworkControllerEventHandler) DeleteResource(obj, cachedObj interface{}) error {
        switch h.objType {
        case factory.NodeType:
                node, ok := obj.(*corev1.Node)
                if !ok {
                        return fmt.Errorf("could not cast %T object to Node", obj)
                }
                return h.oc.deleteNodeEvent(node)
        default:
                return h.oc.DeleteSecondaryNetworkResourceCommon(h.objType, obj, cachedObj)
        }
}

// UpdateResource updates the specified object in the cluster to its version in newObj according to its
// type and returns the error, if any, yielded during the object update.
// Given an old and a new object; The inRetryCache boolean argument is to indicate if the given resource
// is in the retryCache or not.
func (h *secondaryLayer2NetworkControllerEventHandler) UpdateResource(oldObj, newObj interface{}, inRetryCache bool) error {
        switch h.objType {
        case factory.NodeType:
                newNode, ok := newObj.(*corev1.Node)
                if !ok {
                        return fmt.Errorf("could not cast %T object to Node", newObj)
                }
                oldNode, ok := oldObj.(*kapi.Node)
                if !ok {
                        return fmt.Errorf("could not cast oldObj of type %T to *kapi.Node", oldObj)
                }
                newNodeIsLocalZoneNode := h.oc.isLocalZoneNode(newNode)
                nodeSubnetChanged := nodeSubnetChanged(oldNode, newNode)
                if newNodeIsLocalZoneNode {
                        var nodeSyncsParam *nodeSyncs
                        if h.oc.isLocalZoneNode(oldNode) {
                                // determine what actually changed in this update
                                syncMgmtPort := macAddressChanged(oldNode, newNode) || nodeSubnetChanged
                                nodeSyncsParam = &nodeSyncs{syncMgmtPort: syncMgmtPort}
                        } else {
                                klog.Infof("Node %s moved from the remote zone %s to local zone %s.",
                                        newNode.Name, util.GetNodeZone(oldNode), util.GetNodeZone(newNode))
                                // The node is now a local zone node. Trigger a full node sync.
                                nodeSyncsParam = &nodeSyncs{syncMgmtPort: true}
                        }

                        return h.oc.addUpdateLocalNodeEvent(newNode, nodeSyncsParam)
                } else {
                        return h.oc.addUpdateRemoteNodeEvent(newNode)
                }
        default:
                return h.oc.UpdateSecondaryNetworkResourceCommon(h.objType, oldObj, newObj, inRetryCache)
        }
}

func (h *secondaryLayer2NetworkControllerEventHandler) SyncFunc(objs []interface{}) error {
        var syncFunc func([]interface{}) error

        if h.syncFunc != nil {
                // syncFunc was provided explicitly
                syncFunc = h.syncFunc
        } else {
                switch h.objType {
                case factory.NodeType:
                        syncFunc = h.oc.syncNodes

                case factory.PodType:
                        syncFunc = h.oc.syncPodsForSecondaryNetwork

                case factory.NamespaceType:
                        syncFunc = h.oc.syncNamespaces

                case factory.MultiNetworkPolicyType:
                        syncFunc = h.oc.syncMultiNetworkPolicies

                case factory.IPAMClaimsType:
                        syncFunc = h.oc.syncIPAMClaims

                default:
                        return fmt.Errorf("no sync function for object type %s", h.objType)
                }
        }
        if syncFunc == nil {
                return nil
        }
        return syncFunc(objs)
}

// IsObjectInTerminalState returns true if the given object is a in terminal state.
// This is used now for pods that are either in a PodSucceeded or in a PodFailed state.
func (h *secondaryLayer2NetworkControllerEventHandler) IsObjectInTerminalState(obj interface{}) bool {
        return h.baseHandler.isObjectInTerminalState(h.objType, obj)
}

// SecondaryLayer2NetworkController is created for logical network infrastructure and policy
// for a secondary layer2 network
type SecondaryLayer2NetworkController struct {
        BaseSecondaryLayer2NetworkController

        // Node-specific syncMaps used by node event handler
        mgmtPortFailed sync.Map
}

// NewSecondaryLayer2NetworkController create a new OVN controller for the given secondary layer2 nad
func NewSecondaryLayer2NetworkController(cnci *CommonNetworkControllerInfo, netInfo util.NetInfo) *SecondaryLayer2NetworkController {

        stopChan := make(chan struct{})

        ipv4Mode, ipv6Mode := netInfo.IPMode()
        addressSetFactory := addressset.NewOvnAddressSetFactory(cnci.nbClient, ipv4Mode, ipv6Mode)

        lsManagerFactoryFn := lsm.NewL2SwitchManager
        if netInfo.IsPrimaryNetwork() {
                lsManagerFactoryFn = lsm.NewL2SwitchManagerForUserDefinedPrimaryNetwork
        }
        oc := &SecondaryLayer2NetworkController{
                BaseSecondaryLayer2NetworkController: BaseSecondaryLayer2NetworkController{

                        BaseSecondaryNetworkController: BaseSecondaryNetworkController{
                                BaseNetworkController: BaseNetworkController{
                                        CommonNetworkControllerInfo: *cnci,
                                        controllerName:              getNetworkControllerName(netInfo.GetNetworkName()),
                                        NetInfo:                     netInfo,
                                        lsManager:                   lsManagerFactoryFn(),
                                        logicalPortCache:            newPortCache(stopChan),
                                        namespaces:                  make(map[string]*namespaceInfo),
                                        namespacesMutex:             sync.Mutex{},
                                        addressSetFactory:           addressSetFactory,
                                        networkPolicies:             syncmap.NewSyncMap[*networkPolicy](),
                                        sharedNetpolPortGroups:      syncmap.NewSyncMap[*defaultDenyPortGroups](),
                                        podSelectorAddressSets:      syncmap.NewSyncMap[*PodSelectorAddressSet](),
                                        stopChan:                    stopChan,
                                        wg:                          &sync.WaitGroup{},
                                        localZoneNodes:              &sync.Map{},
                                        cancelableCtx:               util.NewCancelableContext(),
                                },
                        },
                },
                mgmtPortFailed: sync.Map{},
        }

        if config.OVNKubernetesFeature.EnableInterconnect {
                oc.zoneICHandler = zoneinterconnect.NewZoneInterconnectHandler(oc.NetInfo, oc.nbClient, oc.sbClient, oc.watchFactory)
        }

        if oc.allocatesPodAnnotation() {
                var claimsReconciler persistentips.PersistentAllocations
                if oc.allowPersistentIPs() {
                        ipamClaimsReconciler := persistentips.NewIPAMClaimReconciler(
                                oc.kube,
                                oc.NetInfo,
                                oc.watchFactory.IPAMClaimsInformer().Lister(),
                        )
                        oc.ipamClaimsReconciler = ipamClaimsReconciler
                        claimsReconciler = ipamClaimsReconciler
                }
                oc.podAnnotationAllocator = pod.NewPodAnnotationAllocator(
                        netInfo,
                        cnci.watchFactory.PodCoreInformer().Lister(),
                        cnci.kube,
                        claimsReconciler)
        }

        // disable multicast support for secondary networks
        // TBD: changes needs to be made to support multicast in secondary networks
        oc.multicastSupport = false

        oc.initRetryFramework()
        return oc
}

// Start starts the secondary layer2 controller, handles all events and creates all needed logical entities
func (oc *SecondaryLayer2NetworkController) Start(ctx context.Context) error {
        klog.Infof("Starting controller for secondary network %s", oc.GetNetworkName())

        start := time.Now()
        defer func() {
                klog.Infof("Starting controller for secondary network %s took %v", oc.GetNetworkName(), time.Since(start))
        }()

        if err := oc.BaseNetworkController.init(); err != nil {
                return err
        }

        if err := oc.Init(); err != nil {
                return err
        }

        return oc.run(ctx)
}

func (oc *SecondaryLayer2NetworkController) run(ctx context.Context) error {
        return oc.BaseSecondaryLayer2NetworkController.run()
}

// Cleanup cleans up logical entities for the given network, called from net-attach-def routine
// could be called from a dummy Controller (only has CommonNetworkControllerInfo set)
func (oc *SecondaryLayer2NetworkController) Cleanup() error {
        return oc.BaseSecondaryLayer2NetworkController.cleanup()
}

func (oc *SecondaryLayer2NetworkController) Init() error {
        _, err := oc.initializeLogicalSwitch(oc.GetNetworkScopedSwitchName(types.OVNLayer2Switch), oc.Subnets(), oc.ExcludeSubnets())
        return err
}

func (oc *SecondaryLayer2NetworkController) Stop() {
        klog.Infof("Stoping controller for secondary network %s", oc.GetNetworkName())
        oc.BaseSecondaryLayer2NetworkController.stop()
}

func (oc *SecondaryLayer2NetworkController) initRetryFramework() {
        oc.retryNodes = oc.newRetryFramework(factory.NodeType)
        oc.retryPods = oc.newRetryFramework(factory.PodType)
        if oc.allocatesPodAnnotation() && oc.NetInfo.AllowsPersistentIPs() {
                oc.retryIPAMClaims = oc.newRetryFramework(factory.IPAMClaimsType)
        }

        // For secondary networks, we don't have to watch namespace events if
        // multi-network policy support is not enabled. We don't support
        // multi-network policy for IPAM-less secondary networks either.
        if util.IsMultiNetworkPoliciesSupportEnabled() {
                oc.retryNamespaces = oc.newRetryFramework(factory.NamespaceType)
                oc.retryNetworkPolicies = oc.newRetryFramework(factory.MultiNetworkPolicyType)
        }
}

// newRetryFramework builds and returns a retry framework for the input resource type;
func (oc *SecondaryLayer2NetworkController) newRetryFramework(
        objectType reflect.Type) *retry.RetryFramework {
        eventHandler := &secondaryLayer2NetworkControllerEventHandler{
                baseHandler:  baseNetworkControllerEventHandler{},
                objType:      objectType,
                watchFactory: oc.watchFactory,
                oc:           oc,
                syncFunc:     nil,
        }
        resourceHandler := &retry.ResourceHandler{
                HasUpdateFunc:          hasResourceAnUpdateFunc(objectType),
                NeedsUpdateDuringRetry: needsUpdateDuringRetry(objectType),
                ObjType:                objectType,
                EventHandler:           eventHandler,
        }
        return retry.NewRetryFramework(
                oc.stopChan,
                oc.wg,
                oc.watchFactory,
                resourceHandler,
        )
}

func (oc *SecondaryLayer2NetworkController) addUpdateLocalNodeEvent(node *corev1.Node, nSyncs *nodeSyncs) error {
        var errs []error

        if util.IsNetworkSegmentationSupportEnabled() && oc.IsPrimaryNetwork() {
                if nSyncs.syncMgmtPort {
                        // Layer 2 networks have a single, large subnet, that's the one
                        // associated to the controller.  Take the management port IP from
                        // there.
                        subnets := oc.Subnets()
                        hostSubnets := make([]*net.IPNet, 0, len(subnets))
                        for _, subnet := range oc.Subnets() {
                                hostSubnets = append(hostSubnets, subnet.CIDR)
                        }
                        if _, err := oc.syncNodeManagementPortNoRouteHostSubnets(node, oc.GetNetworkScopedSwitchName(types.OVNLayer2Switch), hostSubnets); err != nil {
                                errs = append(errs, err)
                                oc.mgmtPortFailed.Store(node.Name, true)
                        } else {
                                oc.mgmtPortFailed.Delete(node.Name)
                        }
                }
        }

        errs = append(errs, oc.BaseSecondaryLayer2NetworkController.addUpdateLocalNodeEvent(node))

        err := utilerrors.Join(errs...)
        if err != nil {
                oc.recordNodeErrorEvent(node, err)
        }
        return err
}

func (oc *SecondaryLayer2NetworkController) deleteNodeEvent(node *corev1.Node) error {
        oc.localZoneNodes.Delete(node.Name)
        oc.mgmtPortFailed.Delete(node.Name)
        return nil
}

1	package ovn
2
3	import (
4	"context"
5	"fmt"
6	"net"
7	"reflect"
8	"sync"
9	"time"
10
11	mnpapi "github.com/k8snetworkplumbingwg/multi-networkpolicy/pkg/apis/k8s.cni.cncf.io/v1beta1"
12	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/allocator/pod"
13	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/config"
14	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/factory"
15	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/metrics"
16	addressset "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/address_set"
17	lsm "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/logical_switch_manager"
18	zoneinterconnect "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/ovn/zone_interconnect"
19	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/persistentips"
20	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/retry"
21	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/syncmap"
22	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/types"
23	"github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util"
24	utilerrors "github.com/ovn-org/ovn-kubernetes/go-controller/pkg/util/errors"
25
26	corev1 "k8s.io/api/core/v1"
27	kapi "k8s.io/api/core/v1"
28	"k8s.io/klog/v2"
29	)
30
31	// method/structure shared by all layer 2 network controller, including localnet and layer2 network controllres.
32
33	type secondaryLayer2NetworkControllerEventHandler struct {
34	baseHandler baseNetworkControllerEventHandler
35	watchFactory *factory.WatchFactory
36	objType reflect.Type
37	oc *SecondaryLayer2NetworkController
38	syncFunc func([]interface{}) error
39	}
40
41	// AreResourcesEqual returns true if, given two objects of a known resource type, the update logic for this resource
42	// type considers them equal and therefore no update is needed. It returns false when the two objects are not considered
43	// equal and an update needs be executed. This is regardless of how the update is carried out (whether with a dedicated update
44	// function or with a delete on the old obj followed by an add on the new obj).
45	func (h *secondaryLayer2NetworkControllerEventHandler) AreResourcesEqual(obj1, obj2 interface{}) (bool, error) {	×
46	return h.baseHandler.areResourcesEqual(h.objType, obj1, obj2)	×
47	}	×
48
49	// GetInternalCacheEntry returns the internal cache entry for this object, given an object and its type.
50	// This is now used only for pods, which will get their the logical port cache entry.
51	func (h *secondaryLayer2NetworkControllerEventHandler) GetInternalCacheEntry(obj interface{}) interface{} {	3✔
52	return h.oc.GetInternalCacheEntryForSecondaryNetwork(h.objType, obj)	3✔
53	}	3✔
54
55	// GetResourceFromInformerCache returns the latest state of the object, given an object key and its type.
56	// from the informers cache.
57	func (h *secondaryLayer2NetworkControllerEventHandler) GetResourceFromInformerCache(key string) (interface{}, error) {	×
58	return h.baseHandler.getResourceFromInformerCache(h.objType, h.watchFactory, key)	×
59	}	×
60
61	// RecordAddEvent records the add event on this given object.
62	func (h *secondaryLayer2NetworkControllerEventHandler) RecordAddEvent(obj interface{}) {	12✔
63	switch h.objType {	12✔
64	case factory.MultiNetworkPolicyType:	2✔
65	mnp := obj.(*mnpapi.MultiNetworkPolicy)	2✔
66	klog.V(5).Infof("Recording add event on multinetwork policy %s/%s", mnp.Namespace, mnp.Name)	2✔
67	metrics.GetConfigDurationRecorder().Start("multinetworkpolicy", mnp.Namespace, mnp.Name)	2✔
68	}
69	}
70
71	// RecordUpdateEvent records the udpate event on this given object.
72	func (h *secondaryLayer2NetworkControllerEventHandler) RecordUpdateEvent(obj interface{}) {	×
73	h.baseHandler.recordAddEvent(h.objType, obj)	×
74	}	×
75
76	// RecordDeleteEvent records the delete event on this given object.
77	func (h *secondaryLayer2NetworkControllerEventHandler) RecordDeleteEvent(obj interface{}) {	3✔
78	h.baseHandler.recordAddEvent(h.objType, obj)	3✔
79	}	3✔
80
81	// RecordSuccessEvent records the success event on this given object.
82	func (h *secondaryLayer2NetworkControllerEventHandler) RecordSuccessEvent(obj interface{}) {	14✔
83	h.baseHandler.recordAddEvent(h.objType, obj)	14✔
84	}	14✔
85
86	// RecordErrorEvent records the error event on this given object.
87	func (h *secondaryLayer2NetworkControllerEventHandler) RecordErrorEvent(obj interface{}, reason string, err error) {	1✔
88	}	1✔
89
90	// IsResourceScheduled returns true if the given object has been scheduled.
91	// Only applied to pods for now. Returns true for all other types.
92	func (h *secondaryLayer2NetworkControllerEventHandler) IsResourceScheduled(obj interface{}) bool {	×
93	return h.baseHandler.isResourceScheduled(h.objType, obj)	×
94	}	×
95
96	// AddResource adds the specified object to the cluster according to its type and returns the error,
97	// if any, yielded during object creation.
98	// Given an object to add and a boolean specifying if the function was executed from iterateRetryResources
99	func (h *secondaryLayer2NetworkControllerEventHandler) AddResource(obj interface{}, fromRetryLoop bool) error {	12✔
100	switch h.objType {	12✔
101	case factory.NodeType:	2✔
102	node, ok := obj.(*corev1.Node)	2✔
103	if !ok {	2✔
104	return fmt.Errorf("could not cast %T object to Node", obj)	×
105	}	×
106	if h.oc.isLocalZoneNode(node) {	3✔
107	var nodeParams *nodeSyncs	1✔
108	if fromRetryLoop {	1✔
109	_, syncMgmtPort := h.oc.mgmtPortFailed.Load(node.Name)	×
110	nodeParams = &nodeSyncs{syncMgmtPort: syncMgmtPort}	×
111	} else {	1✔
112	nodeParams = &nodeSyncs{syncMgmtPort: true}	1✔
113	}	1✔
114	return h.oc.addUpdateLocalNodeEvent(node, nodeParams)	1✔
115	}
116	return h.oc.addUpdateRemoteNodeEvent(node)	1✔
117	default:	10✔
118	return h.oc.AddSecondaryNetworkResourceCommon(h.objType, obj)	10✔
119	}
120	}
121
122	// DeleteResource deletes the object from the cluster according to the delete logic of its resource type.
123	// Given an object and optionally a cachedObj; cachedObj is the internal cache entry for this object,
124	// used for now for pods and network policies.
125	func (h *secondaryLayer2NetworkControllerEventHandler) DeleteResource(obj, cachedObj interface{}) error {	3✔
126	switch h.objType {	3✔
127	case factory.NodeType:	×
128	node, ok := obj.(*corev1.Node)	×
129	if !ok {	×
130	return fmt.Errorf("could not cast %T object to Node", obj)	×
131	}	×
132	return h.oc.deleteNodeEvent(node)	×
133	default:	3✔
134	return h.oc.DeleteSecondaryNetworkResourceCommon(h.objType, obj, cachedObj)	3✔
135	}
136	}
137
138	// UpdateResource updates the specified object in the cluster to its version in newObj according to its
139	// type and returns the error, if any, yielded during the object update.
140	// Given an old and a new object; The inRetryCache boolean argument is to indicate if the given resource
141	// is in the retryCache or not.
142	func (h *secondaryLayer2NetworkControllerEventHandler) UpdateResource(oldObj, newObj interface{}, inRetryCache bool) error {	×
143	switch h.objType {	×
144	case factory.NodeType:	×
145	newNode, ok := newObj.(*corev1.Node)	×
146	if !ok {	×
147	return fmt.Errorf("could not cast %T object to Node", newObj)	×
148	}	×
149	oldNode, ok := oldObj.(*kapi.Node)	×
150	if !ok {	×
151	return fmt.Errorf("could not cast oldObj of type %T to *kapi.Node", oldObj)	×
152	}	×
153	newNodeIsLocalZoneNode := h.oc.isLocalZoneNode(newNode)	×
154	nodeSubnetChanged := nodeSubnetChanged(oldNode, newNode)	×
155	if newNodeIsLocalZoneNode {	×
156	var nodeSyncsParam *nodeSyncs	×
157	if h.oc.isLocalZoneNode(oldNode) {	×
158	// determine what actually changed in this update	×
159	syncMgmtPort := macAddressChanged(oldNode, newNode) \|\| nodeSubnetChanged	×
160	nodeSyncsParam = &nodeSyncs{syncMgmtPort: syncMgmtPort}	×
161	} else {	×
162	klog.Infof("Node %s moved from the remote zone %s to local zone %s.",	×
163	newNode.Name, util.GetNodeZone(oldNode), util.GetNodeZone(newNode))	×
164	// The node is now a local zone node. Trigger a full node sync.	×
165	nodeSyncsParam = &nodeSyncs{syncMgmtPort: true}	×
166	}	×
167
168	return h.oc.addUpdateLocalNodeEvent(newNode, nodeSyncsParam)	×
169	} else {	×
170	return h.oc.addUpdateRemoteNodeEvent(newNode)	×
171	}	×
172	default:	×
173	return h.oc.UpdateSecondaryNetworkResourceCommon(h.objType, oldObj, newObj, inRetryCache)	×
174	}
175	}
176
177	func (h *secondaryLayer2NetworkControllerEventHandler) SyncFunc(objs []interface{}) error {	13✔
178	var syncFunc func([]interface{}) error	13✔
179		13✔
180	if h.syncFunc != nil {	13✔
181	// syncFunc was provided explicitly	×
182	syncFunc = h.syncFunc	×
183	} else {	13✔
184	switch h.objType {	13✔
185	case factory.NodeType:	2✔
186	syncFunc = h.oc.syncNodes	2✔
187
188	case factory.PodType:	3✔
189	syncFunc = h.oc.syncPodsForSecondaryNetwork	3✔
190
191	case factory.NamespaceType:	4✔
192	syncFunc = h.oc.syncNamespaces	4✔
193
194	case factory.MultiNetworkPolicyType:	4✔
195	syncFunc = h.oc.syncMultiNetworkPolicies	4✔
196
197	case factory.IPAMClaimsType:	×
198	syncFunc = h.oc.syncIPAMClaims	×
199
200	default:	×
201	return fmt.Errorf("no sync function for object type %s", h.objType)	×
202	}
203	}
204	if syncFunc == nil {	13✔
205	return nil	×
206	}	×
207	return syncFunc(objs)	13✔
208	}
209
210	// IsObjectInTerminalState returns true if the given object is a in terminal state.
211	// This is used now for pods that are either in a PodSucceeded or in a PodFailed state.
212	func (h *secondaryLayer2NetworkControllerEventHandler) IsObjectInTerminalState(obj interface{}) bool {	15✔
213	return h.baseHandler.isObjectInTerminalState(h.objType, obj)	15✔
214	}	15✔
215
216	// SecondaryLayer2NetworkController is created for logical network infrastructure and policy
217	// for a secondary layer2 network
218	type SecondaryLayer2NetworkController struct {
219	BaseSecondaryLayer2NetworkController
220
221	// Node-specific syncMaps used by node event handler
222	mgmtPortFailed sync.Map
223	}
224
225	// NewSecondaryLayer2NetworkController create a new OVN controller for the given secondary layer2 nad
226	func NewSecondaryLayer2NetworkController(cnci CommonNetworkControllerInfo, netInfo util.NetInfo) SecondaryLayer2NetworkController {	4✔
227		4✔
228	stopChan := make(chan struct{})	4✔
229		4✔
230	ipv4Mode, ipv6Mode := netInfo.IPMode()	4✔
231	addressSetFactory := addressset.NewOvnAddressSetFactory(cnci.nbClient, ipv4Mode, ipv6Mode)	4✔
232		4✔
233	lsManagerFactoryFn := lsm.NewL2SwitchManager	4✔
234	if netInfo.IsPrimaryNetwork() {	4✔
235	lsManagerFactoryFn = lsm.NewL2SwitchManagerForUserDefinedPrimaryNetwork	×
236	}	×
237	oc := &SecondaryLayer2NetworkController{	4✔
238	BaseSecondaryLayer2NetworkController: BaseSecondaryLayer2NetworkController{	4✔
239		4✔
240	BaseSecondaryNetworkController: BaseSecondaryNetworkController{	4✔
241	BaseNetworkController: BaseNetworkController{	4✔
242	CommonNetworkControllerInfo: *cnci,	4✔
243	controllerName: getNetworkControllerName(netInfo.GetNetworkName()),	4✔
244	NetInfo: netInfo,	4✔
245	lsManager: lsManagerFactoryFn(),	4✔
246	logicalPortCache: newPortCache(stopChan),	4✔
247	namespaces: make(map[string]*namespaceInfo),	4✔
248	namespacesMutex: sync.Mutex{},	4✔
249	addressSetFactory: addressSetFactory,	4✔
250	networkPolicies: syncmap.NewSyncMap[*networkPolicy](),	4✔
251	sharedNetpolPortGroups: syncmap.NewSyncMap[*defaultDenyPortGroups](),	4✔
252	podSelectorAddressSets: syncmap.NewSyncMap[*PodSelectorAddressSet](),	4✔
253	stopChan: stopChan,	4✔
254	wg: &sync.WaitGroup{},	4✔
255	localZoneNodes: &sync.Map{},	4✔
256	cancelableCtx: util.NewCancelableContext(),	4✔
257	},	4✔
258	},	4✔
259	},	4✔
260	mgmtPortFailed: sync.Map{},	4✔
261	}	4✔
262		4✔
263	if config.OVNKubernetesFeature.EnableInterconnect {	5✔
264	oc.zoneICHandler = zoneinterconnect.NewZoneInterconnectHandler(oc.NetInfo, oc.nbClient, oc.sbClient, oc.watchFactory)	1✔
265	}	1✔
266
267	if oc.allocatesPodAnnotation() {	7✔
268	var claimsReconciler persistentips.PersistentAllocations	3✔
269	if oc.allowPersistentIPs() {	3✔
270	ipamClaimsReconciler := persistentips.NewIPAMClaimReconciler(	×
271	oc.kube,	×
272	oc.NetInfo,	×
273	oc.watchFactory.IPAMClaimsInformer().Lister(),	×
274	)	×
275	oc.ipamClaimsReconciler = ipamClaimsReconciler	×
276	claimsReconciler = ipamClaimsReconciler	×
277	}	×
278	oc.podAnnotationAllocator = pod.NewPodAnnotationAllocator(	3✔
279	netInfo,	3✔
280	cnci.watchFactory.PodCoreInformer().Lister(),	3✔
281	cnci.kube,	3✔
282	claimsReconciler)	3✔
283	}
284
285	// disable multicast support for secondary networks
286	// TBD: changes needs to be made to support multicast in secondary networks
287	oc.multicastSupport = false	4✔
288		4✔
289	oc.initRetryFramework()	4✔
290	return oc	4✔
291	}
292
293	// Start starts the secondary layer2 controller, handles all events and creates all needed logical entities
294	func (oc *SecondaryLayer2NetworkController) Start(ctx context.Context) error {	×
295	klog.Infof("Starting controller for secondary network %s", oc.GetNetworkName())	×
296		×
297	start := time.Now()	×
298	defer func() {	×
299	klog.Infof("Starting controller for secondary network %s took %v", oc.GetNetworkName(), time.Since(start))	×
300	}()	×
301
NEW 302	if err := oc.BaseNetworkController.init(); err != nil {	×
NEW 303	return err	×
NEW 304	}	×
305
306	if err := oc.Init(); err != nil {	×
UNCOV 307	return err	×
UNCOV 308	}	×
309
310	return oc.run(ctx)	×
311	}
312
UNCOV 313	func (oc *SecondaryLayer2NetworkController) run(ctx context.Context) error {	×
UNCOV 314	return oc.BaseSecondaryLayer2NetworkController.run()	×
315	}	×
316
317	// Cleanup cleans up logical entities for the given network, called from net-attach-def routine
318	// could be called from a dummy Controller (only has CommonNetworkControllerInfo set)
319	func (oc *SecondaryLayer2NetworkController) Cleanup() error {	×
320	return oc.BaseSecondaryLayer2NetworkController.cleanup()	×
321	}	×
322
UNCOV 323	func (oc *SecondaryLayer2NetworkController) Init() error {	×
324	_, err := oc.initializeLogicalSwitch(oc.GetNetworkScopedSwitchName(types.OVNLayer2Switch), oc.Subnets(), oc.ExcludeSubnets())	×
325	return err	×
326	}	×
327
UNCOV 328	func (oc *SecondaryLayer2NetworkController) Stop() {	×
UNCOV 329	klog.Infof("Stoping controller for secondary network %s", oc.GetNetworkName())	×
UNCOV 330	oc.BaseSecondaryLayer2NetworkController.stop()	×
UNCOV 331	}	×
332
333	func (oc *SecondaryLayer2NetworkController) initRetryFramework() {	4✔
334	oc.retryNodes = oc.newRetryFramework(factory.NodeType)	4✔
335	oc.retryPods = oc.newRetryFramework(factory.PodType)	4✔
336	if oc.allocatesPodAnnotation() && oc.NetInfo.AllowsPersistentIPs() {	4✔
UNCOV 337	oc.retryIPAMClaims = oc.newRetryFramework(factory.IPAMClaimsType)	×
UNCOV 338	}	×
339
340	// For secondary networks, we don't have to watch namespace events if
341	// multi-network policy support is not enabled. We don't support
342	// multi-network policy for IPAM-less secondary networks either.
343	if util.IsMultiNetworkPoliciesSupportEnabled() {	8✔
344	oc.retryNamespaces = oc.newRetryFramework(factory.NamespaceType)	4✔
345	oc.retryNetworkPolicies = oc.newRetryFramework(factory.MultiNetworkPolicyType)	4✔
346	}	4✔
347	}
348
349	// newRetryFramework builds and returns a retry framework for the input resource type;
350	func (oc *SecondaryLayer2NetworkController) newRetryFramework(
351	objectType reflect.Type) *retry.RetryFramework {	16✔
352	eventHandler := &secondaryLayer2NetworkControllerEventHandler{	16✔
353	baseHandler: baseNetworkControllerEventHandler{},	16✔
354	objType: objectType,	16✔
355	watchFactory: oc.watchFactory,	16✔
356	oc: oc,	16✔
357	syncFunc: nil,	16✔
358	}	16✔
359	resourceHandler := &retry.ResourceHandler{	16✔
360	HasUpdateFunc: hasResourceAnUpdateFunc(objectType),	16✔
361	NeedsUpdateDuringRetry: needsUpdateDuringRetry(objectType),	16✔
362	ObjType: objectType,	16✔
363	EventHandler: eventHandler,	16✔
364	}	16✔
365	return retry.NewRetryFramework(	16✔
366	oc.stopChan,	16✔
367	oc.wg,	16✔
368	oc.watchFactory,	16✔
369	resourceHandler,	16✔
370	)	16✔
371	}	16✔
372
373	func (oc SecondaryLayer2NetworkController) addUpdateLocalNodeEvent(node corev1.Node, nSyncs *nodeSyncs) error {	1✔
374	var errs []error	1✔
375		1✔
376	if util.IsNetworkSegmentationSupportEnabled() && oc.IsPrimaryNetwork() {	1✔
377	if nSyncs.syncMgmtPort {	×
378	// Layer 2 networks have a single, large subnet, that's the one	×
379	// associated to the controller. Take the management port IP from	×
380	// there.	×
381	subnets := oc.Subnets()	×
382	hostSubnets := make([]*net.IPNet, 0, len(subnets))	×
383	for _, subnet := range oc.Subnets() {	×
384	hostSubnets = append(hostSubnets, subnet.CIDR)	×
385	}	×
386	if _, err := oc.syncNodeManagementPortNoRouteHostSubnets(node, oc.GetNetworkScopedSwitchName(types.OVNLayer2Switch), hostSubnets); err != nil {	×
387	errs = append(errs, err)	×
UNCOV 388	oc.mgmtPortFailed.Store(node.Name, true)	×
UNCOV 389	} else {	×
UNCOV 390	oc.mgmtPortFailed.Delete(node.Name)	×
UNCOV 391	}	×
392	}
393	}
394
395	errs = append(errs, oc.BaseSecondaryLayer2NetworkController.addUpdateLocalNodeEvent(node))	1✔
396		1✔
397	err := utilerrors.Join(errs...)	1✔
398	if err != nil {	1✔
UNCOV 399	oc.recordNodeErrorEvent(node, err)	×
400	}	×
401	return err	1✔
402	}
403
404	func (oc SecondaryLayer2NetworkController) deleteNodeEvent(node corev1.Node) error {	×
UNCOV 405	oc.localZoneNodes.Delete(node.Name)	×
UNCOV 406	oc.mgmtPortFailed.Delete(node.Name)	×
UNCOV 407	return nil	×
UNCOV 408	}	×

ovn-org / ovn-kubernetes / 10013062981

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