14547474456

Committed 19 Apr 2025 08:37AM UTC coverage: 21.73% (-0.007%) from 21.737%

Build # 14547474456

Build Type

push

github

Committed by

web-flow

Commit Message

modernize: simplify code by using modern constructs (#5163)

Signed-off-by: Mengxin Liu <liumengxinfly@gmail.com>

Run Details

87 of 376 new or added lines in 94 files covered. (23.14%)

1 existing line in 1 file now uncovered.

10251 of 47175 relevant lines covered (21.73%)

0.25 hits per line

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

6.12

/pkg/controller/security_group.go

package controller

import (
        "context"
        "encoding/hex"
        "errors"
        "fmt"
        "net"
        "reflect"
        "slices"
        "strings"

        "github.com/cnf/structhash"
        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"

        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) enqueueAddSg(obj any) {
        key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()
        klog.V(3).Infof("enqueue add securityGroup %s", key)
        c.addOrUpdateSgQueue.Add(key)
}

func (c *Controller) enqueueUpdateSg(oldObj, newObj any) {
        oldSg := oldObj.(*kubeovnv1.SecurityGroup)
        newSg := newObj.(*kubeovnv1.SecurityGroup)
        if !reflect.DeepEqual(oldSg.Spec, newSg.Spec) {
                key := cache.MetaObjectToName(newSg).String()
                klog.V(3).Infof("enqueue update securityGroup %s", key)
                c.addOrUpdateSgQueue.Add(key)
        }
}

func (c *Controller) enqueueDeleteSg(obj any) {
        key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()
        klog.V(3).Infof("enqueue delete securityGroup %s", key)
        c.delSgQueue.Add(key)
}

func (c *Controller) initDefaultDenyAllSecurityGroup() error {
        pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)
        if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{
                "type": "security_group",
                sgKey:  util.DenyAllSecurityGroup,
        }); err != nil {
                klog.Errorf("create port group for sg %s: %v", util.DenyAllSecurityGroup, err)
                return err
        }

        if err := c.OVNNbClient.CreateSgDenyAllACL(util.DenyAllSecurityGroup); err != nil {
                klog.Errorf("create deny all acl for sg %s: %v", util.DenyAllSecurityGroup, err)
                return err
        }

        c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)
        return nil
}

func (c *Controller) syncSecurityGroup() error {
        sgs, err := c.sgsLister.List(labels.Everything())
        if err != nil {
                klog.Errorf("failed to list security groups: %v", err)
                return err
        }
        for _, sg := range sgs {
                lost, err := c.OVNNbClient.SGLostACL(sg)
                if err != nil {
                        err = fmt.Errorf("failed to check if security group %s lost acl: %w", sg.Name, err)
                        klog.Error(err)
                        return err
                }
                if lost {
                        if err := c.handleAddOrUpdateSg(sg.Name, true); err != nil {
                                klog.Errorf("failed to sync security group %s: %v", sg.Name, err)
                        }
                }
        }
        return nil
}

// updateDenyAllSgPorts set lsp to deny which security_groups is not empty
func (c *Controller) updateDenyAllSgPorts() error {
        // list all lsp which security_groups is not empty
        lsps, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{sgsKey: ""})
        if err != nil {
                klog.Errorf("list logical switch ports with security_groups is not empty: %v", err)
                return err
        }

        addPorts := make([]string, 0, len(lsps))
        for _, lsp := range lsps {
                /* skip lsp which security_group does not exist */
                // sgs format: sg1/sg2/sg3
                sgs := strings.Split(lsp.ExternalIDs[sgsKey], "/")
                allNotExist, err := c.securityGroupAllNotExist(sgs)
                if err != nil {
                        klog.Error(err)
                        return err
                }

                if allNotExist {
                        continue
                }

                addPorts = append(addPorts, lsp.Name)
        }
        pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)

        klog.V(6).Infof("setting ports of port group %s to %v", pgName, addPorts)
        if err = c.OVNNbClient.PortGroupSetPorts(pgName, addPorts); err != nil {
                klog.Error(err)
                return err
        }

        return nil
}

