diff --git a/node/node.go b/node/node.go index a372b1c255..a89110599f 100644 --- a/node/node.go +++ b/node/node.go @@ -568,6 +568,17 @@ func (n *Node) Attach() (*rpc.Client, error) { return rpc.DialInProc(n.inprocHandler), nil } +// RPCHandler returns the in-process RPC request handler. +func (n *Node) RPCHandler() (*rpc.Server, error) { + n.lock.RLock() + defer n.lock.RUnlock() + + if n.inprocHandler == nil { + return nil, ErrNodeStopped + } + return n.inprocHandler, nil +} + // Server retrieves the currently running P2P network layer. This method is meant // only to inspect fields of the currently running server, life cycle management // should be left to this Node entity. diff --git a/p2p/adapters/docker.go b/p2p/adapters/docker.go index a5e05076c0..d74dd13b87 100644 --- a/p2p/adapters/docker.go +++ b/p2p/adapters/docker.go @@ -2,52 +2,127 @@ package adapters import ( "fmt" - "net" - // "net/http" - // "time" + "io" + "io/ioutil" + "os" + "os/exec" + "path/filepath" + "runtime" + "sync" + + "github.com/docker/docker/pkg/reexec" + "github.com/ethereum/go-ethereum/node" ) -func NewRemoteNode(id *NodeId, n Network) *RemoteNode { - return &RemoteNode{ - ID: id, - Network: n, +// DockerNode is a NodeAdapter which wraps an ExecNode but exec's the current +// binary in a docker container rather than locally +type DockerNode struct { + ExecNode +} + +// NewDockerNode creates a new DockerNode, building the docker image if +// necessary +func NewDockerNode(id *NodeId, service string) (*DockerNode, error) { + if runtime.GOOS != "linux" { + return nil, fmt.Errorf("NewDockerNode can only be used on Linux as it uses the current binary (which must be a Linux binary)") } + + if _, exists := serviceFuncs[service]; !exists { + return nil, fmt.Errorf("unknown node service %q", service) + } + + // build the docker image + var err error + dockerOnce.Do(func() { + err = buildDockerImage() + }) + if err != nil { + return nil, err + } + + // generate the config + conf := node.DefaultConfig + conf.DataDir = "/data" + conf.P2P.NoDiscovery = true + conf.P2P.NAT = nil + + node := &DockerNode{ + ExecNode: ExecNode{ + ID: id, + Service: service, + Config: &conf, + }, + } + node.newCmd = node.dockerCommand + return node, nil } -// RemoteNode is the network adapter that -type RemoteNode struct { - ID *NodeId - addr net.Addr - Network +// dockerCommand returns a command which exec's the binary in a docker +// container. +// +// It uses a shell so that we can pass the _P2P_NODE_CONFIG environment +// variable to the container using the --env flag. +func (n *DockerNode) dockerCommand() *exec.Cmd { + return exec.Command( + "sh", "-c", + fmt.Sprintf( + `exec docker run --interactive --env _P2P_NODE_CONFIG="${_P2P_NODE_CONFIG}" %s p2p-node %s %s`, + dockerImage, n.Service, n.ID.String(), + ), + ) } -func Name(id []byte) string { - return fmt.Sprintf("test-%08x", id) -} +// dockerImage is the name of the docker image +const dockerImage = "p2p-node" + +// dockerOnce is used to build the docker image only once +var dockerOnce sync.Once + +// buildDockerImage builds the docker image which is used to run devp2p nodes +// using docker. +// +// It adds the current binary as "p2p-node" so that it runs execP2PNode +// when executed. +func buildDockerImage() error { + // create a directory to use as the docker build context + dir, err := ioutil.TempDir("", "p2p-docker") + if err != nil { + return err + } + defer os.RemoveAll(dir) + + // copy the current binary into the build context + bin, err := os.Open(reexec.Self()) + if err != nil { + return err + } + defer bin.Close() + dst, err := os.OpenFile(filepath.Join(dir, "self.bin"), os.O_WRONLY|os.O_CREATE, 0755) + if err != nil { + return err + } + defer dst.Close() + if _, err := io.Copy(dst, bin); err != nil { + return err + } + + // create the Dockerfile + dockerfile := []byte(` +FROM ubuntu:16.04 +RUN mkdir /data +ADD self.bin /bin/p2p-node + `) + if err := ioutil.WriteFile(filepath.Join(dir, "Dockerfile"), dockerfile, 0644); err != nil { + return err + } + + // run 'docker build' + cmd := exec.Command("docker", "build", "-t", dockerImage, dir) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Run(); err != nil { + return fmt.Errorf("error building docker image: %s", err) + } -// inject(s) sends an RPC command remotely via ssh to the particular dockernode -func (self *RemoteNode) inject(string) error { return nil } - -func (self *RemoteNode) LocalAddr() []byte { - return []byte(self.addr.String()) -} - -func (self *RemoteNode) ParseAddr(p []byte, s string) ([]byte, error) { - return p, nil -} - -func (self *RemoteNode) Disconnect(rid []byte) error { - // ssh+ipc -> drop - // assumes the remote node is running the p2p module as part of the protocol - cmd := fmt.Sprintf(`p2p.Drop("%v")`, string(rid)) - return self.inject(cmd) -} - -func (self *RemoteNode) Connect(rid []byte) error { - // ssh+ipc -> connect - // - cmd := fmt.Sprintf(`admin.addPeer("%v")`, string(rid)) - return self.inject(cmd) -} diff --git a/p2p/adapters/exec.go b/p2p/adapters/exec.go index 398a1c99e8..5da7ec9488 100644 --- a/p2p/adapters/exec.go +++ b/p2p/adapters/exec.go @@ -1,13 +1,17 @@ package adapters import ( + "context" "encoding/json" "errors" "fmt" + "io" + "net" "os" "os/exec" "os/signal" "path/filepath" + "strings" "syscall" "time" @@ -36,7 +40,12 @@ func RegisterService(name string, f serviceFunc) { } // ExecNode is a NodeAdapter which starts the node by exec'ing the current -// binary and running a registered serviceFunc +// binary and running a registered serviceFunc. +// +// Communication with the node is performed using RPC over stdin / stdout +// so that we don't need access to either the node's filesystem or TCP stack +// (so for example we can run the node in a remote Docker container and +// still communicate with it). type ExecNode struct { ID *NodeId Service string @@ -45,6 +54,8 @@ type ExecNode struct { Cmd *exec.Cmd Client *rpc.Client Info *p2p.NodeInfo + + newCmd func() *exec.Cmd } // NewExecNode creates a new ExecNode which will run the given service using a @@ -63,17 +74,18 @@ func NewExecNode(id *NodeId, service, baseDir string) (*ExecNode, error) { // generate the config conf := node.DefaultConfig conf.DataDir = filepath.Join(dir, "data") - conf.IPCPath = filepath.Join(dir, "ipc.sock") conf.P2P.ListenAddr = "127.0.0.1:0" conf.P2P.NoDiscovery = true conf.P2P.NAT = nil - return &ExecNode{ + node := &ExecNode{ ID: id, Service: service, Dir: dir, Config: &conf, - }, nil + } + node.newCmd = node.execCommand + return node, nil } // Addr returns the node's enode URL @@ -104,33 +116,26 @@ func (n *ExecNode) Start() (err error) { return fmt.Errorf("error generating node config: %s", err) } + // create a net.Pipe for RPC communication over stdin / stdout + pipe1, pipe2 := net.Pipe() + // start the node - cmd := &exec.Cmd{ - Path: reexec.Self(), - Args: []string{"p2p-node", n.Service, n.ID.String()}, - Stdout: os.Stdout, - Stderr: os.Stderr, - Env: append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", conf)), - } + cmd := n.newCmd() + cmd.Stdin = pipe1 + cmd.Stdout = pipe1 + cmd.Stderr = os.Stderr + cmd.Env = append(os.Environ(), fmt.Sprintf("_P2P_NODE_CONFIG=%s", conf)) if err := cmd.Start(); err != nil { return fmt.Errorf("error starting node: %s", err) } n.Cmd = cmd - // create the RPC client - for start := time.Now(); time.Since(start) < 10*time.Second; time.Sleep(50 * time.Millisecond) { - n.Client, err = rpc.Dial(n.Config.IPCPath) - if err == nil { - break - } - } - if n.Client == nil { - return fmt.Errorf("error creating IPC client: %s", err) - } - - // load info + // create the RPC client and load the node info + n.Client = rpc.NewClientWithConn(pipe2) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() var info p2p.NodeInfo - if err := n.Client.Call(&info, "admin_nodeInfo"); err != nil { + if err := n.Client.CallContext(ctx, &info, "admin_nodeInfo"); err != nil { return fmt.Errorf("error getting node info: %s", err) } n.Info = &info @@ -138,6 +143,16 @@ func (n *ExecNode) Start() (err error) { return nil } +// execCommand returns a command which runs the node locally by exec'ing +// the current binary but setting argv[0] to "p2p-node" so that the child +// runs execP2PNode +func (n *ExecNode) execCommand() *exec.Cmd { + return &exec.Cmd{ + Path: reexec.Self(), + Args: []string{"p2p-node", n.Service, n.ID.String()}, + } +} + // Stop stops the node by first sending SIGTERM and then SIGKILL if the node // doesn't stop within 5s func (n *ExecNode) Stop() error { @@ -221,6 +236,20 @@ func execP2PNode() { log.Crit(fmt.Sprintf("unknown node service %q", serviceName)) } + // use explicit IP address in ListenAddr so that Enode URL is usable + if strings.HasPrefix(conf.P2P.ListenAddr, ":") { + addrs, err := net.InterfaceAddrs() + if err != nil { + log.Crit("error getting IP address", "err", err) + } + for _, addr := range addrs { + if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() { + conf.P2P.ListenAddr = ip.IP.String() + conf.P2P.ListenAddr + break + } + } + } + // start the devp2p stack stack, err := node.New(&conf) if err != nil { @@ -233,6 +262,15 @@ func execP2PNode() { log.Crit("error starting node", "err", err) } + // use stdin / stdout for RPC to avoid the parent needing to access + // either the local filesystem or TCP stack (useful when running in + // Docker) + handler, err := stack.RPCHandler() + if err != nil { + log.Crit("error getting RPC server", "err", err) + } + go handler.ServeCodec(rpc.NewJSONCodec(&stdioConn{os.Stdin, os.Stdout}), rpc.OptionMethodInvocation|rpc.OptionSubscriptions) + go func() { sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) @@ -244,3 +282,14 @@ func execP2PNode() { stack.Wait() } + +// stdioConn wraps os.Stdin / os.Stdout with a nop Close method so we can +// use them to handle RPC messages +type stdioConn struct { + io.Reader + io.Writer +} + +func (r *stdioConn) Close() error { + return nil +} diff --git a/p2p/adapters/inproc.go b/p2p/adapters/inproc.go index 221b1b2328..9877607638 100644 --- a/p2p/adapters/inproc.go +++ b/p2p/adapters/inproc.go @@ -166,7 +166,7 @@ func (self *SimNode) RunProtocol(id *NodeId, rw, rrw p2p.MsgReadWriter, peer *Pe return nil } log.Trace(fmt.Sprintf("protocol starting on peer %v (connection with %v)", self.Id, id)) - p := p2p.NewPeer(id.NodeID, Name(id.Bytes()), []p2p.Cap{}) + p := p2p.NewPeer(id.NodeID, id.Label(), []p2p.Cap{}) go func() { self.network.DidConnect(self.Id, id) err := self.Run(p, rw) diff --git a/rpc/client.go b/rpc/client.go index 591986987e..55ea1783a8 100644 --- a/rpc/client.go +++ b/rpc/client.go @@ -178,6 +178,14 @@ func DialContext(ctx context.Context, rawurl string) (*Client, error) { } } +// NewClientWithConn creates an RPC client which uses the provided net.Conn. +func NewClientWithConn(conn net.Conn) *Client { + client, _ := newClient(context.Background(), func(context.Context) (net.Conn, error) { + return conn, nil + }) + return client +} + func newClient(initctx context.Context, connectFunc func(context.Context) (net.Conn, error)) (*Client, error) { conn, err := connectFunc(initctx) if err != nil { diff --git a/swarm/network/simulations/discovery/discovery_test.go b/swarm/network/simulations/discovery/discovery_test.go index 72871d2c35..2db46f84fb 100644 --- a/swarm/network/simulations/discovery/discovery_test.go +++ b/swarm/network/simulations/discovery/discovery_test.go @@ -41,6 +41,30 @@ func TestMain(m *testing.M) { os.Exit(m.Run()) } +func TestDiscoverySimulationDockerAdapter(t *testing.T) { + setup := func(net *simulations.Network, trigger chan *adapters.NodeId) { + var ids []*adapters.NodeId + + // TODO: get events from the devp2p node + time.AfterFunc(10*time.Second, func() { + for _, id := range ids { + trigger <- id + } + }) + + net.SetNaf(func(conf *simulations.NodeConfig) adapters.NodeAdapter { + node, err := adapters.NewDockerNode(conf.Id, serviceName) + if err != nil { + panic(err) + } + ids = append(ids, conf.Id) + return node + }) + } + + testDiscoverySimulation(t, setup) +} + func TestDiscoverySimulationExecAdapter(t *testing.T) { baseDir, err := ioutil.TempDir("", "swarm-test") if err != nil { @@ -52,7 +76,7 @@ func TestDiscoverySimulationExecAdapter(t *testing.T) { var ids []*adapters.NodeId // TODO: get events from the devp2p node - time.AfterFunc(5*time.Second, func() { + time.AfterFunc(10*time.Second, func() { for _, id := range ids { trigger <- id } @@ -133,7 +157,7 @@ func testDiscoverySimulation(t *testing.T, setup func(net *simulations.Network, return true, nil } - timeout := 10 * time.Second + timeout := 30 * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel()