Link implementation in bridge driver

Signed-off-by: Madhu Venugopal <madhu@docker.com>

libnetwork/drivers/bridge/bridge.go

@@ -54,12 +54,13 @@ type EndpointConfiguration struct {
 
 // ContainerConfiguration represents the user specified configuration for a container
 type ContainerConfiguration struct {
-	// TODO : Add container specific configuration when Join processing is handled
+	parentEndpoints []types.UUID
+	childEndpoints  []types.UUID
 }
 
 type bridgeEndpoint struct {
 	id          types.UUID
-	port        *sandbox.Interface
+	intf        *sandbox.Interface
 	config      *EndpointConfiguration // User specified parameters
 	portMapping []netutils.PortBinding // Operation port bindings
 }
@@ -176,6 +177,12 @@ func (d *driver) Config(option map[string]interface{}) error {
 	return nil
 }
 
+func (d *driver) getNetwork(id types.UUID) (*bridgeNetwork, error) {
+	// Just a dummy function to return the only network managed by Bridge driver.
+	// But this API makes the caller code unchanged when we move to support multiple networks.
+	return d.network, nil
+}
+
 // Create a new network using bridge plugin
 func (d *driver) CreateNetwork(id types.UUID, option map[string]interface{}) error {
 	var err error
@@ -472,7 +479,7 @@ func (d *driver) CreateEndpoint(nid, eid types.UUID, epOptions map[string]interf
 	intf.Address = ipv4Addr
 
 	// Store the interface in endpoint, this is needed for cleanup on DeleteEndpoint()
-	endpoint.port = intf
+	endpoint.intf = intf
 
 	// Generate the sandbox info to return
 	sinfo := &sandbox.Info{Interfaces: []*sandbox.Interface{intf}}
@@ -543,14 +550,14 @@ func (d *driver) DeleteEndpoint(nid, eid types.UUID) error {
 	releasePorts(ep)
 
 	// Release the v4 address allocated to this endpoint's sandbox interface
-	err = ipAllocator.ReleaseIP(n.bridge.bridgeIPv4, ep.port.Address.IP)
+	err = ipAllocator.ReleaseIP(n.bridge.bridgeIPv4, ep.intf.Address.IP)
 	if err != nil {
 		return err
 	}
 
 	// Release the v6 address allocated to this endpoint's sandbox interface
 	if config.EnableIPv6 {
-		err := ipAllocator.ReleaseIP(n.bridge.bridgeIPv6, ep.port.AddressIPv6.IP)
+		err := ipAllocator.ReleaseIP(n.bridge.bridgeIPv6, ep.intf.AddressIPv6.IP)
 		if err != nil {
 			return err
 		}
@@ -558,7 +565,7 @@ func (d *driver) DeleteEndpoint(nid, eid types.UUID) error {
 
 	// Try removal of link. Discard error: link pair might have
 	// already been deleted by sandbox delete.
-	link, err := netlink.LinkByName(ep.port.SrcName)
+	link, err := netlink.LinkByName(ep.intf.SrcName)
 	if err == nil {
 		netlink.LinkDel(link)
 	}
@@ -568,11 +575,105 @@ func (d *driver) DeleteEndpoint(nid, eid types.UUID) error {
 
 // Join method is invoked when a Sandbox is attached to an endpoint.
 func (d *driver) Join(nid, eid types.UUID, sboxKey string, options map[string]interface{}) error {
-	return nil
+	var err error
+	if !d.config.EnableICC {
+		err = d.link(nid, eid, options, true)
+	}
+	return err
 }
 
 // Leave method is invoked when a Sandbox detaches from an endpoint.
 func (d *driver) Leave(nid, eid types.UUID, options map[string]interface{}) error {
+	var err error
+	if !d.config.EnableICC {
+		err = d.link(nid, eid, options, false)
+	}
+	return err
+}
+
+func (d *driver) link(nid, eid types.UUID, options map[string]interface{}, enable bool) error {
+	network, err := d.getNetwork(nid)
+	if err != nil {
+		return err
+	}
+	endpoint, err := network.getEndpoint(eid)
+	if err != nil {
+		return err
+	}
+
+	if endpoint == nil {
+		return EndpointNotFoundError(eid)
+	}
+
+	cc, err := parseContainerOptions(options)
+	if err != nil {
+		return err
+	}
+	if cc == nil {
+		return nil
+	}
+
+	if endpoint.config != nil && endpoint.config.PortBindings != nil {
+		for _, p := range cc.parentEndpoints {
+			var parentEndpoint *bridgeEndpoint
+			parentEndpoint, err = network.getEndpoint(p)
+			if err != nil {
+				return err
+			}
+			if parentEndpoint == nil {
+				err = InvalidEndpointIDError(string(p))
+				return err
+			}
+
+			l := newLink(parentEndpoint.intf.Address.IP.String(),
+				endpoint.intf.Address.IP.String(),
+				endpoint.config.PortBindings, d.config.BridgeName)
+			if enable {
+				err = l.Enable()
+				if err != nil {
+					return err
+				}
+				defer func() {
+					if err != nil {
+						l.Disable()
+					}
+				}()
+			} else {
+				l.Disable()
+			}
+		}
+	}
+
+	for _, c := range cc.childEndpoints {
+		var childEndpoint *bridgeEndpoint
+		childEndpoint, err = network.getEndpoint(c)
+		if err != nil {
+			return err
+		}
+		if childEndpoint == nil {
+			err = InvalidEndpointIDError(string(c))
+			return err
+		}
+		if childEndpoint.config == nil || childEndpoint.config.PortBindings == nil {
+			continue
+		}
+		l := newLink(endpoint.intf.Address.IP.String(),
+			childEndpoint.intf.Address.IP.String(),
+			childEndpoint.config.PortBindings, d.config.BridgeName)
+		if enable {
+			err = l.Enable()
+			if err != nil {
+				return err
+			}
+			defer func() {
+				if err != nil {
+					l.Disable()
+				}
+			}()
+		} else {
+			l.Disable()
+		}
+	}
 	return nil
 }
 
@@ -606,11 +707,15 @@ func parseEndpointOptions(epOptions map[string]interface{}) (*EndpointConfigurat
 	return ec, nil
 }
 
-func parseContainerOptions(cOptions interface{}) (*ContainerConfiguration, error) {
+func parseContainerOptions(cOptions map[string]interface{}) (*ContainerConfiguration, error) {
 	if cOptions == nil {
 		return nil, nil
 	}
-	switch opt := cOptions.(type) {
+	genericData := cOptions[options.GenericData]
+	if genericData == nil {
+		return nil, nil
+	}
+	switch opt := genericData.(type) {
 	case options.Generic:
 		opaqueConfig, err := options.GenerateFromModel(opt, &ContainerConfiguration{})
 		if err != nil {
@@ -620,7 +725,7 @@ func parseContainerOptions(cOptions interface{}) (*ContainerConfiguration, error
 	case *ContainerConfiguration:
 		return opt, nil
 	default:
-		return nil, ErrInvalidContainerConfig
+		return nil, nil
 	}
 }
 

libnetwork/drivers/bridge/bridge_test.go

@@ -2,11 +2,15 @@ package bridge
 
 import (
 	"bytes"
+	"fmt"
 	"net"
+	"regexp"
 	"testing"
 
+	"github.com/docker/docker/pkg/iptables"
 	"github.com/docker/libnetwork/netutils"
 	"github.com/docker/libnetwork/pkg/options"
+	"github.com/docker/libnetwork/types"
 	"github.com/vishvananda/netlink"
 )
 
@@ -106,6 +110,136 @@ func TestCreateLinkWithOptions(t *testing.T) {
 	}
 }
 
+func getPortMapping() []netutils.PortBinding {
+	return []netutils.PortBinding{
+		netutils.PortBinding{Proto: netutils.TCP, Port: uint16(230), HostPort: uint16(23000)},
+		netutils.PortBinding{Proto: netutils.UDP, Port: uint16(200), HostPort: uint16(22000)},
+		netutils.PortBinding{Proto: netutils.TCP, Port: uint16(120), HostPort: uint16(12000)},
+	}
+}
+
+func TestLinkContainers(t *testing.T) {
+	defer netutils.SetupTestNetNS(t)()
+
+	_, d := New()
+
+	config := &Configuration{
+		BridgeName:     DefaultBridgeName,
+		EnableIPTables: true,
+		EnableICC:      false,
+	}
+	genericOption := make(map[string]interface{})
+	genericOption[options.GenericData] = config
+
+	if err := d.Config(genericOption); err != nil {
+		t.Fatalf("Failed to setup driver config: %v", err)
+	}
+
+	err := d.CreateNetwork("net1", nil)
+	if err != nil {
+		t.Fatalf("Failed to create bridge: %v", err)
+	}
+
+	portMappings := getPortMapping()
+	epOptions := make(map[string]interface{})
+	epOptions[options.PortMap] = portMappings
+
+	sinfo, err := d.CreateEndpoint("net1", "ep1", epOptions)
+	if err != nil {
+		t.Fatalf("Failed to create an endpoint : %s", err.Error())
+	}
+
+	addr1 := sinfo.Interfaces[0].Address
+	if addr1 == nil {
+		t.Fatalf("No Ipv4 address assigned to the endpoint:  ep1")
+	}
+
+	sinfo, err = d.CreateEndpoint("net1", "ep2", nil)
+	if err != nil {
+		t.Fatalf("Failed to create an endpoint : %s", err.Error())
+	}
+
+	addr2 := sinfo.Interfaces[0].Address
+	if addr2 == nil {
+		t.Fatalf("No Ipv4 address assigned to the endpoint:  ep2")
+	}
+
+	ce := []types.UUID{"ep1"}
+	cConfig := &ContainerConfiguration{childEndpoints: ce}
+	genericOption = make(map[string]interface{})
+	genericOption[options.GenericData] = cConfig
+
+	err = d.Join("net1", "ep2", "", genericOption)
+	if err != nil {
+		t.Fatalf("Failed to link ep1 and ep2")
+	}
+
+	out, err := iptables.Raw("-L", "DOCKER")
+	for _, pm := range portMappings {
+		regex := fmt.Sprintf("%s dpt:%d", pm.Proto.String(), pm.Port)
+		re := regexp.MustCompile(regex)
+		matches := re.FindAllString(string(out[:]), -1)
+		// There will be 2 matches : Port-Mapping and Linking table rules
+		if len(matches) < 2 {
+			t.Fatalf("IP Tables programming failed %s", string(out[:]))
+		}
+
+		regex = fmt.Sprintf("%s spt:%d", pm.Proto.String(), pm.Port)
+		matched, _ := regexp.MatchString(regex, string(out[:]))
+		if !matched {
+			t.Fatalf("IP Tables programming failed %s", string(out[:]))
+		}
+	}
+
+	err = d.Leave("net1", "ep2", genericOption)
+	if err != nil {
+		t.Fatalf("Failed to unlink ep1 and ep2")
+	}
+
+	out, err = iptables.Raw("-L", "DOCKER")
+	for _, pm := range portMappings {
+		regex := fmt.Sprintf("%s dpt:%d", pm.Proto.String(), pm.Port)
+		re := regexp.MustCompile(regex)
+		matches := re.FindAllString(string(out[:]), -1)
+		// There will be 1 match : Port-Mapping
+		if len(matches) > 1 {
+			t.Fatalf("Leave should have deleted relevant IPTables rules  %s", string(out[:]))
+		}
+
+		regex = fmt.Sprintf("%s spt:%d", pm.Proto.String(), pm.Port)
+		matched, _ := regexp.MatchString(regex, string(out[:]))
+		if matched {
+			t.Fatalf("Leave should have deleted relevant IPTables rules  %s", string(out[:]))
+		}
+	}
+
+	// Error condition test with an invalid endpoint-id "ep4"
+	ce = []types.UUID{"ep1", "ep4"}
+	cConfig = &ContainerConfiguration{childEndpoints: ce}
+	genericOption = make(map[string]interface{})
+	genericOption[options.GenericData] = cConfig
+
+	err = d.Join("net1", "ep2", "", genericOption)
+	if err != nil {
+		out, err = iptables.Raw("-L", "DOCKER")
+		for _, pm := range portMappings {
+			regex := fmt.Sprintf("%s dpt:%d", pm.Proto.String(), pm.Port)
+			re := regexp.MustCompile(regex)
+			matches := re.FindAllString(string(out[:]), -1)
+			// There must be 1 match : Port-Mapping
+			if len(matches) > 1 {
+				t.Fatalf("Error handling should rollback relevant IPTables rules  %s", string(out[:]))
+			}
+
+			regex = fmt.Sprintf("%s spt:%d", pm.Proto.String(), pm.Port)
+			matched, _ := regexp.MatchString(regex, string(out[:]))
+			if matched {
+				t.Fatalf("Error handling should rollback relevant IPTables rules  %s", string(out[:]))
+			}
+		}
+	}
+}
+
 func TestValidateConfig(t *testing.T) {
 
 	// Test mtu

libnetwork/drivers/bridge/error.go

@@ -145,6 +145,12 @@ func (name ipTableCfgError) Error() string {
 	return fmt.Sprintf("unexpected request to set IP tables for interface: %s", string(name))
 }
 
+type invalidIPTablesCfgError string
+
+func (action invalidIPTablesCfgError) Error() string {
+	return fmt.Sprintf("Invalid IPTables action '%s'", string(action))
+}
+
 // IPv4AddrRangeError is returned when a valid IP address range couldn't be found.
 type IPv4AddrRangeError string
 
@@ -188,3 +194,10 @@ type IPv6AddrNoMatchError net.IPNet
 func (ipv6 *IPv6AddrNoMatchError) Error() string {
 	return fmt.Sprintf("bridge IPv6 addresses do not match the expected bridge configuration %s", ipv6)
 }
+
+// InvalidLinkIPAddrError is returned when a link is configured to a container with an invalid ip address
+type InvalidLinkIPAddrError string
+
+func (address InvalidLinkIPAddrError) Error() string {
+	return fmt.Sprintf("Cannot link to a container with Invalid IP Address '%s'", string(address))
+}

libnetwork/drivers/bridge/link.go

@@ -0,0 +1,80 @@
+package bridge
+
+import (
+	"fmt"
+	"net"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/docker/docker/pkg/iptables"
+	"github.com/docker/libnetwork/netutils"
+)
+
+type link struct {
+	parentIP string
+	childIP  string
+	ports    []netutils.PortBinding
+	bridge   string
+}
+
+func (l *link) String() string {
+	return fmt.Sprintf("%s <-> %s [%v] on %s", l.parentIP, l.childIP, l.ports, l.bridge)
+}
+
+func newLink(parentIP, childIP string, ports []netutils.PortBinding, bridge string) *link {
+	return &link{
+		childIP:  childIP,
+		parentIP: parentIP,
+		ports:    ports,
+		bridge:   bridge,
+	}
+
+}
+
+func (l *link) Enable() error {
+	// -A == iptables append flag
+	return linkContainers("-A", l.parentIP, l.childIP, l.ports, l.bridge, false)
+}
+
+func (l *link) Disable() {
+	// -D == iptables delete flag
+	err := linkContainers("-D", l.parentIP, l.childIP, l.ports, l.bridge, true)
+	if err != nil {
+		log.Errorf("Error removing IPTables rules for a link %s due to %s", l.String(), err.Error())
+	}
+	// Return proper error once we move to use a proper iptables package
+	// that returns typed errors
+}
+
+func linkContainers(action, parentIP, childIP string, ports []netutils.PortBinding, bridge string,
+	ignoreErrors bool) error {
+	var nfAction iptables.Action
+
+	switch action {
+	case "-A":
+		nfAction = iptables.Append
+	case "-I":
+		nfAction = iptables.Insert
+	case "-D":
+		nfAction = iptables.Delete
+	default:
+		return invalidIPTablesCfgError(action)
+	}
+
+	ip1 := net.ParseIP(parentIP)
+	if ip1 == nil {
+		return InvalidLinkIPAddrError(parentIP)
+	}
+	ip2 := net.ParseIP(childIP)
+	if ip2 == nil {
+		return InvalidLinkIPAddrError(childIP)
+	}
+
+	chain := iptables.Chain{Name: "DOCKER", Bridge: bridge}
+	for _, port := range ports {
+		err := chain.Link(nfAction, ip1, ip2, int(port.Port), port.Proto.String())
+		if !ignoreErrors && err != nil {
+			return err
+		}
+	}
+	return nil
+}

libnetwork/drivers/bridge/port_mapping_test.go

@@ -1,7 +1,6 @@
 package bridge
 
 import (
-	"fmt"
 	"os"
 	"testing"
 
@@ -63,8 +62,6 @@ func TestPortMappingConfig(t *testing.T) {
 		t.Fatalf("operational port mapping data not found on bridgeEndpoint")
 	}
 
-	fmt.Printf("\nendpoint: %v\n", ep.portMapping)
-
 	err = releasePorts(ep)
 	if err != nil {
 		t.Fatalf("Failed to release mapped ports: %v", err)