func (c *Controller) handleAddOrUpdateSg(key string, force bool) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("handle add/update security group %s", key)

        // set 'deny all' for port associated with security group
        if key == util.DenyAllSecurityGroup {
                if err := c.updateDenyAllSgPorts(); err != nil {
                        klog.Errorf("update sg deny all policy failed. %v", err)
                        return err
                }
        }

        cachedSg, err := c.sgsLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        return nil
                }
                klog.Error(err)
                return err
        }
        sg := cachedSg.DeepCopy()

        if err = c.validateSgRule(sg); err != nil {
                klog.Error(err)
                return err
        }

        pgName := ovs.GetSgPortGroupName(sg.Name)
        if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{
                "type": "security_group",
                sgKey:  sg.Name,
        }); err != nil {
                klog.Errorf("create port group for sg %s: %v", sg.Name, err)
                return err
        }

        v4AsName := ovs.GetSgV4AssociatedName(sg.Name)
        v6AsName := ovs.GetSgV6AssociatedName(sg.Name)
        externalIDs := map[string]string{
                sgKey: sg.Name,
        }

        if err = c.OVNNbClient.CreateAddressSet(v4AsName, externalIDs); err != nil {
                klog.Errorf("create address set %s for sg %s: %v", v4AsName, key, err)
                return err
        }

        if err = c.OVNNbClient.CreateAddressSet(v6AsName, externalIDs); err != nil {
                klog.Errorf("create address set %s for sg %s: %v", v6AsName, key, err)
                return err
        }

        var ingressNeedUpdate, egressNeedUpdate bool
        var newIngressMd5, newEgressMd5 string
        if force {
                klog.Infof("force update sg %s", sg.Name)
                ingressNeedUpdate = true
                egressNeedUpdate = true
        } else {
                // check md5
                newIngressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.IngressRules, 1))
                if !sg.Status.IngressLastSyncSuccess || newIngressMd5 != sg.Status.IngressMd5 {
                        klog.Infof("ingress need update, sg:%s", sg.Name)
                        ingressNeedUpdate = true
                }
                newEgressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.EgressRules, 1))
                if !sg.Status.EgressLastSyncSuccess || newEgressMd5 != sg.Status.EgressMd5 {
                        klog.Infof("egress need update, sg:%s", sg.Name)
                        egressNeedUpdate = true
                }

                // check allowSameGroupTraffic switch
                if sg.Status.AllowSameGroupTraffic != sg.Spec.AllowSameGroupTraffic {
                        klog.Infof("both ingress && egress need update, sg:%s", sg.Name)
                        ingressNeedUpdate = true
                        egressNeedUpdate = true
                }
        }

        // update sg rule
        if ingressNeedUpdate {
                if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionToLport); err != nil {
                        sg.Status.IngressLastSyncSuccess = false
                        c.patchSgStatus(sg)
                        klog.Error(err)
                        return err
                }

                if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionToLport); err != nil {
                        klog.Error(err)
                        return err
                }
                sg.Status.IngressMd5 = newIngressMd5
                sg.Status.IngressLastSyncSuccess = true
                c.patchSgStatus(sg)
        }
        if egressNeedUpdate {
                if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionFromLport); err != nil {
                        sg.Status.IngressLastSyncSuccess = false
                        c.patchSgStatus(sg)
                        klog.Error(err)
                        return err
                }

                if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionFromLport); err != nil {
                        klog.Error(err)
                        return err
                }

                sg.Status.EgressMd5 = newEgressMd5
                sg.Status.EgressLastSyncSuccess = true
                c.patchSgStatus(sg)
        }

        // update status
        sg.Status.PortGroup = ovs.GetSgPortGroupName(sg.Name)
        sg.Status.AllowSameGroupTraffic = sg.Spec.AllowSameGroupTraffic
        c.patchSgStatus(sg)
        c.syncSgPortsQueue.Add(key)
        return nil
}

func (c *Controller) validateSgRule(sg *kubeovnv1.SecurityGroup) error {
        // check sg rules
        allRules := append(sg.Spec.IngressRules, sg.Spec.EgressRules...)
        for _, rule := range allRules {
                if rule.IPVersion != "ipv4" && rule.IPVersion != "ipv6" {
                        return errors.New("IPVersion should be 'ipv4' or 'ipv6'")
                }

                if rule.Priority < 1 || rule.Priority > 200 {
                        return fmt.Errorf("priority '%d' is not in the range of 1 to 200", rule.Priority)
                }

                switch rule.RemoteType {
                case kubeovnv1.SgRemoteTypeAddress:
                        if strings.Contains(rule.RemoteAddress, "/") {
                                if _, _, err := net.ParseCIDR(rule.RemoteAddress); err != nil {
                                        return fmt.Errorf("invalid CIDR '%s'", rule.RemoteAddress)
                                }
                        } else {
                                if net.ParseIP(rule.RemoteAddress) == nil {
                                        return fmt.Errorf("invalid ip address '%s'", rule.RemoteAddress)
                                }
                        }
                case kubeovnv1.SgRemoteTypeSg:
                        _, err := c.sgsLister.Get(rule.RemoteSecurityGroup)
                        if err != nil {
                                return fmt.Errorf("failed to get remote sg '%s', %w", rule.RemoteSecurityGroup, err)
                        }
                default:
                        return fmt.Errorf("not support sgRemoteType '%s'", rule.RemoteType)
                }

                if rule.Protocol == kubeovnv1.SgProtocolTCP || rule.Protocol == kubeovnv1.SgProtocolUDP {
                        if rule.PortRangeMin < 1 || rule.PortRangeMin > 65535 || rule.PortRangeMax < 1 || rule.PortRangeMax > 65535 {
                                return errors.New("portRange is out of range")
                        }
                        if rule.PortRangeMin > rule.PortRangeMax {
                                return errors.New("portRange err, range Minimum value greater than maximum value")
                        }
                }
        }
        return nil
}

func (c *Controller) patchSgStatus(sg *kubeovnv1.SecurityGroup) {
        bytes, err := sg.Status.Bytes()
        if err != nil {
                klog.Error(err)
                return
        }
        if _, err = c.config.KubeOvnClient.KubeovnV1().SecurityGroups().Patch(context.Background(), sg.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {
                klog.Error("patch security group status failed", err)
        }
}

func (c *Controller) handleDeleteSg(key string) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("handle delete security group %s", key)

        if err := c.OVNNbClient.DeleteSecurityGroup(key); err != nil {
                klog.Errorf("delete sg %s: %v", key, err)
                return err
        }

        return nil
}

func (c *Controller) syncSgLogicalPort(key string) error {
        c.sgKeyMutex.LockKey(key)
        defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()
        klog.Infof("sync lsp for security group %s", key)

        sgPorts, err := c.OVNNbClient.ListLogicalSwitchPorts(false, map[string]string{"associated_sg_" + key: "true"}, nil)
        if err != nil {
                klog.Errorf("failed to find logical port, %v", err)
                return err
        }

        var ports, v4s, v6s, addresses []string
        for _, lsp := range sgPorts {
                ports = append(ports, lsp.Name)
                if len(lsp.PortSecurity) != 0 {
                        addresses = lsp.PortSecurity
                } else {
                        addresses = lsp.Addresses
                }
                for _, as := range addresses {
                        fields := strings.Fields(as)
                        if len(fields) < 2 {
                                continue
                        }
                        for _, address := range fields[1:] {
                                if strings.Contains(address, ":") {
                                        v6s = append(v6s, address)
                                } else {
                                        v4s = append(v4s, address)
                                }
                        }
                }
        }

        sg, err := c.sgsLister.Get(key)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        klog.Warningf("no security group %s", key)
                        return nil
                }
                klog.Errorf("failed to get security group %s: %v", key, err)
                return err
        }

        if err = c.OVNNbClient.PortGroupSetPorts(sg.Status.PortGroup, ports); err != nil {
                klog.Errorf("add ports to port group %s: %v", sg.Status.PortGroup, err)
                return err
        }

        v4AsName := ovs.GetSgV4AssociatedName(key)
        if err := c.OVNNbClient.AddressSetUpdateAddress(v4AsName, v4s...); err != nil {
                klog.Errorf("set ips to address set %s: %v", v4AsName, err)
                return err
        }

        v6AsName := ovs.GetSgV6AssociatedName(key)
        if err := c.OVNNbClient.AddressSetUpdateAddress(v6AsName, v6s...); err != nil {
                klog.Errorf("set ips to address set %s: %v", v6AsName, err)
                return err
        }

        c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)
        return nil
}

func (c *Controller) getPortSg(port *ovnnb.LogicalSwitchPort) ([]string, error) {
        var sgList []string
        for key, value := range port.ExternalIDs {
                if strings.HasPrefix(key, "associated_sg_") && value == "true" {
                        sgName := strings.ReplaceAll(key, "associated_sg_", "")
                        sgList = append(sgList, sgName)
                }
        }
        return sgList, nil
}

func (c *Controller) reconcilePortSg(portName, securityGroups string) error {
        port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, false)
        if err != nil {
                klog.Errorf("failed to get logical switch port %s: %v", portName, err)
                return err
        }
        oldSgList, err := c.getPortSg(port)
        if err != nil {
                klog.Errorf("get port sg failed, %v", err)
                return err
        }
        klog.Infof("reconcile port sg, port='%s', oldSgList='%s', newSgList='%s'", portName, oldSgList, securityGroups)

        newSgList := strings.Split(securityGroups, ",")
        diffSgList := util.DiffStringSlice(oldSgList, newSgList)
        for _, sgName := range diffSgList {
                if sgName == "" {
                        continue
                }
                needAssociated := "false"
                if slices.Contains(newSgList, sgName) {
                        needAssociated = "true"
                }

                if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"associated_sg_" + sgName: needAssociated}); err != nil {
                        klog.Errorf("set logical switch port %s external_ids: %v", portName, err)
                        return err
                }
                c.syncSgPortsQueue.Add(sgName)
        }

        if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"security_groups": strings.ReplaceAll(securityGroups, ",", "/")}); err != nil {
                klog.Errorf("set logical switch port %s external_ids: %v", portName, err)
                return err
        }

        return nil
}

// securityGroupAllNotExist return true if all sgs does not exist
func (c *Controller) securityGroupAllNotExist(sgs []string) (bool, error) {
        if len(sgs) == 0 {
                return true, nil
        }

        notExistsCount := 0
        // sgs format: sg1/sg2/sg3
        for _, sg := range sgs {
                ok, err := c.OVNNbClient.PortGroupExists(ovs.GetSgPortGroupName(sg))
                if err != nil {
                        klog.Error(err)
                        return true, err
                }

                if !ok {
                        notExistsCount++
                }
        }

        return notExistsCount == len(sgs), nil
}

1	package controller
2
3	import (
4	"context"
5	"encoding/hex"
6	"errors"
7	"fmt"
8	"net"
9	"reflect"
10	"slices"
11	"strings"
12
13	"github.com/cnf/structhash"
14	k8serrors "k8s.io/apimachinery/pkg/api/errors"
15	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
16	"k8s.io/apimachinery/pkg/labels"
17	"k8s.io/apimachinery/pkg/types"
18	"k8s.io/client-go/tools/cache"
19	"k8s.io/klog/v2"
20
21	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
22	"github.com/kubeovn/kube-ovn/pkg/ovs"
23	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
24	"github.com/kubeovn/kube-ovn/pkg/util"
25	)
26
NEW 27	func (c *Controller) enqueueAddSg(obj any) {	×
28	key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()	×
29	klog.V(3).Infof("enqueue add securityGroup %s", key)	×
30	c.addOrUpdateSgQueue.Add(key)	×
31	}	×
32
NEW 33	func (c *Controller) enqueueUpdateSg(oldObj, newObj any) {	×
34	oldSg := oldObj.(*kubeovnv1.SecurityGroup)	×
35	newSg := newObj.(*kubeovnv1.SecurityGroup)	×
36	if !reflect.DeepEqual(oldSg.Spec, newSg.Spec) {	×
37	key := cache.MetaObjectToName(newSg).String()	×
38	klog.V(3).Infof("enqueue update securityGroup %s", key)	×
39	c.addOrUpdateSgQueue.Add(key)	×
40	}	×
41	}
42
NEW 43	func (c *Controller) enqueueDeleteSg(obj any) {	×
44	key := cache.MetaObjectToName(obj.(*kubeovnv1.SecurityGroup)).String()	×
45	klog.V(3).Infof("enqueue delete securityGroup %s", key)	×
46	c.delSgQueue.Add(key)	×
47	}	×
48
49	func (c *Controller) initDefaultDenyAllSecurityGroup() error {	×
50	pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)	×
51	if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{	×
52	"type": "security_group",	×
53	sgKey: util.DenyAllSecurityGroup,	×
54	}); err != nil {	×
55	klog.Errorf("create port group for sg %s: %v", util.DenyAllSecurityGroup, err)	×
56	return err	×
57	}	×
58
59	if err := c.OVNNbClient.CreateSgDenyAllACL(util.DenyAllSecurityGroup); err != nil {	×
60	klog.Errorf("create deny all acl for sg %s: %v", util.DenyAllSecurityGroup, err)	×
61	return err	×
62	}	×
63
64	c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)	×
65	return nil	×
66	}
67
68	func (c *Controller) syncSecurityGroup() error {	×
69	sgs, err := c.sgsLister.List(labels.Everything())	×
70	if err != nil {	×
71	klog.Errorf("failed to list security groups: %v", err)	×
72	return err	×
73	}	×
74	for _, sg := range sgs {	×
75	lost, err := c.OVNNbClient.SGLostACL(sg)	×
76	if err != nil {	×
77	err = fmt.Errorf("failed to check if security group %s lost acl: %w", sg.Name, err)	×
78	klog.Error(err)	×
79	return err	×
80	}	×
81	if lost {	×
82	if err := c.handleAddOrUpdateSg(sg.Name, true); err != nil {	×
83	klog.Errorf("failed to sync security group %s: %v", sg.Name, err)	×
84	}	×
85	}
86	}
87	return nil	×
88	}
89
90	// updateDenyAllSgPorts set lsp to deny which security_groups is not empty
91	func (c *Controller) updateDenyAllSgPorts() error {	×
92	// list all lsp which security_groups is not empty	×
93	lsps, err := c.OVNNbClient.ListNormalLogicalSwitchPorts(true, map[string]string{sgsKey: ""})	×
94	if err != nil {	×
95	klog.Errorf("list logical switch ports with security_groups is not empty: %v", err)	×
96	return err	×
97	}	×
98
99	addPorts := make([]string, 0, len(lsps))	×
100	for _, lsp := range lsps {	×
101	/* skip lsp which security_group does not exist */	×
102	// sgs format: sg1/sg2/sg3	×
103	sgs := strings.Split(lsp.ExternalIDs[sgsKey], "/")	×
104	allNotExist, err := c.securityGroupAllNotExist(sgs)	×
105	if err != nil {	×
106	klog.Error(err)	×
107	return err	×
108	}	×
109
110	if allNotExist {	×
111	continue	×
112	}
113
114	addPorts = append(addPorts, lsp.Name)	×
115	}
116	pgName := ovs.GetSgPortGroupName(util.DenyAllSecurityGroup)	×
117		×
118	klog.V(6).Infof("setting ports of port group %s to %v", pgName, addPorts)	×
119	if err = c.OVNNbClient.PortGroupSetPorts(pgName, addPorts); err != nil {	×
120	klog.Error(err)	×
121	return err	×
122	}	×
123
124	return nil	×
125	}
126
127	func (c *Controller) handleAddOrUpdateSg(key string, force bool) error {	×
128	c.sgKeyMutex.LockKey(key)	×
129	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
130	klog.Infof("handle add/update security group %s", key)	×
131		×
132	// set 'deny all' for port associated with security group	×
133	if key == util.DenyAllSecurityGroup {	×
134	if err := c.updateDenyAllSgPorts(); err != nil {	×
135	klog.Errorf("update sg deny all policy failed. %v", err)	×
136	return err	×
137	}	×
138	}
139
140	cachedSg, err := c.sgsLister.Get(key)	×
141	if err != nil {	×
142	if k8serrors.IsNotFound(err) {	×
143	return nil	×
144	}	×
145	klog.Error(err)	×
146	return err	×
147	}
148	sg := cachedSg.DeepCopy()	×
149		×
150	if err = c.validateSgRule(sg); err != nil {	×
151	klog.Error(err)	×
152	return err	×
153	}	×
154
155	pgName := ovs.GetSgPortGroupName(sg.Name)	×
156	if err := c.OVNNbClient.CreatePortGroup(pgName, map[string]string{	×
157	"type": "security_group",	×
158	sgKey: sg.Name,	×
159	}); err != nil {	×
160	klog.Errorf("create port group for sg %s: %v", sg.Name, err)	×
161	return err	×
162	}	×
163
164	v4AsName := ovs.GetSgV4AssociatedName(sg.Name)	×
165	v6AsName := ovs.GetSgV6AssociatedName(sg.Name)	×
166	externalIDs := map[string]string{	×
167	sgKey: sg.Name,	×
168	}	×
169		×
170	if err = c.OVNNbClient.CreateAddressSet(v4AsName, externalIDs); err != nil {	×
171	klog.Errorf("create address set %s for sg %s: %v", v4AsName, key, err)	×
172	return err	×
173	}	×
174
175	if err = c.OVNNbClient.CreateAddressSet(v6AsName, externalIDs); err != nil {	×
176	klog.Errorf("create address set %s for sg %s: %v", v6AsName, key, err)	×
177	return err	×
178	}	×
179
180	var ingressNeedUpdate, egressNeedUpdate bool	×
181	var newIngressMd5, newEgressMd5 string	×
182	if force {	×
183	klog.Infof("force update sg %s", sg.Name)	×
184	ingressNeedUpdate = true	×
185	egressNeedUpdate = true	×
186	} else {	×
187	// check md5	×
188	newIngressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.IngressRules, 1))	×
189	if !sg.Status.IngressLastSyncSuccess \|\| newIngressMd5 != sg.Status.IngressMd5 {	×
190	klog.Infof("ingress need update, sg:%s", sg.Name)	×
191	ingressNeedUpdate = true	×
192	}	×
193	newEgressMd5 = hex.EncodeToString(structhash.Md5(sg.Spec.EgressRules, 1))	×
194	if !sg.Status.EgressLastSyncSuccess \|\| newEgressMd5 != sg.Status.EgressMd5 {	×
195	klog.Infof("egress need update, sg:%s", sg.Name)	×
196	egressNeedUpdate = true	×
197	}	×
198
199	// check allowSameGroupTraffic switch
200	if sg.Status.AllowSameGroupTraffic != sg.Spec.AllowSameGroupTraffic {	×
201	klog.Infof("both ingress && egress need update, sg:%s", sg.Name)	×
202	ingressNeedUpdate = true	×
203	egressNeedUpdate = true	×
204	}	×
205	}
206
207	// update sg rule
208	if ingressNeedUpdate {	×
209	if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionToLport); err != nil {	×
210	sg.Status.IngressLastSyncSuccess = false	×
211	c.patchSgStatus(sg)	×
212	klog.Error(err)	×
213	return err	×
214	}	×
215
216	if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionToLport); err != nil {	×
217	klog.Error(err)	×
218	return err	×
219	}	×
220	sg.Status.IngressMd5 = newIngressMd5	×
221	sg.Status.IngressLastSyncSuccess = true	×
222	c.patchSgStatus(sg)	×
223	}
224	if egressNeedUpdate {	×
225	if err = c.OVNNbClient.UpdateSgACL(sg, ovnnb.ACLDirectionFromLport); err != nil {	×
226	sg.Status.IngressLastSyncSuccess = false	×
227	c.patchSgStatus(sg)	×
228	klog.Error(err)	×
229	return err	×
230	}	×
231
232	if err := c.OVNNbClient.CreateSgBaseACL(sg.Name, ovnnb.ACLDirectionFromLport); err != nil {	×
233	klog.Error(err)	×
234	return err	×
235	}	×
236
237	sg.Status.EgressMd5 = newEgressMd5	×
238	sg.Status.EgressLastSyncSuccess = true	×
239	c.patchSgStatus(sg)	×
240	}
241
242	// update status
243	sg.Status.PortGroup = ovs.GetSgPortGroupName(sg.Name)	×
244	sg.Status.AllowSameGroupTraffic = sg.Spec.AllowSameGroupTraffic	×
245	c.patchSgStatus(sg)	×
246	c.syncSgPortsQueue.Add(key)	×
247	return nil	×
248	}
249
250	func (c Controller) validateSgRule(sg kubeovnv1.SecurityGroup) error {	×
251	// check sg rules	×
252	allRules := append(sg.Spec.IngressRules, sg.Spec.EgressRules...)	×
253	for _, rule := range allRules {	×
254	if rule.IPVersion != "ipv4" && rule.IPVersion != "ipv6" {	×
255	return errors.New("IPVersion should be 'ipv4' or 'ipv6'")	×
256	}	×
257
258	if rule.Priority < 1 \|\| rule.Priority > 200 {	×
259	return fmt.Errorf("priority '%d' is not in the range of 1 to 200", rule.Priority)	×
260	}	×
261
262	switch rule.RemoteType {	×
263	case kubeovnv1.SgRemoteTypeAddress:	×
264	if strings.Contains(rule.RemoteAddress, "/") {	×
265	if _, _, err := net.ParseCIDR(rule.RemoteAddress); err != nil {	×
266	return fmt.Errorf("invalid CIDR '%s'", rule.RemoteAddress)	×
267	}	×
268	} else {	×
269	if net.ParseIP(rule.RemoteAddress) == nil {	×
270	return fmt.Errorf("invalid ip address '%s'", rule.RemoteAddress)	×
271	}	×
272	}
273	case kubeovnv1.SgRemoteTypeSg:	×
274	_, err := c.sgsLister.Get(rule.RemoteSecurityGroup)	×
275	if err != nil {	×
276	return fmt.Errorf("failed to get remote sg '%s', %w", rule.RemoteSecurityGroup, err)	×
277	}	×
278	default:	×
279	return fmt.Errorf("not support sgRemoteType '%s'", rule.RemoteType)	×
280	}
281
282	if rule.Protocol == kubeovnv1.SgProtocolTCP \|\| rule.Protocol == kubeovnv1.SgProtocolUDP {	×
283	if rule.PortRangeMin < 1 \|\| rule.PortRangeMin > 65535 \|\| rule.PortRangeMax < 1 \|\| rule.PortRangeMax > 65535 {	×
284	return errors.New("portRange is out of range")	×
285	}	×
286	if rule.PortRangeMin > rule.PortRangeMax {	×
287	return errors.New("portRange err, range Minimum value greater than maximum value")	×
288	}	×
289	}
290	}
291	return nil	×
292	}
293
294	func (c Controller) patchSgStatus(sg kubeovnv1.SecurityGroup) {	×
295	bytes, err := sg.Status.Bytes()	×
296	if err != nil {	×
297	klog.Error(err)	×
298	return	×
299	}	×
300	if _, err = c.config.KubeOvnClient.KubeovnV1().SecurityGroups().Patch(context.Background(), sg.Name, types.MergePatchType, bytes, metav1.PatchOptions{}, "status"); err != nil {	×
301	klog.Error("patch security group status failed", err)	×
302	}	×
303	}
304
305	func (c *Controller) handleDeleteSg(key string) error {	×
306	c.sgKeyMutex.LockKey(key)	×
307	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
308	klog.Infof("handle delete security group %s", key)	×
309		×
310	if err := c.OVNNbClient.DeleteSecurityGroup(key); err != nil {	×
311	klog.Errorf("delete sg %s: %v", key, err)	×
312	return err	×
313	}	×
314
315	return nil	×
316	}
317
318	func (c *Controller) syncSgLogicalPort(key string) error {	×
319	c.sgKeyMutex.LockKey(key)	×
320	defer func() { _ = c.sgKeyMutex.UnlockKey(key) }()	×
321	klog.Infof("sync lsp for security group %s", key)	×
322		×
NEW 323	sgPorts, err := c.OVNNbClient.ListLogicalSwitchPorts(false, map[string]string{"associated_sg_" + key: "true"}, nil)	×
324	if err != nil {	×
325	klog.Errorf("failed to find logical port, %v", err)	×
326	return err	×
327	}	×
328
329	var ports, v4s, v6s, addresses []string	×
330	for _, lsp := range sgPorts {	×
331	ports = append(ports, lsp.Name)	×
332	if len(lsp.PortSecurity) != 0 {	×
333	addresses = lsp.PortSecurity	×
334	} else {	×
335	addresses = lsp.Addresses	×
336	}	×
337	for _, as := range addresses {	×
338	fields := strings.Fields(as)	×
339	if len(fields) < 2 {	×
340	continue	×
341	}
342	for _, address := range fields[1:] {	×
343	if strings.Contains(address, ":") {	×
344	v6s = append(v6s, address)	×
345	} else {	×
346	v4s = append(v4s, address)	×
347	}	×
348	}
349	}
350	}
351
352	sg, err := c.sgsLister.Get(key)	×
353	if err != nil {	×
354	if k8serrors.IsNotFound(err) {	×
355	klog.Warningf("no security group %s", key)	×
356	return nil	×
357	}	×
358	klog.Errorf("failed to get security group %s: %v", key, err)	×
359	return err	×
360	}
361
362	if err = c.OVNNbClient.PortGroupSetPorts(sg.Status.PortGroup, ports); err != nil {	×
363	klog.Errorf("add ports to port group %s: %v", sg.Status.PortGroup, err)	×
364	return err	×
365	}	×
366
367	v4AsName := ovs.GetSgV4AssociatedName(key)	×
368	if err := c.OVNNbClient.AddressSetUpdateAddress(v4AsName, v4s...); err != nil {	×
369	klog.Errorf("set ips to address set %s: %v", v4AsName, err)	×
370	return err	×
371	}	×
372
373	v6AsName := ovs.GetSgV6AssociatedName(key)	×
374	if err := c.OVNNbClient.AddressSetUpdateAddress(v6AsName, v6s...); err != nil {	×
375	klog.Errorf("set ips to address set %s: %v", v6AsName, err)	×
376	return err	×
377	}	×
378
379	c.addOrUpdateSgQueue.Add(util.DenyAllSecurityGroup)	×
380	return nil	×
381	}
382
383	func (c Controller) getPortSg(port ovnnb.LogicalSwitchPort) ([]string, error) {	1✔
384	var sgList []string	1✔
385	for key, value := range port.ExternalIDs {	2✔
386	if strings.HasPrefix(key, "associated_sg_") && value == "true" {	2✔
387	sgName := strings.ReplaceAll(key, "associated_sg_", "")	1✔
388	sgList = append(sgList, sgName)	1✔
389	}	1✔
390	}
391	return sgList, nil	1✔
392	}
393
394	func (c *Controller) reconcilePortSg(portName, securityGroups string) error {	×
395	port, err := c.OVNNbClient.GetLogicalSwitchPort(portName, false)	×
396	if err != nil {	×
397	klog.Errorf("failed to get logical switch port %s: %v", portName, err)	×
398	return err	×
399	}	×
400	oldSgList, err := c.getPortSg(port)	×
401	if err != nil {	×
402	klog.Errorf("get port sg failed, %v", err)	×
403	return err	×
404	}	×
405	klog.Infof("reconcile port sg, port='%s', oldSgList='%s', newSgList='%s'", portName, oldSgList, securityGroups)	×
406		×
407	newSgList := strings.Split(securityGroups, ",")	×
408	diffSgList := util.DiffStringSlice(oldSgList, newSgList)	×
409	for _, sgName := range diffSgList {	×
410	if sgName == "" {	×
411	continue	×
412	}
413	needAssociated := "false"	×
414	if slices.Contains(newSgList, sgName) {	×
415	needAssociated = "true"	×
416	}	×
417
NEW 418	if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"associated_sg_" + sgName: needAssociated}); err != nil {	×
419	klog.Errorf("set logical switch port %s external_ids: %v", portName, err)	×
420	return err	×
421	}	×
422	c.syncSgPortsQueue.Add(sgName)	×
423	}
424
425	if err = c.OVNNbClient.SetLogicalSwitchPortExternalIDs(portName, map[string]string{"security_groups": strings.ReplaceAll(securityGroups, ",", "/")}); err != nil {	×
426	klog.Errorf("set logical switch port %s external_ids: %v", portName, err)	×
427	return err	×
428	}	×
429
430	return nil	×
431	}
432
433	// securityGroupAllNotExist return true if all sgs does not exist
434	func (c *Controller) securityGroupAllNotExist(sgs []string) (bool, error) {	1✔
435	if len(sgs) == 0 {	2✔
436	return true, nil	1✔
437	}	1✔
438
439	notExistsCount := 0	1✔
440	// sgs format: sg1/sg2/sg3	1✔
441	for _, sg := range sgs {	2✔
442	ok, err := c.OVNNbClient.PortGroupExists(ovs.GetSgPortGroupName(sg))	1✔
443	if err != nil {	1✔
444	klog.Error(err)	×
445	return true, err	×
446	}	×
447
448	if !ok {	2✔
449	notExistsCount++	1✔
450	}	1✔
451	}
452
453	return notExistsCount == len(sgs), nil	1✔
454	}

kubeovn / kube-ovn / 14547474456

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