Commit 9f3ef68e authored by Brendan Burns's avatar Brendan Burns

integrate bandwidth shaping and the kubelet.

parent de60651c
...@@ -58,6 +58,7 @@ import ( ...@@ -58,6 +58,7 @@ import (
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/bandwidth"
utilErrors "k8s.io/kubernetes/pkg/util/errors" utilErrors "k8s.io/kubernetes/pkg/util/errors"
"k8s.io/kubernetes/pkg/util/mount" "k8s.io/kubernetes/pkg/util/mount"
nodeutil "k8s.io/kubernetes/pkg/util/node" nodeutil "k8s.io/kubernetes/pkg/util/node"
...@@ -552,6 +553,9 @@ type Kubelet struct { ...@@ -552,6 +553,9 @@ type Kubelet struct {
// DNS resolver configuration file. This can be used in conjunction with // DNS resolver configuration file. This can be used in conjunction with
// clusterDomain and clusterDNS. // clusterDomain and clusterDNS.
resolverConfig string resolverConfig string
// Optionally shape the bandwidth of a pod
shaper bandwidth.BandwidthShaper
} }
// getRootDir returns the full path to the directory under which kubelet can // getRootDir returns the full path to the directory under which kubelet can
...@@ -1314,6 +1318,31 @@ func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecont ...@@ -1314,6 +1318,31 @@ func (kl *Kubelet) syncPod(pod *api.Pod, mirrorPod *api.Pod, runningPod kubecont
return err return err
} }
ingress, egress, err := extractBandwidthResources(pod)
if err != nil {
return err
}
if egress != nil || ingress != nil {
if pod.Spec.HostNetwork {
kl.recorder.Event(pod, "host network not supported", "Bandwidth shaping is not currently supported on the host network")
} else if kl.shaper != nil {
status, found := kl.statusManager.GetPodStatus(pod.UID)
if !found {
statusPtr, err := kl.containerRuntime.GetPodStatus(pod)
if err != nil {
glog.Errorf("Error getting pod for bandwidth shaping")
return err
}
status = *statusPtr
}
if len(status.PodIP) > 0 {
err = kl.shaper.ReconcileCIDR(fmt.Sprintf("%s/32", status.PodIP), egress, ingress)
}
} else {
kl.recorder.Event(pod, "nil shaper", "Pod requests bandwidth shaping, but the shaper is undefined")
}
}
if isStaticPod(pod) { if isStaticPod(pod) {
if mirrorPod != nil && !kl.podManager.IsMirrorPodOf(mirrorPod, pod) { if mirrorPod != nil && !kl.podManager.IsMirrorPodOf(mirrorPod, pod) {
// The mirror pod is semantically different from the static pod. Remove // The mirror pod is semantically different from the static pod. Remove
...@@ -1391,6 +1420,48 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*api.Pod, runningPods []*kubeco ...@@ -1391,6 +1420,48 @@ func (kl *Kubelet) cleanupOrphanedPodDirs(pods []*api.Pod, runningPods []*kubeco
return utilErrors.NewAggregate(errlist) return utilErrors.NewAggregate(errlist)
} }
func (kl *Kubelet) cleanupBandwidthLimits(allPods []*api.Pod) error {
if kl.shaper == nil {
return nil
}
currentCIDRs, err := kl.shaper.GetCIDRs()
if err != nil {
return err
}
possibleCIDRs := util.StringSet{}
for ix := range allPods {
pod := allPods[ix]
ingress, egress, err := extractBandwidthResources(pod)
if err != nil {
return err
}
if ingress == nil && egress == nil {
glog.V(8).Infof("Not a bandwidth limited container...")
continue
}
status, found := kl.statusManager.GetPodStatus(pod.UID)
if !found {
statusPtr, err := kl.containerRuntime.GetPodStatus(pod)
if err != nil {
return err
}
status = *statusPtr
}
if status.Phase == api.PodRunning {
possibleCIDRs.Insert(fmt.Sprintf("%s/32", status.PodIP))
}
}
for _, cidr := range currentCIDRs {
if !possibleCIDRs.Has(cidr) {
glog.V(2).Infof("Removing CIDR: %s (%v)", cidr, possibleCIDRs)
if err := kl.shaper.Reset(cidr); err != nil {
return err
}
}
}
return nil
}
// Compares the map of current volumes to the map of desired volumes. // Compares the map of current volumes to the map of desired volumes.
// If an active volume does not have a respective desired volume, clean it up. // If an active volume does not have a respective desired volume, clean it up.
func (kl *Kubelet) cleanupOrphanedVolumes(pods []*api.Pod, runningPods []*kubecontainer.Pod) error { func (kl *Kubelet) cleanupOrphanedVolumes(pods []*api.Pod, runningPods []*kubecontainer.Pod) error {
...@@ -1623,6 +1694,11 @@ func (kl *Kubelet) HandlePodCleanups() error { ...@@ -1623,6 +1694,11 @@ func (kl *Kubelet) HandlePodCleanups() error {
glog.Errorf("Failed to cleanup terminated pods: %v", err) glog.Errorf("Failed to cleanup terminated pods: %v", err)
} }
// Clear out any old bandwith rules
if err = kl.cleanupBandwidthLimits(allPods); err != nil {
return err
}
kl.backOff.GC() kl.backOff.GC()
return err return err
} }
...@@ -2071,7 +2147,11 @@ func (kl *Kubelet) reconcileCBR0(podCIDR string) error { ...@@ -2071,7 +2147,11 @@ func (kl *Kubelet) reconcileCBR0(podCIDR string) error {
if err := ensureCbr0(cidr); err != nil { if err := ensureCbr0(cidr); err != nil {
return err return err
} }
return nil if kl.shaper == nil {
glog.V(5).Info("Shaper is nil, creating")
kl.shaper = bandwidth.NewTCShaper("cbr0")
}
return kl.shaper.ReconcileInterface()
} }
// updateNodeStatus updates node status to master with retries. // updateNodeStatus updates node status to master with retries.
...@@ -2642,3 +2722,38 @@ func (kl *Kubelet) ListenAndServeReadOnly(address net.IP, port uint) { ...@@ -2642,3 +2722,38 @@ func (kl *Kubelet) ListenAndServeReadOnly(address net.IP, port uint) {
func (kl *Kubelet) GetRuntime() kubecontainer.Runtime { func (kl *Kubelet) GetRuntime() kubecontainer.Runtime {
return kl.containerRuntime return kl.containerRuntime
} }
var minRsrc = resource.MustParse("1k")
var maxRsrc = resource.MustParse("1P")
func validateBandwidthIsReasonable(rsrc *resource.Quantity) error {
if rsrc.Value() < minRsrc.Value() {
return fmt.Errorf("resource is unreasonably small (< 1kbit)")
}
if rsrc.Value() > maxRsrc.Value() {
return fmt.Errorf("resoruce is unreasonably large (> 1Pbit)")
}
return nil
}
func extractBandwidthResources(pod *api.Pod) (ingress, egress *resource.Quantity, err error) {
str, found := pod.Annotations["kubernetes.io/ingress-bandwidth"]
if found {
if ingress, err = resource.ParseQuantity(str); err != nil {
return nil, nil, err
}
if err := validateBandwidthIsReasonable(ingress); err != nil {
return nil, nil, err
}
}
str, found = pod.Annotations["kubernetes.io/egress-bandwidth"]
if found {
if egress, err = resource.ParseQuantity(str); err != nil {
return nil, nil, err
}
if err := validateBandwidthIsReasonable(egress); err != nil {
return nil, nil, err
}
}
return ingress, egress, nil
}
...@@ -48,6 +48,7 @@ import ( ...@@ -48,6 +48,7 @@ import (
"k8s.io/kubernetes/pkg/runtime" "k8s.io/kubernetes/pkg/runtime"
"k8s.io/kubernetes/pkg/types" "k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util" "k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/bandwidth"
"k8s.io/kubernetes/pkg/version" "k8s.io/kubernetes/pkg/version"
"k8s.io/kubernetes/pkg/volume" "k8s.io/kubernetes/pkg/volume"
_ "k8s.io/kubernetes/pkg/volume/host_path" _ "k8s.io/kubernetes/pkg/volume/host_path"
...@@ -3387,3 +3388,247 @@ func TestDoesNotDeletePodDirsIfContainerIsRunning(t *testing.T) { ...@@ -3387,3 +3388,247 @@ func TestDoesNotDeletePodDirsIfContainerIsRunning(t *testing.T) {
testKubelet.fakeRuntime.PodList = []*kubecontainer.Pod{} testKubelet.fakeRuntime.PodList = []*kubecontainer.Pod{}
syncAndVerifyPodDir(t, testKubelet, pods, []*api.Pod{apiPod}, false) syncAndVerifyPodDir(t, testKubelet, pods, []*api.Pod{apiPod}, false)
} }
func TestCleanupBandwidthLimits(t *testing.T) {
tests := []struct {
status *api.PodStatus
pods []*api.Pod
inputCIDRs []string
expectResetCIDRs []string
cacheStatus bool
expectedCalls []string
name string
}{
{
status: &api.PodStatus{
PodIP: "1.2.3.4",
Phase: api.PodRunning,
},
pods: []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "10M",
},
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
},
},
},
inputCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectResetCIDRs: []string{"2.3.4.5/32", "5.6.7.8/32"},
expectedCalls: []string{"GetPodStatus"},
name: "pod running",
},
{
status: &api.PodStatus{
PodIP: "1.2.3.4",
Phase: api.PodRunning,
},
pods: []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "10M",
},
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
},
},
},
inputCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectResetCIDRs: []string{"2.3.4.5/32", "5.6.7.8/32"},
expectedCalls: []string{},
cacheStatus: true,
name: "pod running with cache",
},
{
status: &api.PodStatus{
PodIP: "1.2.3.4",
Phase: api.PodFailed,
},
pods: []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "10M",
},
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
},
},
},
inputCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectResetCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectedCalls: []string{"GetPodStatus"},
name: "pod not running",
},
{
status: &api.PodStatus{
PodIP: "1.2.3.4",
Phase: api.PodFailed,
},
pods: []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "10M",
},
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
},
},
},
inputCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectResetCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectedCalls: []string{},
cacheStatus: true,
name: "pod not running with cache",
},
{
status: &api.PodStatus{
PodIP: "1.2.3.4",
Phase: api.PodRunning,
},
pods: []*api.Pod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
},
},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
},
},
},
inputCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
expectResetCIDRs: []string{"1.2.3.4/32", "2.3.4.5/32", "5.6.7.8/32"},
name: "no bandwidth limits",
},
}
for _, test := range tests {
shaper := &bandwidth.FakeShaper{
CIDRs: test.inputCIDRs,
}
testKube := newTestKubelet(t)
testKube.kubelet.shaper = shaper
testKube.fakeRuntime.PodStatus = *test.status
if test.cacheStatus {
for _, pod := range test.pods {
testKube.kubelet.statusManager.SetPodStatus(pod, *test.status)
}
}
err := testKube.kubelet.cleanupBandwidthLimits(test.pods)
if err != nil {
t.Errorf("unexpected error: %v (%s)", test.name)
}
if !reflect.DeepEqual(shaper.ResetCIDRs, test.expectResetCIDRs) {
t.Errorf("[%s]\nexpected: %v, saw: %v", test.name, test.expectResetCIDRs, shaper.ResetCIDRs)
}
if test.cacheStatus {
if len(testKube.fakeRuntime.CalledFunctions) != 0 {
t.Errorf("unexpected function calls: %v", testKube.fakeRuntime.CalledFunctions)
}
} else if !reflect.DeepEqual(testKube.fakeRuntime.CalledFunctions, test.expectedCalls) {
t.Errorf("[%s], expected %v, saw %v", test.name, test.expectedCalls, testKube.fakeRuntime.CalledFunctions)
}
}
}
func TestExtractBandwidthResources(t *testing.T) {
four, _ := resource.ParseQuantity("4M")
ten, _ := resource.ParseQuantity("10M")
twenty, _ := resource.ParseQuantity("20M")
tests := []struct {
pod *api.Pod
expectedIngress *resource.Quantity
expectedEgress *resource.Quantity
expectError bool
}{
{
pod: &api.Pod{},
},
{
pod: &api.Pod{
ObjectMeta: api.ObjectMeta{
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "10M",
},
},
},
expectedIngress: ten,
},
{
pod: &api.Pod{
ObjectMeta: api.ObjectMeta{
Annotations: map[string]string{
"kubernetes.io/egress-bandwidth": "10M",
},
},
},
expectedEgress: ten,
},
{
pod: &api.Pod{
ObjectMeta: api.ObjectMeta{
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "4M",
"kubernetes.io/egress-bandwidth": "20M",
},
},
},
expectedIngress: four,
expectedEgress: twenty,
},
{
pod: &api.Pod{
ObjectMeta: api.ObjectMeta{
Annotations: map[string]string{
"kubernetes.io/ingress-bandwidth": "foo",
},
},
},
expectError: true,
},
}
for _, test := range tests {
ingress, egress, err := extractBandwidthResources(test.pod)
if test.expectError {
if err == nil {
t.Errorf("unexpected non-error")
}
continue
}
if err != nil {
t.Errorf("unexpected error: %v", err)
continue
}
if !reflect.DeepEqual(ingress, test.expectedIngress) {
t.Errorf("expected: %v, saw: %v", ingress, test.expectedIngress)
}
if !reflect.DeepEqual(egress, test.expectedEgress) {
t.Errorf("expected: %v, saw: %v", egress, test.expectedEgress)
}
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
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.
*/
package bandwidth
import (
"errors"
"k8s.io/kubernetes/pkg/api/resource"
)
type FakeShaper struct {
CIDRs []string
ResetCIDRs []string
}
func (f *FakeShaper) Limit(cidr string, egress, ingress *resource.Quantity) error {
return errors.New("unimplemented")
}
func (f *FakeShaper) Reset(cidr string) error {
f.ResetCIDRs = append(f.ResetCIDRs, cidr)
return nil
}
func (f *FakeShaper) ReconcileInterface() error {
return errors.New("unimplemented")
}
func (f *FakeShaper) ReconcileCIDR(cidr string, egress, ingress *resource.Quantity) error {
return errors.New("unimplemented")
}
func (f *FakeShaper) GetCIDRs() ([]string, error) {
return f.CIDRs, nil
}
...@@ -26,7 +26,13 @@ type BandwidthShaper interface { ...@@ -26,7 +26,13 @@ type BandwidthShaper interface {
// 'ingress' bandwidth limit applies to all packets on the interface whose destination matches 'cidr' // 'ingress' bandwidth limit applies to all packets on the interface whose destination matches 'cidr'
// Limits are aggregate limits for the CIDR, not per IP address. CIDRs must be unique, but can be overlapping, traffic // Limits are aggregate limits for the CIDR, not per IP address. CIDRs must be unique, but can be overlapping, traffic
// that matches multiple CIDRs counts against all limits. // that matches multiple CIDRs counts against all limits.
Limit(cidr string, egress, ingress resource.Quantity) error Limit(cidr string, egress, ingress *resource.Quantity) error
// Remove a bandwidth limit for a particular CIDR on a particular network interface // Remove a bandwidth limit for a particular CIDR on a particular network interface
Reset(cidr string) error Reset(cidr string) error
// Reconcile the interface managed by this shaper with the state on the ground.
ReconcileInterface() error
// Reconcile a CIDR managed by this shaper with the state on the ground
ReconcileCIDR(cidr string, egress, ingress *resource.Quantity) error
// GetCIDRs returns the set of CIDRs that are being managed by this shaper
GetCIDRs() ([]string, error)
} }
...@@ -47,7 +47,6 @@ func NewTCShaper(iface string) BandwidthShaper { ...@@ -47,7 +47,6 @@ func NewTCShaper(iface string) BandwidthShaper {
e: exec.New(), e: exec.New(),
iface: iface, iface: iface,
} }
shaper.initializeInterface()
return shaper return shaper
} }
...@@ -105,15 +104,34 @@ func hexCIDR(cidr string) (string, error) { ...@@ -105,15 +104,34 @@ func hexCIDR(cidr string) (string, error) {
return hexIP + "/" + hexMask, nil return hexIP + "/" + hexMask, nil
} }
func (t *tcShaper) findCIDRClass(cidr string) (class, handle string, err error) { // Convert a CIDR from hex representation to text, opposite of the above.
func asciiCIDR(cidr string) (string, error) {
parts := strings.Split(cidr, "/")
if len(parts) != 2 {
return "", fmt.Errorf("unexpected CIDR format: %s", cidr)
}
ipData, err := hex.DecodeString(parts[0])
if err != nil {
return "", err
}
ip := net.IP(ipData)
maskData, err := hex.DecodeString(parts[1])
mask := net.IPMask(maskData)
size, _ := mask.Size()
return fmt.Sprintf("%s/%d", ip.String(), size), nil
}
func (t *tcShaper) findCIDRClass(cidr string) (class, handle string, found bool, err error) {
data, err := t.e.Command("tc", "filter", "show", "dev", t.iface).CombinedOutput() data, err := t.e.Command("tc", "filter", "show", "dev", t.iface).CombinedOutput()
if err != nil { if err != nil {
return "", "", err return "", "", false, err
} }
hex, err := hexCIDR(cidr) hex, err := hexCIDR(cidr)
if err != nil { if err != nil {
return "", "", err return "", "", false, err
} }
spec := fmt.Sprintf("match %s", hex) spec := fmt.Sprintf("match %s", hex)
...@@ -133,15 +151,15 @@ func (t *tcShaper) findCIDRClass(cidr string) (class, handle string, err error) ...@@ -133,15 +151,15 @@ func (t *tcShaper) findCIDRClass(cidr string) (class, handle string, err error)
// expected tc line: // expected tc line:
// filter parent 1: protocol ip pref 1 u32 fh 800::800 order 2048 key ht 800 bkt 0 flowid 1:1 // filter parent 1: protocol ip pref 1 u32 fh 800::800 order 2048 key ht 800 bkt 0 flowid 1:1
if len(parts) != 19 { if len(parts) != 19 {
return "", "", fmt.Errorf("unexpected output from tc: %s %d (%v)", filter, len(parts), parts) return "", "", false, fmt.Errorf("unexpected output from tc: %s %d (%v)", filter, len(parts), parts)
} }
return parts[18], parts[9], nil return parts[18], parts[9], true, nil
} }
} }
return "", "", fmt.Errorf("Failed to find cidr: %s on interface: %s", cidr, t.iface) return "", "", false, nil
} }
func makeKBitString(rsrc resource.Quantity) string { func makeKBitString(rsrc *resource.Quantity) string {
return fmt.Sprintf("%dkbit", (rsrc.Value() / 1000)) return fmt.Sprintf("%dkbit", (rsrc.Value() / 1000))
} }
...@@ -150,27 +168,91 @@ func (t *tcShaper) makeNewClass(rate string) (int, error) { ...@@ -150,27 +168,91 @@ func (t *tcShaper) makeNewClass(rate string) (int, error) {
if err != nil { if err != nil {
return -1, err return -1, err
} }
if err := t.execAndLog("tc", "class", "add", "dev", t.iface, "parent", "1:", "classid", fmt.Sprintf("1:%d", class), "htb", "rate", rate); err != nil { if err := t.execAndLog("tc", "class", "add",
"dev", t.iface,
"parent", "1:",
"classid", fmt.Sprintf("1:%d", class),
"htb", "rate", rate); err != nil {
return -1, err return -1, err
} }
return class, nil return class, nil
} }
func (t *tcShaper) Limit(cidr string, upload, download resource.Quantity) (err error) { func (t *tcShaper) Limit(cidr string, upload, download *resource.Quantity) (err error) {
var downloadClass, uploadClass int var downloadClass, uploadClass int
if download != nil {
if downloadClass, err = t.makeNewClass(makeKBitString(download)); err != nil { if downloadClass, err = t.makeNewClass(makeKBitString(download)); err != nil {
return err return err
} }
if err := t.execAndLog("tc", "filter", "add",
"dev", t.iface,
"protocol", "ip",
"parent", "1:0",
"prio", "1", "u32",
"match", "ip", "dst", cidr,
"flowid", fmt.Sprintf("1:%d", downloadClass)); err != nil {
return err
}
}
if upload != nil {
if uploadClass, err = t.makeNewClass(makeKBitString(upload)); err != nil { if uploadClass, err = t.makeNewClass(makeKBitString(upload)); err != nil {
return err return err
} }
if err := t.execAndLog("tc", "filter", "add",
"dev", t.iface,
"protocol", "ip",
"parent", "1:0",
"prio", "1", "u32",
"match", "ip", "src", cidr,
"flowid", fmt.Sprintf("1:%d", uploadClass)); err != nil {
return err
}
}
return nil
}
// tests to see if an interface exists, if it does, return true and the status line for the interface
// returns false, "", <err> if an error occurs.
func (t *tcShaper) interfaceExists() (bool, string, error) {
data, err := t.e.Command("tc", "qdisc", "show", "dev", t.iface).CombinedOutput()
if err != nil {
return false, "", err
}
value := strings.TrimSpace(string(data))
if len(value) == 0 {
return false, "", nil
}
return true, value, nil
}
if err := t.execAndLog("tc", "filter", "add", "dev", t.iface, "protocol", "ip", "parent", "1:0", "prio", "1", "u32", "match", "ip", "dst", cidr, "flowid", fmt.Sprintf("1:%d", downloadClass)); err != nil { func (t *tcShaper) ReconcileCIDR(cidr string, upload, download *resource.Quantity) error {
_, _, found, err := t.findCIDRClass(cidr)
if err != nil {
return err return err
} }
if err := t.execAndLog("tc", "filter", "add", "dev", t.iface, "protocol", "ip", "parent", "1:0", "prio", "1", "u32", "match", "ip", "src", cidr, "flowid", fmt.Sprintf("1:%d", uploadClass)); err != nil { if !found {
return t.Limit(cidr, upload, download)
}
// TODO: actually check bandwidth limits here
return nil
}
func (t *tcShaper) ReconcileInterface() error {
exists, output, err := t.interfaceExists()
if err != nil {
return err return err
} }
if !exists {
glog.V(4).Info("Didn't find bandwidth interface, creating")
return t.initializeInterface()
}
fields := strings.Split(output, " ")
if len(fields) != 12 || fields[1] != "htb" || fields[2] != "1:" {
if err := t.deleteInterface(fields[2]); err != nil {
return err
}
return t.initializeInterface()
}
return nil return nil
} }
...@@ -179,12 +261,54 @@ func (t *tcShaper) initializeInterface() error { ...@@ -179,12 +261,54 @@ func (t *tcShaper) initializeInterface() error {
} }
func (t *tcShaper) Reset(cidr string) error { func (t *tcShaper) Reset(cidr string) error {
class, handle, err := t.findCIDRClass(cidr) class, handle, found, err := t.findCIDRClass(cidr)
if err != nil { if err != nil {
return err return err
} }
if err := t.execAndLog("tc", "filter", "del", "dev", t.iface, "parent", "1:", "proto", "ip", "prio", "1", "handle", handle, "u32"); err != nil { if !found {
return fmt.Errorf("Failed to find cidr: %s on interface: %s", cidr, t.iface)
}
if err := t.execAndLog("tc", "filter", "del",
"dev", t.iface,
"parent", "1:",
"proto", "ip",
"prio", "1",
"handle", handle, "u32"); err != nil {
return err return err
} }
return t.execAndLog("tc", "class", "del", "dev", t.iface, "parent", "1:", "classid", class) return t.execAndLog("tc", "class", "del", "dev", t.iface, "parent", "1:", "classid", class)
} }
func (t *tcShaper) deleteInterface(class string) error {
return t.execAndLog("tc", "qdisc", "delete", "dev", t.iface, "root", "handle", class)
}
func (t *tcShaper) GetCIDRs() ([]string, error) {
data, err := t.e.Command("tc", "filter", "show", "dev", t.iface).CombinedOutput()
if err != nil {
return nil, err
}
result := []string{}
scanner := bufio.NewScanner(bytes.NewBuffer(data))
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if len(line) == 0 {
continue
}
if strings.Contains(line, "match") {
parts := strings.Split(line, " ")
// expected tc line:
// match <cidr> at <number>
if len(parts) != 4 {
return nil, fmt.Errorf("unexpected output: %v", parts)
}
cidr, err := asciiCIDR(parts[1])
if err != nil {
return nil, err
}
result = append(result, cidr)
}
}
return result, nil
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment