Merge pull request #2374 from weaveworks/issues/1726-plugin-arp

Perform arp configuration for plugin-connected containers

common/utils.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/j-keck/arping"
 	"github.com/vishvananda/netlink"
 	"github.com/vishvananda/netns"
 
@@ -34,6 +35,7 @@ type NetDev struct {
 }
 
 // Search the network namespace of a process for interfaces matching a predicate
+// Note that the predicate is called while the goroutine is inside the process' netns
 func FindNetDevs(processID int, match func(link netlink.Link) bool) ([]NetDev, error) {
 	var netDevs []NetDev
 
@@ -137,3 +139,20 @@ func GetBridgeNetDev(bridgeName string) ([]NetDev, error) {
 		return link.Attrs().Name == bridgeName
 	})
 }
+
+// Do post-attach configuration of all veths we have created
+func ConfigureARPforVeths(processID int, prefix string) error {
+	_, err := FindNetDevs(processID, func(link netlink.Link) bool {
+		ifName := link.Attrs().Name
+		if strings.HasPrefix(ifName, prefix) {
+			weavenet.ConfigureARPCache(ifName)
+			if addrs, err := netlink.AddrList(link, netlink.FAMILY_V4); err == nil {
+				for _, addr := range addrs {
+					arping.GratuitousArpOverIfaceByName(addr.IPNet.IP, ifName)
+				}
+			}
+		}
+		return false
+	})
+	return err
+}

plugin/net/driver.go

@@ -22,23 +22,24 @@ const (
 )
 
 type network struct {
+	isOurs            bool
 	hasMulticastRoute bool
 }
 
 type driver struct {
+	name   string
 	scope  string
 	docker *docker.Client
 	sync.RWMutex
-	endpoints map[string]struct{}
-	networks  map[string]network
+	networks map[string]network
 }
 
