Merge pull request #1314 from weaveworks/1209-wait-iface-netlink

Wait for ethwe via netlink, not via polling or signals

Fixes #1209. Fixes #1258. Fixes #1300.
This commit is contained in:
Matthias Radestock committed 2015-08-28 17:44:55 +01:00
commit 6291a89665
12 files changed
+322 -188

No files matched your search

+2
View File
@@ -35,6 +35,7 @@ coverage.html
# Project specific
prog/weaver/weaver
prog/weavedns/weavedns
prog/weavehosts/weavehosts
prog/weaveproxy/weaveproxy
prog/sigproxy/sigproxy
prog/weavewait/weavewait
@@ -49,6 +50,7 @@ releases
.weaveexec.uptodate
weave.tar
prog/weaveexec/weave
prog/weaveexec/weavehosts
prog/weaveexec/weaveproxy
prog/weaveexec/weavewait
prog/weaveexec/sigproxy
+7 -4
View File
@@ -13,11 +13,12 @@ WEAVER_EXE=prog/weaver/weaver
WEAVEPROXY_EXE=prog/weaveproxy/weaveproxy
SIGPROXY_EXE=prog/sigproxy/sigproxy
WEAVEWAIT_EXE=prog/weavewait/weavewait
WEAVEHOSTS_EXE=prog/weavehosts/weavehosts
NETCHECK_EXE=prog/netcheck/netcheck
COVER_EXE=testing/cover/cover
RUNNER_EXE=testing/runner/runner
EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(NETCHECK_EXE) $(COVER_EXE) $(RUNNER_EXE)
EXES=$(WEAVER_EXE) $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEHOSTS_EXE) $(NETCHECK_EXE) $(COVER_EXE) $(RUNNER_EXE)
WEAVER_UPTODATE=.weaver.uptodate
WEAVEEXEC_UPTODATE=.weaveexec.uptodate
@@ -74,11 +75,12 @@ $(NETCHECK_EXE): prog/netcheck/netcheck.go
# Sigproxy and weavewait need separate rules as they fail the netgo check in
# the main build stanza due to not importing net package
$(SIGPROXY_EXE): prog/sigproxy/main.go
$(WEAVEWAIT_EXE): prog/weavewait/main.go
$(WEAVEWAIT_EXE): prog/weavewait/main.go net/*.go
$(WEAVEHOSTS_EXE): prog/weavehosts/weavehosts.go
$(COVER_EXE): testing/cover/cover.go
$(RUNNER_EXE): testing/runner/runner.go
$(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(COVER_EXE) $(RUNNER_EXE):
$(WEAVEWAIT_EXE) $(SIGPROXY_EXE) $(WEAVEHOSTS_EXE) $(COVER_EXE) $(RUNNER_EXE):
go get ./$(@D)
go build -o $@ ./$(@D)
@@ -86,11 +88,12 @@ $(WEAVER_UPTODATE): prog/weaver/Dockerfile $(WEAVER_EXE)
$(SUDO) docker build -t $(WEAVER_IMAGE) prog/weaver
touch $@
$(WEAVEEXEC_UPTODATE): prog/weaveexec/Dockerfile $(DOCKER_DISTRIB) weave $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(NETCHECK_EXE)
$(WEAVEEXEC_UPTODATE): prog/weaveexec/Dockerfile $(DOCKER_DISTRIB) weave $(SIGPROXY_EXE) $(WEAVEPROXY_EXE) $(WEAVEWAIT_EXE) $(WEAVEHOSTS_EXE) $(NETCHECK_EXE)
cp weave prog/weaveexec/weave
cp $(SIGPROXY_EXE) prog/weaveexec/sigproxy
cp $(WEAVEPROXY_EXE) prog/weaveexec/weaveproxy
cp $(WEAVEWAIT_EXE) prog/weaveexec/weavewait
cp $(WEAVEHOSTS_EXE) prog/weaveexec/weavehosts
cp $(NETCHECK_EXE) prog/weaveexec/netcheck
cp $(DOCKER_DISTRIB) prog/weaveexec/docker.tgz
$(SUDO) docker build -t $(WEAVEEXEC_IMAGE) prog/weaveexec
+78 -17
View File
@@ -2,30 +2,54 @@ package net
import (
"fmt"
"math"
"net"
"time"
"syscall"
"github.com/vishvananda/netlink/nl"
)
const (
sleepTime = 100 * time.Millisecond
)
// Wait for an interface to come up.
func EnsureInterface(ifaceName string) (*net.Interface, error) {
s, err := nl.Subscribe(syscall.NETLINK_ROUTE, syscall.RTNLGRP_LINK)
if err != nil {
return nil, err
}
defer s.Close()
iface, err := ensureInterface(s, ifaceName)
if err != nil {
return nil, err
}
return iface, err
}
// Wait `wait` seconds for an interface to come up. Pass zero to check once
// and return immediately, or a negative value to wait indefinitely.
func EnsureInterface(ifaceName string, wait int) (iface *net.Interface, err error) {
if iface, err = findInterface(ifaceName); err == nil || wait == 0 {
return
func ensureInterface(s *nl.NetlinkSocket, ifaceName string) (*net.Interface, error) {
if iface, err := findInterface(ifaceName); err == nil {
return iface, nil
}
i := int64(math.MaxInt64)
if wait > 0 {
i = int64(time.Duration(wait) * time.Second / sleepTime)
netlinkReceiveUntil(s, matchIfName(ifaceName))
iface, err := findInterface(ifaceName)
return iface, err
}
// Wait for an interface to come up and have a route added to the multicast subnet.
// This matches the behaviour in 'weave attach', which is the only context in which
// we expect this to be called. If you change one, change the other to match.
func EnsureInterfaceAndMcastRoute(ifaceName string) (*net.Interface, error) {
s, err := nl.Subscribe(syscall.NETLINK_ROUTE, syscall.RTNLGRP_LINK, syscall.RTNLGRP_IPV4_ROUTE)
if err != nil {
return nil, err
}
for ; err != nil && i > 0; i-- {
time.Sleep(sleepTime)
iface, err = findInterface(ifaceName)
defer s.Close()
iface, err := ensureInterface(s, ifaceName)
if err != nil {
return nil, err
}
return
dest := net.IPv4(224, 0, 0, 0)
if CheckRouteExists(ifaceName, dest) {
return iface, err
}
netlinkReceiveUntil(s, matchRoute(ifaceName, dest))
return iface, err
}
func findInterface(ifaceName string) (iface *net.Interface, err error) {
@@ -37,3 +61,40 @@ func findInterface(ifaceName string) (iface *net.Interface, err error) {
}
return
}
func netlinkReceiveUntil(s *nl.NetlinkSocket, f func(syscall.NetlinkMessage) (bool, error)) error {
for {
msgs, err := s.Receive()
if err != nil {
return err
}
for _, m := range msgs {
if done, err := f(m); done {
return err
}
}
}
}
func matchIfName(ifaceName string) func(m syscall.NetlinkMessage) (bool, error) {
return func(m syscall.NetlinkMessage) (bool, error) {
switch m.Header.Type {
case syscall.RTM_NEWLINK: // receive this type for link 'up'
ifmsg := nl.DeserializeIfInfomsg(m.Data)
attrs, err := syscall.ParseNetlinkRouteAttr(&m)
if err != nil {
return true, err
}
name := ""
for _, attr := range attrs {
if attr.Attr.Type == syscall.IFA_LABEL {
name = string(attr.Value[:len(attr.Value)-1])
}
}
if ifaceName == name && ifmsg.Flags&syscall.IFF_UP != 0 {
return true, nil
}
}
return false, nil
}
}
+67 -5
View File
@@ -3,14 +3,16 @@ package net
import (
"fmt"
"net"
"syscall"
"github.com/vishvananda/netlink"
"github.com/vishvananda/netlink/nl"
)
// A network is considered free if it does not overlap any existing
// routes on this host. This is the same approach taken by Docker.
func CheckNetworkFree(subnet *net.IPNet, ignoreIfaceNames map[string]struct{}) error {
return forEachRoute(ignoreIfaceNames, func(route netlink.Route) error {
return forEachRoute(ignoreIfaceNames, func(name string, route netlink.Route) error {
if route.Dst != nil && overlaps(route.Dst, subnet) {
return fmt.Errorf("Network %s overlaps with existing route %s on host.", subnet, route.Dst)
}
@@ -26,7 +28,7 @@ func overlaps(n1, n2 *net.IPNet) bool {
// For a specific address, we only care if it is actually *inside* an
// existing route, because weave-local traffic never hits IP routing.
func CheckAddressOverlap(addr net.IP, ignoreIfaceNames map[string]struct{}) error {
return forEachRoute(ignoreIfaceNames, func(route netlink.Route) error {
return forEachRoute(ignoreIfaceNames, func(name string, route netlink.Route) error {
if route.Dst != nil && route.Dst.Contains(addr) {
return fmt.Errorf("Address %s overlaps with existing route %s on host.", addr, route.Dst)
}
@@ -34,20 +36,80 @@ func CheckAddressOverlap(addr net.IP, ignoreIfaceNames map[string]struct{}) erro
})
}
func forEachRoute(ignoreIfaceNames map[string]struct{}, check func(netlink.Route) error) error {
func forEachRoute(ignoreIfaceNames map[string]struct{}, check func(name string, r netlink.Route) error) error {
routes, err := netlink.RouteList(nil, netlink.FAMILY_V4)
if err != nil {
return err
}
for _, route := range routes {
if link, err := netlink.LinkByIndex(route.LinkIndex); err == nil {
link, err := netlink.LinkByIndex(route.LinkIndex)
if err == nil {
if _, found := ignoreIfaceNames[link.Attrs().Name]; found {
continue
}
}
if err := check(route); err != nil {
if err := check(link.Attrs().Name, route); err != nil {
return err
}
}
return nil
}
func CheckRouteExists(ifaceName string, dest net.IP) bool {
found := false
forEachRoute(map[string]struct{}{}, func(name string, route netlink.Route) error {
if name == ifaceName && route.Dst.IP.Equal(dest) {
found = true
}
return nil
})
return found
}
func matchRoute(ifaceName string, dest net.IP) func(m syscall.NetlinkMessage) (bool, error) {
return func(m syscall.NetlinkMessage) (bool, error) {
switch m.Header.Type {
case syscall.RTM_NEWROUTE:
route, err := deserializeRoute(m.Data)
if err != nil {
return true, err
}
link, _ := netlink.LinkByIndex(route.LinkIndex)
if link.Attrs().Name == ifaceName && route.Dst.IP.Equal(dest) {
return true, nil
}
}
return false, nil
}
}
// This code proposed for addition to the vishvananda/netlink library
// deserializeRoute decodes a binary netlink message into a Route struct
func deserializeRoute(m []byte) (netlink.Route, error) {
route := netlink.Route{}
msg := nl.DeserializeRtMsg(m)
attrs, err := nl.ParseRouteAttr(m[msg.Len():])
if err != nil {
return route, err
}
route.Scope = netlink.Scope(msg.Scope)
native := nl.NativeEndian()
for _, attr := range attrs {
switch attr.Attr.Type {
case syscall.RTA_GATEWAY:
route.Gw = net.IP(attr.Value)
case syscall.RTA_PREFSRC:
route.Src = net.IP(attr.Value)
case syscall.RTA_DST:
route.Dst = &net.IPNet{
IP: attr.Value,
Mask: net.CIDRMask(int(msg.Dst_len), 8*len(attr.Value)),
}
case syscall.RTA_OIF:
routeIndex := int(native.Uint32(attr.Value[0:4]))
route.LinkIndex = routeIndex
}
}
return route, nil
}
+1 -1
View File
@@ -18,6 +18,6 @@ RUN apk add --update \
&& rm -rf /var/cache/apk/*
ADD ./weave ./sigproxy ./weaveproxy /home/weave/
ADD ./netcheck /usr/bin/
ADD ./netcheck ./weavehosts /usr/bin/
ADD ./weavewait /w/w
ADD ./docker.tgz /
+120
View File
@@ -0,0 +1,120 @@
/* weavehosts: rewrite /etc/hosts as necessary to put the weave address(es) at the top */
package main
import (
"bufio"
"bytes"
"fmt"
"io/ioutil"
"net"
"os"
"sort"
"strings"
)
func main() {
args := os.Args
if len(args) < 4 {
fmt.Fprintln(os.Stderr, "usage: weavehosts hostpath hostname cidr [cidr ...]")
os.Exit(1)
}
checkErr(updateHosts(args[1], args[2], args[3:]))
os.Exit(0)
}
func checkErr(err error) {
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
func updateHosts(path, hostname string, cidrs []string) error {
hosts, err := parseHosts(path)
if err != nil {
return err
}
// Remove existing ips pointing to our hostname
toRemove := []string{}
for ip, addrs := range hosts {
for _, addr := range addrs {
if addr == hostname {
toRemove = append(toRemove, ip)
break
}
}
}
for _, ip := range toRemove {
delete(hosts, ip)
}
// Add the weave ip(s)
for _, cidr := range cidrs {
ip, _, err := net.ParseCIDR(cidr)
checkErr(err)
ipStr := ip.String()
hosts[ipStr] = append(hosts[ipStr], hostname)
}
return writeHosts(path, hosts)
}
func parseHosts(path string) (map[string][]string, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
ips := map[string][]string{}
scanner := bufio.NewScanner(f)
for scanner.Scan() {
line := scanner.Text()
// Remove any comments
if i := strings.IndexByte(line, '#'); i != -1 {
line = line[:i]
}
fields := strings.Fields(line)
if len(fields) > 0 {
ips[fields[0]] = append(ips[fields[0]], fields[1:]...)
}
}
if scanner.Err() != nil {
return nil, scanner.Err()
}
return ips, nil
}
func writeHosts(path string, contents map[string][]string) error {
ips := []string{}
for ip := range contents {
ips = append(ips, ip)
}
sort.Strings(ips)
buf := &bytes.Buffer{}
fmt.Fprintln(buf, "# modified by weave")
for _, ip := range ips {
if addrs := contents[ip]; len(addrs) > 0 {
fmt.Fprintf(buf, "%s\t%s\n", ip, strings.Join(uniqueStrs(addrs), " "))
}
}
return ioutil.WriteFile(path, buf.Bytes(), 644)
}
func uniqueStrs(s []string) []string {
m := map[string]struct{}{}
result := []string{}
for _, str := range s {
if _, ok := m[str]; !ok {
m[str] = struct{}{}
result = append(result, str)
}
}
sort.Strings(result)
return result
}
+1 -3
View File
@@ -43,7 +43,6 @@ func main() {
routerName string
nickName string
password string
wait int
pktdebug bool
logLevel string
prof string
@@ -71,7 +70,6 @@ func main() {
mflag.StringVar(&routerName, []string{"#name", "-name"}, "", "name of router (defaults to MAC of interface)")
mflag.StringVar(&nickName, []string{"#nickname", "-nickname"}, "", "nickname of peer (defaults to hostname)")
mflag.StringVar(&password, []string{"#password", "-password"}, "", "network password")
mflag.IntVar(&wait, []string{"#wait", "-wait"}, -1, "number of seconds to wait for interface to come up (0=don't wait, -1=wait forever)")
mflag.StringVar(&logLevel, []string{"-log-level"}, "info", "logging level (debug, info, warning, error)")
mflag.BoolVar(&pktdebug, []string{"#pktdebug", "#-pktdebug", "-pkt-debug"}, false, "enable per-packet debug logging")
mflag.StringVar(&prof, []string{"#profile", "-profile"}, "", "enable profiling and write profiles to given path")
@@ -118,7 +116,7 @@ func main() {
var err error
if ifaceName != "" {
config.Iface, err = weavenet.EnsureInterface(ifaceName, wait)
config.Iface, err = weavenet.EnsureInterface(ifaceName)
if err != nil {
Log.Fatal(err)
}
+2 -15
View File
@@ -5,7 +5,6 @@ import (
"fmt"
"os"
"os/exec"
"os/signal"
"syscall"
weavenet "github.com/weaveworks/weave/net"
@@ -17,22 +16,10 @@ var (
func main() {
var (
args = os.Args[1:]
notInExec = true
args = os.Args[1:]
)
if len(args) > 0 && args[0] == "-s" {
notInExec = false
args = args[1:]
}
if notInExec {
usr2 := make(chan os.Signal)
signal.Notify(usr2, syscall.SIGUSR2)
<-usr2
}
_, err := weavenet.EnsureInterface("ethwe", -1)
_, err := weavenet.EnsureInterfaceAndMcastRoute("ethwe")
checkErr(err)
if len(args) == 0 {
+1 -2
View File
@@ -35,8 +35,7 @@ func (i *createExecInterceptor) InterceptRequest(r *http.Request) error {
Log.Infof("Leaving container %s alone because %s", container.ID, err)
} else if hasWeaveWait {
Log.Infof("Exec in container %s with WEAVE_CIDR \"%s\"", container.ID, strings.Join(cidrs, " "))
cmd := append(weaveWaitEntrypoint, "-s")
options.Cmd = append(cmd, options.Cmd...)
options.Cmd = append(weaveWaitEntrypoint, options.Cmd...)
}
if err := marshalRequestBody(r, options); err != nil {
+4 -132
View File
@@ -1,18 +1,10 @@
package proxy
import (
"bufio"
"bytes"
"errors"
"fmt"
"io/ioutil"
"net"
"net/http"
"os"
"sort"
"strings"
"github.com/fsouza/go-dockerclient"
. "github.com/weaveworks/weave/common"
)
@@ -40,6 +32,9 @@ func (i *startContainerInterceptor) InterceptResponse(r *http.Response) error {
Log.Infof("Attaching container %s with WEAVE_CIDR \"%s\" to weave network", container.ID, strings.Join(cidrs, " "))
args := []string{"attach"}
args = append(args, cidrs...)
if !i.proxy.NoRewriteHosts {
args = append(args, "--rewrite-hosts")
}
args = append(args, "--or-die", container.ID)
if _, stderr, err := callWeave(args...); err != nil {
Log.Warningf("Attaching container %s to weave network failed: %s", container.ID, string(stderr))
@@ -48,128 +43,5 @@ func (i *startContainerInterceptor) InterceptResponse(r *http.Response) error {
Log.Warningf("Attaching container %s to weave network: %s", container.ID, string(stderr))
}
if !i.proxy.NoRewriteHosts {
ips, err := weaveContainerIPs(container)
if err != nil {
return err
}
if len(ips) > 0 {
if err := updateHosts(container.HostsPath, container.Config.Hostname, ips); err != nil {
return err
}
}
}
return i.proxy.client.KillContainer(docker.KillContainerOptions{ID: container.ID, Signal: docker.SIGUSR2})
}
func weaveContainerIPs(container *docker.Container) ([]net.IP, error) {
stdout, stderr, err := callWeave("ps", container.ID)
if err != nil || len(stderr) > 0 {
return nil, errors.New(string(stderr))
}
if len(stdout) <= 0 {
return nil, nil
}
fields := strings.Fields(string(stdout))
if len(fields) <= 2 {
return nil, nil
}
var ips []net.IP
for _, cidr := range fields[2:] {
ip, _, err := net.ParseCIDR(cidr)
if err != nil {
return nil, err
}
ips = append(ips, ip)
}
return ips, nil
}
func updateHosts(path, hostname string, ips []net.IP) error {
hosts, err := parseHosts(path)
if err != nil {
return err
}
// Remove existing ips pointing to our hostname
toRemove := []string{}
for ip, addrs := range hosts {
for _, addr := range addrs {
if addr == hostname {
toRemove = append(toRemove, ip)
break
}
}
}
for _, ip := range toRemove {
delete(hosts, ip)
}
// Add the weave ip(s)
for _, ip := range ips {
ipStr := ip.String()
hosts[ipStr] = append(hosts[ipStr], hostname)
}
return writeHosts(path, hosts)
}
func parseHosts(path string) (map[string][]string, error) {
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
ips := map[string][]string{}
scanner := bufio.NewScanner(f)
for scanner.Scan() {
line := scanner.Text()
// Remove any comments
if i := strings.IndexByte(line, '#'); i != -1 {
line = line[:i]
}
fields := strings.Fields(line)
if len(fields) > 0 {
ips[fields[0]] = append(ips[fields[0]], fields[1:]...)
}
}
if scanner.Err() != nil {
return nil, scanner.Err()
}
return ips, nil
}
func writeHosts(path string, contents map[string][]string) error {
ips := []string{}
for ip := range contents {
ips = append(ips, ip)
}
sort.Strings(ips)
buf := &bytes.Buffer{}
fmt.Fprintln(buf, "# modified by weave")
for _, ip := range ips {
if addrs := contents[ip]; len(addrs) > 0 {
fmt.Fprintf(buf, "%s\t%s\n", ip, strings.Join(uniqueStrs(addrs), " "))
}
}
return ioutil.WriteFile(path, buf.Bytes(), 644)
}
func uniqueStrs(s []string) []string {
m := map[string]struct{}{}
result := []string{}
for _, str := range s {
if _, ok := m[str]; !ok {
m[str] = struct{}{}
result = append(result, str)
}
}
sort.Strings(result)
return result
return nil
}
+15
View File
@@ -0,0 +1,15 @@
#! /bin/bash
. ./config.sh
start_suite "Abuse of 'start' operation"
weave_on $HOST1 launch
proxy_start_container $HOST1 --name=c1
proxy docker_on $HOST1 create --name=c2 $SMALL_IMAGE grep "^1$" /sys/class/net/ethwe/carrier
# Now start c2 with a sneaky HostConfig
curl -X POST -H Content-Type:application/json -d '{"HostConfig": {"NetworkMode": "container:c1"}}' http://$HOST1:12375/containers/c2/start
assert "docker_on $HOST1 inspect -f '{{.State.Running}} {{.State.ExitCode}}' c2" "false 0"
end_suite
+24 -9
View File
@@ -540,14 +540,15 @@ attach() {
netnsenter ip link set $CONTAINER_IFNAME up || return 1
# Route multicast packets across the weave network.
if ! netnsenter ip route show | grep '^224\.0\.0\.0/4' >/dev/null ; then
netnsenter ip route add 224.0.0.0/4 dev $CONTAINER_IFNAME
fi
for ADDR in "$@" ; do
arp_update $CONTAINER_IFNAME $ADDR "netnsenter"
done
# Route multicast packets across the weave network.
# This must come last in 'attach'. If you change this, change weavewait to match.
if ! netnsenter ip route show | grep '^224\.0\.0\.0/4' >/dev/null ; then
netnsenter ip route add 224.0.0.0/4 dev $CONTAINER_IFNAME
fi
}
detach() {
@@ -1459,14 +1460,28 @@ case "$COMMAND" in
attach)
collect_cidr_args "$@"
shift $CIDR_ARG_COUNT
if [ "$1" = "--or-die" ] ; then
ATTACH_TYPE="_or_die"
shift
fi
while [ $# -gt 0 ]; do
case "$1" in
--or-die)
ATTACH_TYPE="_or_die"
;;
--rewrite-hosts)
REWRITE_HOSTS=1
;;
*)
break
;;
esac
shift
done
[ $# -eq 1 ] || usage
CONTAINER=$(container_id $1)
create_bridge
ipam_cidrs$ATTACH_TYPE $CONTAINER $CIDR_ARGS
if [ -n "$REWRITE_HOSTS" ] && command_exists weavehosts ; then
PATH_AND_NAME=$(docker inspect -f '{{.HostsPath}} {{.Config.Hostname}}' $CONTAINER)
weavehosts $PATH_AND_NAME $ALL_CIDRS
fi
with_container_netns$ATTACH_TYPE $CONTAINER attach $ALL_CIDRS >/dev/null
when_weave_running with_container_fqdn $CONTAINER put_dns_fqdn $ALL_CIDRS
show_addrs $ALL_CIDRS