kotovalexarian-likes-github/moby--moby
1
0
Fork 0
mirror of https://github.com/moby/moby.git synced 2022-11-09 12:21:53 -05:00
Code Releases Activity

Remote driver implementation

Browse source 
In essense, this just involves marshalling structs back and forth to a
remote process, via the plugin client. There are a couple of types
that don't JSONify well, notably `net.IPNet`, so there is some
translation to be done.

To conform to the driverapi interface, we must give the list of
endpoint interfaces to the remote process, and let it puzzle out what
it's supposed to do; including the possibility of returning an error.

The constraints on EndpointInfo are enforced by the remote driver
implementation; namely:

 * It can't be nil

 * If it's got non-empty Interfaces(), the remote process can't put
   more in

In the latter case, or if we fail to add an interface for some
(future) reason, we try to roll the endpoint creation back. Likewise
for join -- if we fail to set the fields of the JoinInfo, we roll the
join back by leaving.

Signed-off-by: Michael Bridgen <mikeb@squaremobius.net>
This commit is contained in:
Michael Bridgen 2015-05-16 17:27:21 +01:00
parent 39fe941365
commit 7934979f40
4 changed files with 707 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

180
libnetwork/drivers/remote/driver.go
View file

@ -1,7 +1,8 @@
package remote
import (
"errors"
"fmt"
"net"
log "github.com/Sirupsen/logrus"
"github.com/docker/docker/pkg/plugins"
@ -9,59 +10,202 @@ import (
"github.com/docker/libnetwork/types"
)
var errNoCallback = errors.New("No Callback handler registered with Driver")
type driver struct {
endpoint *plugins.Client
networkType string
}
// Init does the necessary work to register remote drivers
func newDriver(name string, client *plugins.Client) driverapi.Driver {
return &driver{networkType: name, endpoint: client}
}
// Init makes sure a remote driver is registered when a network driver
// plugin is activated.
func Init(dc driverapi.DriverCallback) error {
plugins.Handle(driverapi.NetworkPluginEndpointType, func(name string, client *plugins.Client) {
// TODO : Handhake with the Remote Plugin goes here
newDriver := &driver{networkType: name, endpoint: client}
if err := dc.RegisterDriver(name, newDriver); err != nil {
log.Errorf("Error registering Driver for %s due to %v", name, err)
if err := dc.RegisterDriver(name, newDriver(name, client)); err != nil {
log.Errorf("error registering driver for %s due to %v", name, err)
}
})
return nil
}
// Config is not implemented for remote drivers, since it is assumed
// to be supplied to the remote process out-of-band (e.g., as command
// line arguments).
func (d *driver) Config(option map[string]interface{}) error {
return &driverapi.ErrNotImplemented{}
}
func (d *driver) CreateNetwork(id types.UUID, option map[string]interface{}) error {
return &driverapi.ErrNotImplemented{}
func (d *driver) call(methodName string, arg interface{}, retVal maybeError) error {
method := driverapi.NetworkPluginEndpointType + "." + methodName
err := d.endpoint.Call(method, arg, retVal)
if err != nil {
return err
}
if e := retVal.getError(); e != "" {
return fmt.Errorf("remote: %s", e)
}
return nil
}
func (d *driver) CreateNetwork(id types.UUID, options map[string]interface{}) error {
create := &createNetworkRequest{
NetworkID: string(id),
Options: options,
}
return d.call("CreateNetwork", create, &createNetworkResponse{})
}
func (d *driver) DeleteNetwork(nid types.UUID) error {
return &driverapi.ErrNotImplemented{}
delete := &deleteNetworkRequest{NetworkID: string(nid)}
return d.call("DeleteNetwork", delete, &deleteNetworkResponse{})
}
func (d *driver) CreateEndpoint(nid, eid types.UUID, epInfo driverapi.EndpointInfo, epOptions map[string]interface{}) error {
return &driverapi.ErrNotImplemented{}
if epInfo == nil {
return fmt.Errorf("must not be called with nil EndpointInfo")
}
reqIfaces := make([]*endpointInterface, len(epInfo.Interfaces()))
for i, iface := range epInfo.Interfaces() {
addr4 := iface.Address()
addr6 := iface.AddressIPv6()
reqIfaces[i] = &endpointInterface{
ID: iface.ID(),
Address: addr4.String(),
AddressIPv6: addr6.String(),
MacAddress: iface.MacAddress().String(),
}
}
create := &createEndpointRequest{
NetworkID: string(nid),
EndpointID: string(eid),
Interfaces: reqIfaces,
Options: epOptions,
}
var res createEndpointResponse
if err := d.call("CreateEndpoint", create, &res); err != nil {
return err
}
ifaces, err := res.parseInterfaces()
if err != nil {
return err
}
if len(reqIfaces) > 0 && len(ifaces) > 0 {
// We're not supposed to add interfaces if there already are
// some. Attempt to roll back
return errorWithRollback("driver attempted to add more interfaces", d.DeleteEndpoint(nid, eid))
}
for _, iface := range ifaces {
var addr4, addr6 net.IPNet
if iface.Address != nil {
addr4 = *(iface.Address)
}
if iface.AddressIPv6 != nil {
addr6 = *(iface.AddressIPv6)
}
if err := epInfo.AddInterface(iface.ID, iface.MacAddress, addr4, addr6); err != nil {
return errorWithRollback(fmt.Sprintf("failed to AddInterface %v: %s", iface, err), d.DeleteEndpoint(nid, eid))
}
}
return nil
}
func errorWithRollback(msg string, err error) error {
rollback := "rolled back"
if err != nil {
rollback = "failed to roll back: " + err.Error()
}
return fmt.Errorf("%s; %s", msg, rollback)
}
func (d *driver) DeleteEndpoint(nid, eid types.UUID) error {
return &driverapi.ErrNotImplemented{}
delete := &deleteEndpointRequest{
NetworkID: string(nid),
EndpointID: string(eid),
}
return d.call("DeleteEndpoint", delete, &deleteEndpointResponse{})
}
func (d *driver) EndpointOperInfo(nid, eid types.UUID) (map[string]interface{}, error) {
return nil, &driverapi.ErrNotImplemented{}
info := &endpointInfoRequest{
NetworkID: string(nid),
EndpointID: string(eid),
}
var res endpointInfoResponse
if err := d.call("EndpointOperInfo", info, &res); err != nil {
return nil, err
}
return res.Value, nil
}
// Join method is invoked when a Sandbox is attached to an endpoint.
func (d *driver) Join(nid, eid types.UUID, sboxKey string, jinfo driverapi.JoinInfo, options map[string]interface{}) error {
return &driverapi.ErrNotImplemented{}
join := &joinRequest{
NetworkID: string(nid),
EndpointID: string(eid),
SandboxKey: sboxKey,
Options: options,
}
var (
res joinResponse
err error
)
if err = d.call("Join", join, &res); err != nil {
return err
}
// Expect each interface ID given by CreateEndpoint to have an
// entry at that index in the names supplied here. In other words,
// if you supply 0..n interfaces with IDs 0..n above, you should
// supply the names in the same order.
ifaceNames := res.InterfaceNames
for _, iface := range jinfo.InterfaceNames() {
i := iface.ID()
if i >= len(ifaceNames) || i < 0 {
return fmt.Errorf("no correlating interface %d in supplied interface names", i)
}
supplied := ifaceNames[i]
if err := iface.SetNames(supplied.SrcName, supplied.DstName); err != nil {
return errorWithRollback(fmt.Sprintf("failed to set interface name: %s", err), d.Leave(nid, eid))
}
}
var addr net.IP
if res.Gateway != "" {
if addr = net.ParseIP(res.Gateway); addr == nil {
return fmt.Errorf(`unable to parse Gateway "%s"`, res.Gateway)
}
if jinfo.SetGateway(addr) != nil {
return errorWithRollback(fmt.Sprintf("failed to set gateway: %v", addr), d.Leave(nid, eid))
}
}
if res.GatewayIPv6 != "" {
if addr = net.ParseIP(res.GatewayIPv6); addr == nil {
return fmt.Errorf(`unable to parse GatewayIPv6 "%s"`, res.GatewayIPv6)
}
if jinfo.SetGatewayIPv6(addr) != nil {
return errorWithRollback(fmt.Sprintf("failed to set gateway IPv6: %v", addr), d.Leave(nid, eid))
}
}
if jinfo.SetHostsPath(res.HostsPath) != nil {
return errorWithRollback(fmt.Sprintf("failed to set hosts path: %s", res.HostsPath), d.Leave(nid, eid))
}
if jinfo.SetResolvConfPath(res.ResolvConfPath) != nil {
return errorWithRollback(fmt.Sprintf("failed to set resolv.conf path: %s", res.ResolvConfPath), d.Leave(nid, eid))
}
return nil
}
// Leave method is invoked when a Sandbox detaches from an endpoint.
func (d *driver) Leave(nid, eid types.UUID) error {
return &driverapi.ErrNotImplemented{}
leave := &leaveRequest{
NetworkID: string(nid),
EndpointID: string(eid),
}
return d.call("Leave", leave, &leaveResponse{})
}
func (d *driver) Type() string {

397
libnetwork/drivers/remote/driver_test.go Normal file
View file

@ -0,0 +1,397 @@
package remote
import (
"encoding/json"
"fmt"
"net"
"net/http"
"os"
"testing"
"github.com/docker/docker/pkg/plugins"
"github.com/docker/libnetwork/driverapi"
_ "github.com/docker/libnetwork/netutils"
"github.com/docker/libnetwork/types"
)
func decodeToMap(r *http.Request) (res map[string]interface{}, err error) {
err = json.NewDecoder(r.Body).Decode(&res)
return
}
func handle(t *testing.T, mux *http.ServeMux, method string, h func(map[string]interface{}) interface{}) {
mux.HandleFunc(fmt.Sprintf("/%s.%s", driverapi.NetworkPluginEndpointType, method), func(w http.ResponseWriter, r *http.Request) {
ask, err := decodeToMap(r)
if err != nil {
t.Fatal(err)
}
answer := h(ask)
err = json.NewEncoder(w).Encode(&answer)
if err != nil {
t.Fatal(err)
}
})
}
func setupPlugin(t *testing.T, name string, mux *http.ServeMux) func() {
if err := os.MkdirAll("/usr/share/docker/plugins", 0755); err != nil {
t.Fatal(err)
}
listener, err := net.Listen("unix", fmt.Sprintf("/usr/share/docker/plugins/%s.sock", name))
if err != nil {
t.Fatal("Could not listen to the plugin socket")
}
mux.HandleFunc("/Plugin.Activate", func(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, `{"Implements": ["%s"]}`, driverapi.NetworkPluginEndpointType)
})
go http.Serve(listener, mux)
return func() {
listener.Close()
if err := os.RemoveAll("/usr/share/docker/plugins"); err != nil {
t.Fatal(err)
}
}
}
type testEndpoint struct {
t *testing.T
id int
src string
dst string
address string
addressIPv6 string
macAddress string
gateway string
gatewayIPv6 string
resolvConfPath string
hostsPath string
}
func (test *testEndpoint) Interfaces() []driverapi.InterfaceInfo {
// return an empty one so we don't trip the check for existing
// interfaces; we don't care about this after that
return []driverapi.InterfaceInfo{}
}
func (test *testEndpoint) AddInterface(ID int, mac net.HardwareAddr, ipv4 net.IPNet, ipv6 net.IPNet) error {
if ID != test.id {
test.t.Fatalf("Wrong ID passed to AddInterface: %d", ID)
}
ip4, net4, _ := net.ParseCIDR(test.address)
ip6, net6, _ := net.ParseCIDR(test.addressIPv6)
if ip4 != nil {
net4.IP = ip4
if !types.CompareIPNet(net4, &ipv4) {
test.t.Fatalf("Wrong address given %+v", ipv4)
}
}
if ip6 != nil {
net6.IP = ip6
if !types.CompareIPNet(net6, &ipv6) {
test.t.Fatalf("Wrong address (IPv6) given %+v", ipv6)
}
}
if test.macAddress != "" && mac.String() != test.macAddress {
test.t.Fatalf("Wrong MAC address given %v", mac)
}
return nil
}
func (test *testEndpoint) InterfaceNames() []driverapi.InterfaceNameInfo {
return []driverapi.InterfaceNameInfo{test}
}
func compareIPs(t *testing.T, kind string, shouldBe string, supplied net.IP) {
ip := net.ParseIP(shouldBe)
if ip == nil {
t.Fatalf(`Invalid IP to test against: "%s"`, shouldBe)
}
if !ip.Equal(supplied) {
t.Fatalf(`%s IPs are not equal: expected "%s", got %v`, kind, shouldBe, supplied)
}
}
func (test *testEndpoint) SetGateway(ipv4 net.IP) error {
compareIPs(test.t, "Gateway", test.gateway, ipv4)
return nil
}
func (test *testEndpoint) SetGatewayIPv6(ipv6 net.IP) error {
compareIPs(test.t, "GatewayIPv6", test.gatewayIPv6, ipv6)
return nil
}
func (test *testEndpoint) SetHostsPath(p string) error {
if p != test.hostsPath {
test.t.Fatalf(`Wrong HostsPath; expected "%s", got "%s"`, test.hostsPath, p)
}
return nil
}
func (test *testEndpoint) SetResolvConfPath(p string) error {
if p != test.resolvConfPath {
test.t.Fatalf(`Wrong ResolvConfPath; expected "%s", got "%s"`, test.resolvConfPath, p)
}
return nil
}
func (test *testEndpoint) SetNames(src string, dst string) error {
if test.src != src {
test.t.Fatalf(`Wrong SrcName; expected "%s", got "%s"`, test.src, src)
}
if test.dst != dst {
test.t.Fatalf(`Wrong DstName; expected "%s", got "%s"`, test.dst, dst)
}
return nil
}
func (test *testEndpoint) ID() int {
return test.id
}
func TestRemoteDriver(t *testing.T) {
var plugin = "test-net-driver"
ep := &testEndpoint{
t: t,
src: "vethsrc",
dst: "vethdst",
address: "192.168.5.7/16",
addressIPv6: "2001:DB8::5:7/48",
macAddress: "7a:56:78:34:12:da",
gateway: "192.168.0.1",
gatewayIPv6: "2001:DB8::1",
hostsPath: "/here/comes/the/host/path",
resolvConfPath: "/there/goes/the/resolv/conf",
}
mux := http.NewServeMux()
defer setupPlugin(t, plugin, mux)()
var networkID string
handle(t, mux, "CreateNetwork", func(msg map[string]interface{}) interface{} {
nid := msg["NetworkID"]
var ok bool
if networkID, ok = nid.(string); !ok {
t.Fatal("RPC did not include network ID string")
}
return map[string]interface{}{}
})
handle(t, mux, "DeleteNetwork", func(msg map[string]interface{}) interface{} {
if nid, ok := msg["NetworkID"]; !ok || nid != networkID {
t.Fatal("Network ID missing or does not match that created")
}
return map[string]interface{}{}
})
handle(t, mux, "CreateEndpoint", func(msg map[string]interface{}) interface{} {
iface := map[string]interface{}{
"ID": ep.id,
"Address": ep.address,
"AddressIPv6": ep.addressIPv6,
"MacAddress": ep.macAddress,
}
return map[string]interface{}{
"Interfaces": []interface{}{iface},
}
})
handle(t, mux, "Join", func(msg map[string]interface{}) interface{} {
options := msg["Options"].(map[string]interface{})
foo, ok := options["foo"].(string)
if !ok || foo != "fooValue" {
t.Fatalf("Did not receive expected foo string in request options: %+v", msg)
}
return map[string]interface{}{
"Gateway": ep.gateway,
"GatewayIPv6": ep.gatewayIPv6,
"HostsPath": ep.hostsPath,
"ResolvConfPath": ep.resolvConfPath,
"InterfaceNames": []map[string]interface{}{
map[string]interface{}{
"SrcName": ep.src,
"DstName": ep.dst,
},
},
}
})
handle(t, mux, "Leave", func(msg map[string]interface{}) interface{} {
return map[string]string{}
})
handle(t, mux, "DeleteEndpoint", func(msg map[string]interface{}) interface{} {
return map[string]interface{}{}
})
handle(t, mux, "EndpointOperInfo", func(msg map[string]interface{}) interface{} {
return map[string]interface{}{
"Value": map[string]string{
"Arbitrary": "key",
"Value": "pairs?",
},
}
})
p, err := plugins.Get(plugin, driverapi.NetworkPluginEndpointType)
if err != nil {
t.Fatal(err)
}
driver := newDriver(plugin, p.Client)
if driver.Type() != plugin {
t.Fatal("Driver type does not match that given")
}
netID := types.UUID("dummy-network")
err = driver.CreateNetwork(netID, map[string]interface{}{})
if err != nil {
t.Fatal(err)
}
endID := types.UUID("dummy-endpoint")
err = driver.CreateEndpoint(netID, endID, ep, map[string]interface{}{})
if err != nil {
t.Fatal(err)
}
joinOpts := map[string]interface{}{"foo": "fooValue"}
err = driver.Join(netID, endID, "sandbox-key", ep, joinOpts)
if err != nil {
t.Fatal(err)
}
if _, err = driver.EndpointOperInfo(netID, endID); err != nil {
t.Fatal(err)
}
if err = driver.Leave(netID, endID); err != nil {
t.Fatal(err)
}
if err = driver.DeleteEndpoint(netID, endID); err != nil {
t.Fatal(err)
}
if err = driver.DeleteNetwork(netID); err != nil {
t.Fatal(err)
}
}
type failEndpoint struct {
t *testing.T
}
func (f *failEndpoint) Interfaces() []*driverapi.InterfaceInfo {
f.t.Fatal("Unexpected call of Interfaces")
return nil
}
func (f *failEndpoint) AddInterface(int, net.HardwareAddr, net.IPNet, net.IPNet) error {
f.t.Fatal("Unexpected call of AddInterface")
return nil
}
func TestDriverError(t *testing.T) {
var plugin = "test-net-driver-error"
mux := http.NewServeMux()
defer setupPlugin(t, plugin, mux)()
handle(t, mux, "CreateEndpoint", func(msg map[string]interface{}) interface{} {
return map[string]interface{}{
"Err": "this should get raised as an error",
}
})
p, err := plugins.Get(plugin, driverapi.NetworkPluginEndpointType)
if err != nil {
t.Fatal(err)
}
driver := newDriver(plugin, p.Client)
if err := driver.CreateEndpoint(types.UUID("dummy"), types.UUID("dummy"), &testEndpoint{t: t}, map[string]interface{}{}); err == nil {
t.Fatalf("Expected error from driver")
}
}
func TestMissingValues(t *testing.T) {
var plugin = "test-net-driver-missing"
mux := http.NewServeMux()
defer setupPlugin(t, plugin, mux)()
ep := &testEndpoint{
t: t,
id: 0,
}
handle(t, mux, "CreateEndpoint", func(msg map[string]interface{}) interface{} {
iface := map[string]interface{}{
"ID": ep.id,
"Address": ep.address,
"AddressIPv6": ep.addressIPv6,
"MacAddress": ep.macAddress,
}
return map[string]interface{}{
"Interfaces": []interface{}{iface},
}
})
p, err := plugins.Get(plugin, driverapi.NetworkPluginEndpointType)
if err != nil {
t.Fatal(err)
}
driver := newDriver(plugin, p.Client)
if err := driver.CreateEndpoint(types.UUID("dummy"), types.UUID("dummy"), ep, map[string]interface{}{}); err != nil {
t.Fatal(err)
}
}
type rollbackEndpoint struct {
}
func (r *rollbackEndpoint) Interfaces() []driverapi.InterfaceInfo {
return []driverapi.InterfaceInfo{}
}
func (r *rollbackEndpoint) AddInterface(_ int, _ net.HardwareAddr, _ net.IPNet, _ net.IPNet) error {
return fmt.Errorf("fail this to trigger a rollback")
}
func TestRollback(t *testing.T) {
var plugin = "test-net-driver-rollback"
mux := http.NewServeMux()
defer setupPlugin(t, plugin, mux)()
rolledback := false
handle(t, mux, "CreateEndpoint", func(msg map[string]interface{}) interface{} {
iface := map[string]interface{}{
"ID": 0,
"Address": "192.168.4.5/16",
"AddressIPv6": "",
"MacAddress": "7a:12:34:56:78:90",
}
return map[string]interface{}{
"Interfaces": []interface{}{iface},
}
})
handle(t, mux, "DeleteEndpoint", func(msg map[string]interface{}) interface{} {
rolledback = true
return map[string]interface{}{}
})
p, err := plugins.Get(plugin, driverapi.NetworkPluginEndpointType)
if err != nil {
t.Fatal(err)
}
driver := newDriver(plugin, p.Client)
ep := &rollbackEndpoint{}
if err := driver.CreateEndpoint(types.UUID("dummy"), types.UUID("dummy"), ep, map[string]interface{}{}); err == nil {
t.Fatalf("Expected error from driver")
}
if !rolledback {
t.Fatalf("Expected to have had DeleteEndpoint called")
}
}

143
libnetwork/drivers/remote/messages.go Normal file
View file

@ -0,0 +1,143 @@
package remote
import "net"
type response struct {
Err string
}
type maybeError interface {
getError() string
}
func (r *response) getError() string {
return r.Err
}
type createNetworkRequest struct {
NetworkID string
Options map[string]interface{}
}
type createNetworkResponse struct {
response
}
type deleteNetworkRequest struct {
NetworkID string
}
type deleteNetworkResponse struct {
response
}
type createEndpointRequest struct {
NetworkID string
EndpointID string
Interfaces []*endpointInterface
Options map[string]interface{}
}
type endpointInterface struct {
ID int
Address string
AddressIPv6 string
MacAddress string
}
type createEndpointResponse struct {
response
Interfaces []*endpointInterface
}
func toAddr(ipAddr string) (*net.IPNet, error) {
ip, ipnet, err := net.ParseCIDR(ipAddr)
if err != nil {
return nil, err
}
ipnet.IP = ip
return ipnet, nil
}
type iface struct {
ID int
Address *net.IPNet
AddressIPv6 *net.IPNet
MacAddress net.HardwareAddr
}
func (r *createEndpointResponse) parseInterfaces() ([]*iface, error) {
var (
ifaces = make([]*iface, len(r.Interfaces))
)
for i, inIf := range r.Interfaces {
var err error
outIf := &iface{ID: inIf.ID}
if inIf.Address != "" {
if outIf.Address, err = toAddr(inIf.Address); err != nil {
return nil, err
}
}
if inIf.AddressIPv6 != "" {
if outIf.AddressIPv6, err = toAddr(inIf.AddressIPv6); err != nil {
return nil, err
}
}
if inIf.MacAddress != "" {
if outIf.MacAddress, err = net.ParseMAC(inIf.MacAddress); err != nil {
return nil, err
}
}
ifaces[i] = outIf
}
return ifaces, nil
}
type deleteEndpointRequest struct {
NetworkID string
EndpointID string
}
type deleteEndpointResponse struct {
response
}
type endpointInfoRequest struct {
NetworkID string
EndpointID string
}
type endpointInfoResponse struct {
response
Value map[string]interface{}
}
type joinRequest struct {
NetworkID string
EndpointID string
SandboxKey string
Options map[string]interface{}
}
type ifaceName struct {
SrcName string
DstName string
}
type joinResponse struct {
response
InterfaceNames []*ifaceName
Gateway string
GatewayIPv6 string
HostsPath string
ResolvConfPath string
}
type leaveRequest struct {
NetworkID string
EndpointID string
}
type leaveResponse struct {
response
}

8
libnetwork/libnetwork_test.go
View file

@ -1277,6 +1277,10 @@ func TestValidRemoteDriver(t *testing.T) {
w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")
fmt.Fprintf(w, `{"Implements": ["%s"]}`, driverapi.NetworkPluginEndpointType)
})
mux.HandleFunc(fmt.Sprintf("/%s.CreateNetwork", driverapi.NetworkPluginEndpointType), func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/vnd.docker.plugins.v1+json")
fmt.Fprintf(w, "null")
})
if err := os.MkdirAll("/usr/share/docker/plugins", 0755); err != nil {
t.Fatal(err)
@ -1299,9 +1303,7 @@ func TestValidRemoteDriver(t *testing.T) {
_, err = controller.NewNetwork("valid-network-driver", "dummy",
libnetwork.NetworkOptionGeneric(getEmptyGenericOption()))
if err != nil {
if _, ok := err.(*driverapi.ErrNotImplemented); !ok {
t.Fatal(err)
}
t.Fatal(err)
}
}