-func New(client *docker.Client, weave *weaveapi.Client, scope string) (skel.Driver, error) {
+func New(client *docker.Client, weave *weaveapi.Client, name, scope string) (skel.Driver, error) {
 	driver := &driver{
-		scope:     scope,
-		docker:    client,
-		endpoints: make(map[string]struct{}),
-		networks:  make(map[string]network),
+		name:     name,
+		scope:    scope,
+		docker:   client,
+		networks: make(map[string]network),
 	}
 
 	_, err := NewWatcher(client, weave, driver)
@@ -61,7 +62,7 @@ func (driver *driver) GetCapabilities() (*api.GetCapabilityResponse, error) {
 
 func (driver *driver) CreateNetwork(create *api.CreateNetworkRequest) error {
 	driver.logReq("CreateNetwork", create, create.NetworkID)
-	_, err := driver.setupNetworkInfo(create.NetworkID, stringOptions(create))
+	_, err := driver.setupNetworkInfo(create.NetworkID, true, stringOptions(create))
 	return err
 }
 
@@ -93,14 +94,10 @@ func (driver *driver) DeleteNetwork(delreq *api.DeleteNetworkRequest) error {
 
 func (driver *driver) CreateEndpoint(create *api.CreateEndpointRequest) (*api.CreateEndpointResponse, error) {
 	driver.logReq("CreateEndpoint", create, create.EndpointID)
-	endID := create.EndpointID
 
 	if create.Interface == nil {
 		return nil, driver.error("CreateEndpoint", "Not supported: creating an interface from within CreateEndpoint")
 	}
-	driver.Lock()
-	driver.endpoints[endID] = struct{}{}
-	driver.Unlock()
 	resp := &api.CreateEndpointResponse{}
 
 	driver.logRes("CreateEndpoint", resp)
@@ -109,19 +106,9 @@ func (driver *driver) CreateEndpoint(create *api.CreateEndpointRequest) (*api.Cr
 
 func (driver *driver) DeleteEndpoint(deleteReq *api.DeleteEndpointRequest) error {
 	driver.logReq("DeleteEndpoint", deleteReq, deleteReq.EndpointID)
-	driver.Lock()
-	delete(driver.endpoints, deleteReq.EndpointID)
-	driver.Unlock()
 	return nil
 }
 
-func (driver *driver) HasEndpoint(endpointID string) bool {
-	driver.Lock()
-	_, found := driver.endpoints[endpointID]
-	driver.Unlock()
-	return found
-}
-
 func (driver *driver) EndpointInfo(req *api.EndpointInfoRequest) (*api.EndpointInfoResponse, error) {
 	driver.logReq("EndpointInfo", req, req.EndpointID)
 	return &api.EndpointInfoResponse{Value: map[string]interface{}{}}, nil
@@ -168,25 +155,27 @@ func (driver *driver) findNetworkInfo(id string) (network, error) {
 	if err != nil {
 		return network, err
 	}
-	return driver.setupNetworkInfo(id, info.Options)
+	return driver.setupNetworkInfo(id, info.Driver == driver.name, info.Options)
 }
 
-func (driver *driver) setupNetworkInfo(id string, options map[string]string) (network, error) {
-	var network network
-	for key, value := range options {
-		switch key {
-		case MulticastOption:
-			if value == "" { // interpret "--opt works.weave.multicast" as "turn it on"
-				network.hasMulticastRoute = true
-			} else {
-				var err error
-				if network.hasMulticastRoute, err = strconv.ParseBool(value); err != nil {
-					return network, fmt.Errorf("unrecognized value %q for option %s", value, key)
-				}
+func (driver *driver) setupNetworkInfo(id string, isOurs bool, options map[string]string) (network, error) {
+	network := network{isOurs: isOurs}
+	if isOurs {
+		for key, value := range options {
+			switch key {
+			case MulticastOption:
+				if value == "" { // interpret "--opt works.weave.multicast" as "turn it on"
+					network.hasMulticastRoute = true
+				} else {
+					var err error
+					if network.hasMulticastRoute, err = strconv.ParseBool(value); err != nil {
+						return network, fmt.Errorf("unrecognized value %q for option %s", value, key)
+					}
 
+				}
+			default:
+				driver.warn("setupNetworkInfo", "unrecognized option: %s", key)
 			}
-		default:
-			driver.warn("setupNetworkInfo", "unrecognized option: %s", key)
 		}
 	}
 	driver.Lock()

plugin/net/watcher.go

@@ -4,7 +4,9 @@ import (
 	"fmt"
 
 	weaveapi "github.com/weaveworks/weave/api"
+	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/common/docker"
+	weavenet "github.com/weaveworks/weave/net"
 )
 
 const (
@@ -32,13 +34,21 @@ func (w *watcher) ContainerStarted(id string) {
 		w.driver.warn("ContainerStarted", "error inspecting container %s: %s", id, err)
 		return
 	}
-	// check that it's on our network, via the endpointID
+	// check that it's on our network
 	for _, net := range info.NetworkSettings.Networks {
-		if w.driver.HasEndpoint(net.EndpointID) {
+		network, err := w.driver.findNetworkInfo(net.NetworkID)
+		if err != nil {
+			w.driver.warn("ContainerStarted", "unable to find network %s info: %s", net.NetworkID, err)
+			continue
+		}
+		if network.isOurs {
 			fqdn := fmt.Sprintf("%s.%s", info.Config.Hostname, info.Config.Domainname)
 			if err := w.weave.RegisterWithDNS(id, fqdn, net.IPAddress); err != nil {
 				w.driver.warn("ContainerStarted", "unable to register %s with weaveDNS: %s", id, err)
 			}
+			if err := common.ConfigureARPforVeths(info.State.Pid, weavenet.VethName); err != nil {
+				w.driver.warn("ContainerStarted", "unable to configure interfaces: %s", err)
+			}
 		}
 	}
 }

prog/plugin/main.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path"
 	"strings"
 	"syscall"
 
@@ -121,7 +122,8 @@ func run(dockerClient *docker.Client, weave *weaveapi.Client, address, meshAddre
 }
 
 func listenAndServe(dockerClient *docker.Client, weave *weaveapi.Client, address string, endChan chan<- error, scope string, withIpam bool) (net.Listener, error) {
-	d, err := netplugin.New(dockerClient, weave, scope)
+	name := strings.TrimSuffix(path.Base(address), ".sock")
+	d, err := netplugin.New(dockerClient, weave, name, scope)
 	if err != nil {
 		return nil, err
 	}

weave

@@ -1758,6 +1758,7 @@ launch_plugin_if_not_running() {
     if ! PLUGIN_CONTAINER=$(docker run -d --name=$PLUGIN_CONTAINER_NAME \
         $(docker_run_options) \
         $RESTART_POLICY \
+        --pid=host \
         -v /run/docker/plugins:/run/docker/plugins \
         -e WEAVE_HTTP_ADDR \
         $WEAVEPLUGIN_DOCKER_ARGS $PLUGIN_IMAGE $COVERAGE_ARGS \