From 99a1cfa8ffc7712807a5dd7129300b8e16f469b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mika=C3=ABl=20Cluseau?= Date: Sun, 6 Sep 2015 22:53:20 +1100 Subject: [PATCH] hairpin support --- pkg/kubelet/dockertools/docker.go | 2 +- pkg/kubelet/dockertools/docker_test.go | 2 +- pkg/kubelet/dockertools/manager.go | 11 +++ pkg/kubelet/network/hairpin/hairpin.go | 101 ++++++++++++++++++++ pkg/kubelet/network/hairpin/hairpin_test.go | 93 ++++++++++++++++++ 5 files changed, 207 insertions(+), 2 deletions(-) create mode 100644 pkg/kubelet/network/hairpin/hairpin.go create mode 100644 pkg/kubelet/network/hairpin/hairpin_test.go diff --git a/pkg/kubelet/dockertools/docker.go b/pkg/kubelet/dockertools/docker.go index d96c201641e..e2632a4fa3e 100644 --- a/pkg/kubelet/dockertools/docker.go +++ b/pkg/kubelet/dockertools/docker.go @@ -130,7 +130,7 @@ func filterHTTPError(err error, image string) error { jerr.Code == http.StatusServiceUnavailable || jerr.Code == http.StatusGatewayTimeout) { glog.V(2).Infof("Pulling image %q failed: %v", image, err) - return fmt.Errorf("image pull failed for %s because the registry is temporarily unavailbe.", image) + return fmt.Errorf("image pull failed for %s because the registry is temporarily unavailable.", image) } else { return err } diff --git a/pkg/kubelet/dockertools/docker_test.go b/pkg/kubelet/dockertools/docker_test.go index 4d095fac87d..867cae23cc3 100644 --- a/pkg/kubelet/dockertools/docker_test.go +++ b/pkg/kubelet/dockertools/docker_test.go @@ -255,7 +255,7 @@ func TestPullWithJSONError(t *testing.T) { "Bad gateway": { "ubuntu", &jsonmessage.JSONError{Code: 502, Message: "\n\n \n \n \n

Oops, there was an error!

\n

We have been contacted of this error, feel free to check out status.docker.com\n to see if there is a bigger issue.

\n\n \n"}, - "because the registry is temporarily unavailbe", + "because the registry is temporarily unavailable", }, } for i, test := range tests { diff --git a/pkg/kubelet/dockertools/manager.go b/pkg/kubelet/dockertools/manager.go index 440d9d6a58d..ee609749102 100644 --- a/pkg/kubelet/dockertools/manager.go +++ b/pkg/kubelet/dockertools/manager.go @@ -42,6 +42,7 @@ import ( "k8s.io/kubernetes/pkg/kubelet/lifecycle" "k8s.io/kubernetes/pkg/kubelet/metrics" "k8s.io/kubernetes/pkg/kubelet/network" + "k8s.io/kubernetes/pkg/kubelet/network/hairpin" "k8s.io/kubernetes/pkg/kubelet/prober" "k8s.io/kubernetes/pkg/kubelet/qos" kubeletTypes "k8s.io/kubernetes/pkg/kubelet/types" @@ -1734,6 +1735,16 @@ func (dm *DockerManager) SyncPod(pod *api.Pod, runningPod kubecontainer.Pod, pod glog.Errorf("Failed to create pod infra container: %v; Skipping pod %q", err, podFullName) return err } + + // Setup the host interface (FIXME: move to networkPlugin when ready) + podInfraContainer, err := dm.client.InspectContainer(string(podInfraContainerID)) + if err != nil { + glog.Errorf("Failed to inspect pod infra container: %v; Skipping pod %q", err, podFullName) + return err + } + if err = hairpin.SetUpContainer(podInfraContainer.State.Pid, "eth0"); err != nil { + glog.Warningf("Hairpin setup failed for pod %q: %v", podFullName, err) + } } // Start everything diff --git a/pkg/kubelet/network/hairpin/hairpin.go b/pkg/kubelet/network/hairpin/hairpin.go new file mode 100644 index 00000000000..744dabd67a5 --- /dev/null +++ b/pkg/kubelet/network/hairpin/hairpin.go @@ -0,0 +1,101 @@ +/* +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 hairpin + +import ( + "fmt" + "io/ioutil" + "net" + osexec "os/exec" + "path" + "regexp" + "strconv" + + "github.com/golang/glog" + "k8s.io/kubernetes/pkg/util/exec" +) + +const ( + sysfsNetPath = "/sys/devices/virtual/net" + hairpinModeRelativePath = "brport/hairpin_mode" + hairpinEnable = "1" +) + +var ( + ethtoolOutputRegex = regexp.MustCompile("peer_ifindex: (\\d+)") +) + +func SetUpContainer(containerPid int, containerInterfaceName string) error { + e := exec.New() + return setUpContainerInternal(e, containerPid, containerInterfaceName) +} + +func setUpContainerInternal(e exec.Interface, containerPid int, containerInterfaceName string) error { + hostIfName, err := findPairInterfaceOfContainerInterface(e, containerPid, containerInterfaceName) + if err != nil { + glog.Infof("Unable to find pair interface, setting up all interfaces: %v", err) + return setUpAllInterfaces() + } + return setUpInterface(hostIfName) +} + +func findPairInterfaceOfContainerInterface(e exec.Interface, containerPid int, containerInterfaceName string) (string, error) { + nsenterPath, err := osexec.LookPath("nsenter") + if err != nil { + return "", err + } + ethtoolPath, err := osexec.LookPath("ethtool") + if err != nil { + return "", err + } + // Get container's interface index + output, err := e.Command(nsenterPath, "-t", fmt.Sprintf("%d", containerPid), "-n", "-F", "--", ethtoolPath, "--statistics", containerInterfaceName).CombinedOutput() + if err != nil { + return "", fmt.Errorf("Unable to query interface %s of container %d: %v", containerInterfaceName, containerPid, err) + } + // look for peer_ifindex + match := ethtoolOutputRegex.FindSubmatch(output) + if match == nil { + return "", fmt.Errorf("No peer_ifindex in interface statistics for %s of container %d", containerInterfaceName, containerPid) + } + peerIfIndex, err := strconv.Atoi(string(match[1])) + if err != nil { // seems impossible (\d+ not numeric) + return "", fmt.Errorf("peer_ifindex wasn't numeric: %s: %v", match[1], err) + } + iface, err := net.InterfaceByIndex(peerIfIndex) + if err != nil { + return "", err + } + return iface.Name, nil +} + +func setUpAllInterfaces() error { + interfaces, err := net.Interfaces() + if err != nil { + return err + } + for _, netIf := range interfaces { + setUpInterface(netIf.Name) // ignore errors + } + return nil +} + +func setUpInterface(ifName string) error { + glog.V(3).Infof("Enabling hairpin on interface %s", ifName) + hairpinModeFile := path.Join(sysfsNetPath, ifName, hairpinModeRelativePath) + return ioutil.WriteFile(hairpinModeFile, []byte(hairpinEnable), 0644) +} diff --git a/pkg/kubelet/network/hairpin/hairpin_test.go b/pkg/kubelet/network/hairpin/hairpin_test.go new file mode 100644 index 00000000000..a590f2c8d3a --- /dev/null +++ b/pkg/kubelet/network/hairpin/hairpin_test.go @@ -0,0 +1,93 @@ +/* +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 hairpin + +import ( + "errors" + "fmt" + "net" + "strings" + "testing" + + "k8s.io/kubernetes/pkg/util/exec" +) + +func TestFindPairInterfaceOfContainerInterface(t *testing.T) { + // there should be at least "lo" on any system + interfaces, _ := net.Interfaces() + validOutput := fmt.Sprintf("garbage\n peer_ifindex: %d", interfaces[0].Index) + invalidOutput := fmt.Sprintf("garbage\n unknown: %d", interfaces[0].Index) + + tests := []struct { + output string + err error + expectedName string + expectErr bool + }{ + { + output: validOutput, + expectedName: interfaces[0].Name, + }, + { + output: invalidOutput, + expectErr: true, + }, + { + output: validOutput, + err: errors.New("error"), + expectErr: true, + }, + } + for _, test := range tests { + fcmd := exec.FakeCmd{ + CombinedOutputScript: []exec.FakeCombinedOutputAction{ + func() ([]byte, error) { return []byte(test.output), test.err }, + }, + } + fexec := exec.FakeExec{ + CommandScript: []exec.FakeCommandAction{ + func(cmd string, args ...string) exec.Cmd { + return exec.InitFakeCmd(&fcmd, cmd, args...) + }, + }, + } + name, err := findPairInterfaceOfContainerInterface(&fexec, 123, "eth0") + if test.expectErr { + if err == nil { + t.Errorf("unexpected non-error") + } + } else { + if err != nil { + t.Errorf("unexpected error: %v", err) + } + } + if name != test.expectedName { + t.Errorf("unexpected name: %s (expected: %s)", name, test.expectedName) + } + } +} + +func TestSetUpInterface(t *testing.T) { + err := setUpInterface("non-existent") + if err == nil { + t.Errorf("unexpected non-error") + } + hairpinModeFile := fmt.Sprintf("%s/%s/%s", sysfsNetPath, "non-existent", hairpinModeRelativePath) + if !strings.Contains(fmt.Sprintf("%v", err), hairpinModeFile) { + t.Errorf("should have tried to open %s", hairpinModeFile) + } +}