feat: support qos

pkg/controller/namespace.go

@@ -151,7 +151,7 @@ func (c *Controller) handleDeleteNamespace(key string) error {
 	}
 
 	for _, ls := range switches {
-		if ls == c.config.DefaultLogicalSwitch || ls == c.config.NodeSwitch {
+		if ls == c.config.DefaultLogicalSwitch || ls == c.config.NodeSwitch || ls == "transit" || ls == "outside" {
 			continue
 		}
 		found := false

pkg/daemon/handler.go

@@ -36,7 +36,7 @@ func (csh CniServerHandler) handleAdd(req *restful.Request, resp *restful.Respon
 		return
 	}
 	klog.Infof("add port request %v", podRequest)
-	var macAddr, ipAddr, cidr, gw string
+	var macAddr, ipAddr, cidr, gw, ingress, egress string
 	for i := 0; i < 10; i++ {
 		pod, err := csh.KubeClient.CoreV1().Pods(podRequest.PodNamespace).Get(podRequest.PodName, v1.GetOptions{})
 		if err != nil {
@@ -48,6 +48,8 @@ func (csh CniServerHandler) handleAdd(req *restful.Request, resp *restful.Respon
 		ipAddr = pod.Annotations[util.IpAddressAnnotation]
 		cidr = pod.Annotations[util.CidrAnnotation]
 		gw = pod.Annotations[util.GatewayAnnotation]
+		ingress = pod.Annotations[util.IngressRateAnnotation]
+		egress = pod.Annotations[util.EgressRateAnnotation]
 
 		if macAddr == "" || ipAddr == "" || cidr == "" || gw == "" {
 			// wait controller assign an address
@@ -58,7 +60,7 @@ func (csh CniServerHandler) handleAdd(req *restful.Request, resp *restful.Respon
 	}
 
 	klog.Infof("create container mac %s, ip %s, cidr %s, gw %s", macAddr, ipAddr, cidr, gw)
-	err = csh.configureNic(podRequest.PodName, podRequest.PodNamespace, podRequest.NetNs, podRequest.ContainerID, macAddr, ipAddr, gw)
+	err = csh.configureNic(podRequest.PodName, podRequest.PodNamespace, podRequest.NetNs, podRequest.ContainerID, macAddr, ipAddr, gw, ingress, egress)
 	if err != nil {
 		klog.Errorf("configure nic failed %v", err)
 		resp.WriteHeaderAndEntity(http.StatusInternalServerError, err)

pkg/daemon/ovs.go

@@ -8,9 +8,11 @@ import (
 	"k8s.io/klog"
 	"net"
 	"os/exec"
+	"strconv"
+	"strings"
 )
 
-func (csh CniServerHandler) configureNic(podName, podNamespace, netns, containerID, mac, ip, gateway string) error {
+func (csh CniServerHandler) configureNic(podName, podNamespace, netns, containerID, mac, ip, gateway, ingress, egress string) error {
 	var err error
 	hostNicName, containerNicName := generateNicName(containerID)
 	// Create a veth pair, put one end to container ,the other to ovs port
@@ -48,6 +50,11 @@ func (csh CniServerHandler) configureNic(podName, podNamespace, netns, container
 		return err
 	}
 
+	err = setPodBandwidth(containerID, hostNicName, ingress, egress)
+	if err != nil {
+		return err
+	}
+
 	podNS, err := ns.GetNS(netns)
 	if err != nil {
 		return fmt.Errorf("failed to open netns %q: %v", netns, err)
@@ -80,6 +87,12 @@ func (csh CniServerHandler) deleteNic(netns, containerID string) error {
 	if err != nil {
 		return fmt.Errorf("delete host link %s failed %v", hostLink, err)
 	}
+
+	err = clearPodBandwidth(containerID)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -202,3 +215,113 @@ func configureNodeNic(portName, ip, mac string) error {
 	}
 	return nil
 }
+
+func clearPodBandwidth(sandboxID string) error {
+	// interfaces will have the same name as ports
+	portList, err := ovsFind("interface", "name", "external-ids:sandbox="+sandboxID)
+	if err != nil {
+		return err
+	}
+
+	// Clear the QoS for any ports of this sandbox
+	for _, port := range portList {
+		if err = ovsClear("port", port, "qos"); err != nil {
+			return err
+		}
+	}
+
+	// Now that the QoS is unused remove it
+	qosList, err := ovsFind("qos", "_uuid", "external-ids:sandbox="+sandboxID)
+	if err != nil {
+		return err
+	}
+	for _, qos := range qosList {
+		if err := ovsDestroy("qos", qos); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func setPodBandwidth(sandboxID, ifname string, ingress, egress string) error {
+	ingressMPS, _ := strconv.Atoi(ingress)
+	ingressKPS := ingressMPS * 1000
+	if ingressKPS > 0 {
+		// ingress_policing_rate is in Kbps
+		err := ovsSet("interface", ifname, fmt.Sprintf("ingress_policing_rate=%d", ingressKPS))
+		if err != nil {
+			return err
+		}
+	}
+	egressMPS, _ := strconv.Atoi(egress)
+	egressBPS := egressMPS * 1000 * 1000
+	if egressBPS > 0 {
+		qos, err := ovsCreate("qos", "type=linux-htb", fmt.Sprintf("other-config:max-rate=%d", egressBPS), "external-ids=sandbox="+sandboxID)
+		if err != nil {
+			return err
+		}
+		err = ovsSet("port", ifname, fmt.Sprintf("qos=%s", qos))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func ovsExec(args ...string) (string, error) {
+	args = append([]string{"--timeout=30"}, args...)
+	output, err := exec.Command("ovs-vsctl", args...).CombinedOutput()
+	if err != nil {
+		return "", fmt.Errorf("failed to run 'ovs-vsctl %s': %v\n  %q", strings.Join(args, " "), err, string(output))
+	}
+
+	outStr := string(output)
+	trimmed := strings.TrimSpace(outStr)
+	// If output is a single line, strip the trailing newline
+	if strings.Count(trimmed, "\n") == 0 {
+		outStr = trimmed
+	}
+
+	return outStr, nil
+}
+
+func ovsCreate(table string, values ...string) (string, error) {
+	args := append([]string{"create", table}, values...)
+	return ovsExec(args...)
+}
+
+func ovsDestroy(table, record string) error {
+	_, err := ovsExec("--if-exists", "destroy", table, record)
+	return err
+}
+
+func ovsSet(table, record string, values ...string) error {
+	args := append([]string{"set", table, record}, values...)
+	_, err := ovsExec(args...)
+	return err
+}
+
+// Returns the given column of records that match the condition
+func ovsFind(table, column, condition string) ([]string, error) {
+	output, err := ovsExec("--no-heading", "--columns="+column, "find", table, condition)
+	if err != nil {
+		return nil, err
+	}
+	values := strings.Split(output, "\n\n")
+	// We want "bare" values for strings, but we can't pass --bare to ovs-vsctl because
+	// it breaks more complicated types. So try passing each value through Unquote();
+	// if it fails, that means the value wasn't a quoted string, so use it as-is.
+	for i, val := range values {
+		if unquoted, err := strconv.Unquote(val); err == nil {
+			values[i] = unquoted
+		}
+	}
+	return values, nil
+}
+
+func ovsClear(table, record string, columns ...string) error {
+	args := append([]string{"--if-exists", "clear", table, record}, columns...)
+	_, err := ovsExec(args...)
+	return err
+}

pkg/util/const.go

@@ -7,6 +7,9 @@ const (
 	GatewayAnnotation    = "ovn.kubernetes.io/gateway"
 	IpPoolAnnotation     = "ovn.kubernetes.io/ip_pool"
 
+	IngressRateAnnotation = "ovn.kubernetes.io/ingress_rate"
+	EgressRateAnnotation  = "ovn.kubernetes.io/egress_rate"
+
 	PortNameAnnotation = "ovn.kubernetes.io/port_name"
 
 	LogicalSwitchAnnotation = "ovn.kubernetes.io/logical_switch"