From 004deefd9dd061e73d9a54ee2721afab4ee8ecf2 Mon Sep 17 00:00:00 2001 From: MengxinLiu Date: Wed, 20 Mar 2019 10:45:29 +0800 Subject: [PATCH] feat: support qos --- pkg/controller/namespace.go | 2 +- pkg/daemon/handler.go | 6 +- pkg/daemon/ovs.go | 125 +++++++++++++++++++++++++++++++++++- pkg/util/const.go | 3 + 4 files changed, 132 insertions(+), 4 deletions(-) diff --git a/pkg/controller/namespace.go b/pkg/controller/namespace.go index b64c31ef20a..856ae53911c 100644 --- a/pkg/controller/namespace.go +++ b/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 diff --git a/pkg/daemon/handler.go b/pkg/daemon/handler.go index 32c85037c3f..c386535b88d 100644 --- a/pkg/daemon/handler.go +++ b/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) diff --git a/pkg/daemon/ovs.go b/pkg/daemon/ovs.go index c84166d08d8..0f1880cb2ec 100644 --- a/pkg/daemon/ovs.go +++ b/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 +} diff --git a/pkg/util/const.go b/pkg/util/const.go index aa4ef45c775..9dd41482a1f 100644 --- a/pkg/util/const.go +++ b/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"