ITQC/containernetworking-plugins
24
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 34 Wiki Activity

Add check support for: bridge, ipvlan, macvlan, p2p, vlan and host-device main plugins

Browse Source 
host-local and static ipam plugins
  tuning, bandwidth and portmap meta plugins

  Utility functions created for common PrevResult checking

  Fix windows build
This commit is contained in:
Michael Cambria
2018-12-06 15:42:37 -05:00
parent 82a0651d0a
commit 74a2596573
28 changed files with 3759 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
plugins/meta/bandwidth/bandwidth_linux_test.go
View File

@ -35,6 +35,49 @@ import (
"github.com/onsi/gomega/gexec"
)
func buildOneConfig(name, cniVersion string, orig *PluginConf, prevResult types.Result) (*PluginConf, []byte, error) {
var err error
inject := map[string]interface{}{
"name": name,
"cniVersion": cniVersion,
}
// Add previous plugin result
if prevResult != nil {
inject["prevResult"] = prevResult
}
// Ensure every config uses the same name and version
config := make(map[string]interface{})
confBytes, err := json.Marshal(orig)
if err != nil {
return nil, nil, err
}
err = json.Unmarshal(confBytes, &config)
if err != nil {
return nil, nil, fmt.Errorf("unmarshal existing network bytes: %s", err)
}
for key, value := range inject {
config[key] = value
}
newBytes, err := json.Marshal(config)
if err != nil {
return nil, nil, err
}
conf := &PluginConf{}
if err := json.Unmarshal(newBytes, &conf); err != nil {
return nil, nil, fmt.Errorf("error parsing configuration: %s", err)
}
return conf, newBytes, nil
}
var _ = Describe("bandwidth test", func() {
var (
hostNs ns.NetNS
@ -643,7 +686,6 @@ var _ = Describe("bandwidth test", func() {
containerWithTbfResult, err := current.GetResult(containerWithTbfRes)
Expect(err).NotTo(HaveOccurred())
tbfPluginConf := PluginConf{}
tbfPluginConf.RuntimeConfig.Bandwidth = &BandwidthEntry{
IngressBurst: burstInBits,
@ -654,7 +696,7 @@ var _ = Describe("bandwidth test", func() {
tbfPluginConf.Name = "mynet"
tbfPluginConf.CNIVersion = "0.3.0"
tbfPluginConf.Type = "bandwidth"
tbfPluginConf.RawPrevResult = &map[string]interface{}{
tbfPluginConf.RawPrevResult = map[string]interface{}{
"ips": containerWithTbfResult.IPs,
"interfaces": containerWithTbfResult.Interfaces,
}
@ -663,7 +705,6 @@ var _ = Describe("bandwidth test", func() {
IPs: containerWithTbfResult.IPs,
Interfaces: containerWithTbfResult.Interfaces,
}
conf, err := json.Marshal(tbfPluginConf)
Expect(err).NotTo(HaveOccurred())
@ -725,4 +766,169 @@ var _ = Describe("bandwidth test", func() {
}, 1)
})
Context("when chaining bandwidth plugin with PTP using 0.4.0 config", func() {
var ptpConf string
var rateInBits int
var burstInBits int
var packetInBytes int
var containerWithoutTbfNS ns.NetNS
var containerWithTbfNS ns.NetNS
var portServerWithTbf int
var portServerWithoutTbf int
var containerWithTbfRes types.Result
var containerWithoutTbfRes types.Result
var echoServerWithTbf *gexec.Session
var echoServerWithoutTbf *gexec.Session
BeforeEach(func() {
rateInBytes := 1000
rateInBits = rateInBytes * 8
burstInBits = rateInBits * 2
packetInBytes = rateInBytes * 25
ptpConf = `{
"cniVersion": "0.4.0",
"name": "myBWnet",
"type": "ptp",
"ipMasq": true,
"mtu": 512,
"ipam": {
"type": "host-local",
"subnet": "10.1.2.0/24"
}
}`
containerWithTbfIFName := "ptp0"
containerWithoutTbfIFName := "ptp1"
var err error
containerWithTbfNS, err = testutils.NewNS()
Expect(err).NotTo(HaveOccurred())
containerWithoutTbfNS, err = testutils.NewNS()
Expect(err).NotTo(HaveOccurred())
By("create two containers, and use the bandwidth plugin on one of them")
Expect(hostNs.Do(func(ns.NetNS) error {
defer GinkgoRecover()
containerWithTbfRes, _, err = testutils.CmdAdd(containerWithTbfNS.Path(), "dummy", containerWithTbfIFName, []byte(ptpConf), func() error {
r, err := invoke.DelegateAdd(context.TODO(), "ptp", []byte(ptpConf), nil)
Expect(r.Print()).To(Succeed())
return err
})
Expect(err).NotTo(HaveOccurred())
containerWithoutTbfRes, _, err = testutils.CmdAdd(containerWithoutTbfNS.Path(), "dummy2", containerWithoutTbfIFName, []byte(ptpConf), func() error {
r, err := invoke.DelegateAdd(context.TODO(), "ptp", []byte(ptpConf), nil)
Expect(r.Print()).To(Succeed())
return err
})
Expect(err).NotTo(HaveOccurred())
containerWithTbfResult, err := current.GetResult(containerWithTbfRes)
Expect(err).NotTo(HaveOccurred())
tbfPluginConf := &PluginConf{}
err = json.Unmarshal([]byte(ptpConf), &tbfPluginConf)
Expect(err).NotTo(HaveOccurred())
tbfPluginConf.RuntimeConfig.Bandwidth = &BandwidthEntry{
IngressBurst: burstInBits,
IngressRate: rateInBits,
EgressBurst: burstInBits,
EgressRate: rateInBits,
}
tbfPluginConf.Type = "bandwidth"
cniVersion := "0.4.0"
_, newConfBytes, err := buildOneConfig("myBWnet", cniVersion, tbfPluginConf, containerWithTbfResult)
Expect(err).NotTo(HaveOccurred())
args := &skel.CmdArgs{
ContainerID: "dummy3",
Netns: containerWithTbfNS.Path(),
IfName: containerWithTbfIFName,
StdinData: newConfBytes,
}
result, out, err := testutils.CmdAdd(containerWithTbfNS.Path(), args.ContainerID, "", newConfBytes, func() error { return cmdAdd(args) })
Expect(err).NotTo(HaveOccurred(), string(out))
// Do CNI Check
checkConf := &PluginConf{}
err = json.Unmarshal([]byte(ptpConf), &checkConf)
Expect(err).NotTo(HaveOccurred())
checkConf.RuntimeConfig.Bandwidth = &BandwidthEntry{
IngressBurst: burstInBits,
IngressRate: rateInBits,
EgressBurst: burstInBits,
EgressRate: rateInBits,
}
checkConf.Type = "bandwidth"
_, newCheckBytes, err := buildOneConfig("myBWnet", cniVersion, checkConf, result)
Expect(err).NotTo(HaveOccurred())
args = &skel.CmdArgs{
ContainerID: "dummy3",
Netns: containerWithTbfNS.Path(),
IfName: containerWithTbfIFName,
StdinData: newCheckBytes,
}
err = testutils.CmdCheck(containerWithTbfNS.Path(), args.ContainerID, "", newCheckBytes, func() error { return cmdCheck(args) })
Expect(err).NotTo(HaveOccurred())
return nil
})).To(Succeed())
By("starting a tcp server on both containers")
portServerWithTbf, echoServerWithTbf, err = startEchoServerInNamespace(containerWithTbfNS)
Expect(err).NotTo(HaveOccurred())
portServerWithoutTbf, echoServerWithoutTbf, err = startEchoServerInNamespace(containerWithoutTbfNS)
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
containerWithTbfNS.Close()
containerWithoutTbfNS.Close()
if echoServerWithoutTbf != nil {
echoServerWithoutTbf.Kill()
}
if echoServerWithTbf != nil {
echoServerWithTbf.Kill()
}
})
Measure("limits ingress traffic on veth device", func(b Benchmarker) {
var runtimeWithLimit time.Duration
var runtimeWithoutLimit time.Duration
By("gather timing statistics about both containers")
By("sending tcp traffic to the container that has traffic shaped", func() {
runtimeWithLimit = b.Time("with tbf", func() {
result, err := current.GetResult(containerWithTbfRes)
Expect(err).NotTo(HaveOccurred())
makeTcpClientInNS(hostNs.Path(), result.IPs[0].Address.IP.String(), portServerWithTbf, packetInBytes)
})
})
By("sending tcp traffic to the container that does not have traffic shaped", func() {
runtimeWithoutLimit = b.Time("without tbf", func() {
result, err := current.GetResult(containerWithoutTbfRes)
Expect(err).NotTo(HaveOccurred())
makeTcpClientInNS(hostNs.Path(), result.IPs[0].Address.IP.String(), portServerWithoutTbf, packetInBytes)
})
})
Expect(runtimeWithLimit).To(BeNumerically(">", runtimeWithoutLimit+1000*time.Millisecond))
}, 1)
})
})

164
plugins/meta/bandwidth/main.go
View File

@ -50,10 +50,6 @@ type PluginConf struct {
Bandwidth *BandwidthEntry `json:"bandwidth,omitempty"`
} `json:"runtimeConfig,omitempty"`
// RuntimeConfig *struct{} `json:"runtimeConfig"`
RawPrevResult *map[string]interface{} `json:"prevResult"`
PrevResult *current.Result `json:"-"`
*BandwidthEntry
}
@ -65,21 +61,6 @@ func parseConfig(stdin []byte) (*PluginConf, error) {
return nil, fmt.Errorf("failed to parse network configuration: %v", err)
}
if conf.RawPrevResult != nil {
resultBytes, err := json.Marshal(conf.RawPrevResult)
if err != nil {
return nil, fmt.Errorf("could not serialize prevResult: %v", err)
}
res, err := version.NewResult(conf.CNIVersion, resultBytes)
if err != nil {
return nil, fmt.Errorf("could not parse prevResult: %v", err)
}
conf.RawPrevResult = nil
conf.PrevResult, err = current.NewResultFromResult(res)
if err != nil {
return nil, fmt.Errorf("could not convert result to current version: %v", err)
}
}
bandwidth := getBandwidth(&conf)
if bandwidth != nil {
err := validateRateAndBurst(bandwidth.IngressRate, bandwidth.IngressBurst)
@ -92,6 +73,18 @@ func parseConfig(stdin []byte) (*PluginConf, error) {
}
}
if conf.RawPrevResult != nil {
var err error
if err = version.ParsePrevResult(&conf.NetConf); err != nil {
return nil, fmt.Errorf("could not parse prevResult: %v", err)
}
_, err = current.NewResultFromResult(conf.PrevResult)
if err != nil {
return nil, fmt.Errorf("could not convert result to current version: %v", err)
}
}
return &conf, nil
}
@ -167,7 +160,11 @@ func cmdAdd(args *skel.CmdArgs) error {
return fmt.Errorf("must be called as chained plugin")
}
hostInterface, err := getHostInterface(conf.PrevResult.Interfaces)
result, err := current.NewResultFromResult(conf.PrevResult)
if err != nil {
return fmt.Errorf("could not convert result to current version: %v", err)
}
hostInterface, err := getHostInterface(result.Interfaces)
if err != nil {
return err
}
@ -200,7 +197,7 @@ func cmdAdd(args *skel.CmdArgs) error {
return err
}
conf.PrevResult.Interfaces = append(conf.PrevResult.Interfaces, &current.Interface{
result.Interfaces = append(result.Interfaces, &current.Interface{
Name: ifbDeviceName,
Mac: ifbDevice.Attrs().HardwareAddr.String(),
})
@ -210,7 +207,7 @@ func cmdAdd(args *skel.CmdArgs) error {
}
}
return types.PrintResult(conf.PrevResult, conf.CNIVersion)
return types.PrintResult(result, conf.CNIVersion)
}
func cmdDel(args *skel.CmdArgs) error {
@ -233,10 +230,125 @@ func cmdDel(args *skel.CmdArgs) error {
func main() {
// TODO: implement plugin version
skel.PluginMain(cmdAdd, cmdGet, cmdDel, version.PluginSupports("0.3.0", "0.3.1", version.Current()), "TODO")
skel.PluginMain(cmdAdd, cmdCheck, cmdDel, version.PluginSupports("0.3.0", "0.3.1", version.Current()), "TODO")
}
func cmdGet(args *skel.CmdArgs) error {
// TODO: implement
return fmt.Errorf("not implemented")
func SafeQdiscList(link netlink.Link) ([]netlink.Qdisc, error) {
qdiscs, err := netlink.QdiscList(link)
if err != nil {
return nil, err
}
result := []netlink.Qdisc{}
for _, qdisc := range qdiscs {
// filter out pfifo_fast qdiscs because
// older kernels don't return them
_, pfifo := qdisc.(*netlink.PfifoFast)
if !pfifo {
result = append(result, qdisc)
}
}
return result, nil
}
func cmdCheck(args *skel.CmdArgs) error {
bwConf, err := parseConfig(args.StdinData)
if err != nil {
return err
}
if bwConf.PrevResult == nil {
return fmt.Errorf("must be called as a chained plugin")
}
result, err := current.NewResultFromResult(bwConf.PrevResult)
if err != nil {
return fmt.Errorf("could not convert result to current version: %v", err)
}
hostInterface, err := getHostInterface(result.Interfaces)
if err != nil {
return err
}
link, err := netlink.LinkByName(hostInterface.Name)
if err != nil {
return err
}
bandwidth := getBandwidth(bwConf)
if bandwidth.IngressRate > 0 && bandwidth.IngressBurst > 0 {
rateInBytes := bandwidth.IngressRate / 8
burstInBytes := bandwidth.IngressBurst / 8
bufferInBytes := buffer(uint64(rateInBytes), uint32(burstInBytes))
latency := latencyInUsec(latencyInMillis)
limitInBytes := limit(uint64(rateInBytes), latency, uint32(burstInBytes))
qdiscs, err := SafeQdiscList(link)
if err != nil {
return err
}
if len(qdiscs) == 0 {
return fmt.Errorf("Failed to find qdisc")
}
for _, qdisc := range qdiscs {
tbf, isTbf := qdisc.(*netlink.Tbf)
if !isTbf {
break
}
if tbf.Rate != uint64(rateInBytes) {
return fmt.Errorf("Rate doesn't match")
}
if tbf.Limit != uint32(limitInBytes) {
return fmt.Errorf("Limit doesn't match")
}
if tbf.Buffer != uint32(bufferInBytes) {
return fmt.Errorf("Buffer doesn't match")
}
}
}
if bandwidth.EgressRate > 0 && bandwidth.EgressBurst > 0 {
rateInBytes := bandwidth.EgressRate / 8
burstInBytes := bandwidth.EgressBurst / 8
bufferInBytes := buffer(uint64(rateInBytes), uint32(burstInBytes))
latency := latencyInUsec(latencyInMillis)
limitInBytes := limit(uint64(rateInBytes), latency, uint32(burstInBytes))
ifbDeviceName, err := getIfbDeviceName(bwConf.Name, args.ContainerID)
if err != nil {
return err
}
ifbDevice, err := netlink.LinkByName(ifbDeviceName)
if err != nil {
return fmt.Errorf("get ifb device: %s", err)
}
qdiscs, err := SafeQdiscList(ifbDevice)
if err != nil {
return err
}
if len(qdiscs) == 0 {
return fmt.Errorf("Failed to find qdisc")
}
for _, qdisc := range qdiscs {
tbf, isTbf := qdisc.(*netlink.Tbf)
if !isTbf {
break
}
if tbf.Rate != uint64(rateInBytes) {
return fmt.Errorf("Rate doesn't match")
}
if tbf.Limit != uint32(limitInBytes) {
return fmt.Errorf("Limit doesn't match")
}
if tbf.Buffer != uint32(bufferInBytes) {
return fmt.Errorf("Buffer doesn't match")
}
}
}
return nil
}