plugins/meta/bandwith: traffic shaping plugin

Add chained plugin to add a tbf qdisc to shape ingress/egress traffic

plugins/meta/bandwidth/ifb_creator.go

@@ -0,0 +1,165 @@
+package bandwidth
+
+// Copyright 2018 CNI authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import (
+	"fmt"
+	"github.com/containernetworking/plugins/pkg/ip"
+	"github.com/vishvananda/netlink"
+	"net"
+	"syscall"
+)
+
+const latencyInMillis = 25
+
+func CreateIfb(ifbDeviceName string, mtu int) error {
+	err := netlink.LinkAdd(&netlink.Ifb{
+		LinkAttrs: netlink.LinkAttrs{
+			Name:  ifbDeviceName,
+			Flags: net.FlagUp,
+			MTU:   mtu,
+		},
+	})
+
+	if err != nil {
+		return fmt.Errorf("adding link: %s", err)
+	}
+
+	return nil
+}
+
+func TeardownIfb(deviceName string) error {
+	_, err := ip.DelLinkByNameAddr(deviceName)
+	if err != nil && err == ip.ErrLinkNotFound {
+		return nil
+	}
+	return err
+}
+
+func CreateIngressQdisc(rateInBits, burstInBits int, hostDeviceName string) error {
+	hostDevice, err := netlink.LinkByName(hostDeviceName)
+	if err != nil {
+		return fmt.Errorf("get host device: %s", err)
+	}
+	return createTBF(rateInBits, burstInBits, hostDevice.Attrs().Index)
+}
+
+func CreateEgressQdisc(rateInBits, burstInBits int, hostDeviceName string, ifbDeviceName string) error {
+	ifbDevice, err := netlink.LinkByName(ifbDeviceName)
+	if err != nil {
+		return fmt.Errorf("get ifb device: %s", err)
+	}
+	hostDevice, err := netlink.LinkByName(hostDeviceName)
+	if err != nil {
+		return fmt.Errorf("get host device: %s", err)
+	}
+
+	// add qdisc ingress on host device
+	ingress := &netlink.Ingress{
+		QdiscAttrs: netlink.QdiscAttrs{
+			LinkIndex: hostDevice.Attrs().Index,
+			Handle:    netlink.MakeHandle(0xffff, 0), // ffff:
+			Parent:    netlink.HANDLE_INGRESS,
+		},
+	}
+
+	err = netlink.QdiscAdd(ingress)
+	if err != nil {
+		return fmt.Errorf("create ingress qdisc: %s", err)
+	}
+
+	// add filter on host device to mirror traffic to ifb device
+	filter := &netlink.U32{
+		FilterAttrs: netlink.FilterAttrs{
+			LinkIndex: hostDevice.Attrs().Index,
+			Parent:    ingress.QdiscAttrs.Handle,
+			Priority:  1,
+			Protocol:  syscall.ETH_P_ALL,
+		},
+		ClassId:    netlink.MakeHandle(1, 1),
+		RedirIndex: ifbDevice.Attrs().Index,
+		Actions: []netlink.Action{
+			&netlink.MirredAction{
+				ActionAttrs:  netlink.ActionAttrs{},
+				MirredAction: netlink.TCA_EGRESS_REDIR,
+				Ifindex:      ifbDevice.Attrs().Index,
+			},
+		},
+	}
+	err = netlink.FilterAdd(filter)
+	if err != nil {
+		return fmt.Errorf("add filter: %s", err)
+	}
+
+	// throttle traffic on ifb device
+	err = createTBF(rateInBits, burstInBits, ifbDevice.Attrs().Index)
+	if err != nil {
+		return fmt.Errorf("create ifb qdisc: %s", err)
+	}
+	return nil
+}
+
+func createTBF(rateInBits, burstInBits, linkIndex int) error {
+	// Equivalent to
+	// tc qdisc add dev link root tbf
+	//		rate netConf.BandwidthLimits.Rate
+	//		burst netConf.BandwidthLimits.Burst
+	if rateInBits <= 0 {
+		return fmt.Errorf("invalid rate: %d", rateInBits)
+	}
+	if burstInBits <= 0 {
+		return fmt.Errorf("invalid burst: %d", burstInBits)
+	}
+	rateInBytes := rateInBits / 8
+	bufferInBytes := buffer(uint64(rateInBytes), uint32(burstInBits))
+	latency := latencyInUsec(latencyInMillis)
+	limitInBytes := limit(uint64(rateInBytes), latency, uint32(bufferInBytes))
+
+	qdisc := &netlink.Tbf{
+		QdiscAttrs: netlink.QdiscAttrs{
+			LinkIndex: linkIndex,
+			Handle:    netlink.MakeHandle(1, 0),
+			Parent:    netlink.HANDLE_ROOT,
+		},
+		Limit:  uint32(limitInBytes),
+		Rate:   uint64(rateInBytes),
+		Buffer: uint32(bufferInBytes),
+	}
+	err := netlink.QdiscAdd(qdisc)
+	if err != nil {
+		return fmt.Errorf("create qdisc: %s", err)
+	}
+	return nil
+}
+
+func tick2Time(tick uint32) uint32 {
+	return uint32(float64(tick) / float64(netlink.TickInUsec()))
+}
+
+func time2Tick(time uint32) uint32 {
+	return uint32(float64(time) * float64(netlink.TickInUsec()))
+}
+
+func buffer(rate uint64, burst uint32) uint32 {
+	return time2Tick(uint32(float64(burst) * float64(netlink.TIME_UNITS_PER_SEC) / float64(rate)))
+}
+
+func limit(rate uint64, latency float64, buffer uint32) uint32 {
+	return uint32(float64(rate) / float64(netlink.TIME_UNITS_PER_SEC) * (latency + float64(tick2Time(buffer))))
+}
+
+func latencyInUsec(latencyInMillis float64) float64 {
+	return float64(netlink.TIME_UNITS_PER_SEC) * (latencyInMillis / 1000.0)
+}