dmsg hypervisor

cmd/hypervisor/commands/root.go

@@ -1,16 +1,16 @@
 package commands
 
 import (
+	"context"
 	"fmt"
-	"net"
 	"net/http"
 	"os"
 
 	"github.com/skycoin/skycoin/src/util/logging"
-	"github.com/spf13/cobra"
-
 	"github.com/skycoin/skywire/pkg/hypervisor"
+	"github.com/skycoin/skywire/pkg/snet"
 	"github.com/skycoin/skywire/pkg/util/pathutil"
+	"github.com/spf13/cobra"
 )
 
 const configEnv = "SW_HYPERVISOR_CONFIG"
@@ -62,7 +62,27 @@ var rootCmd = &cobra.Command{
 
 		log.Infof("serving RPC on '%s'", rpcAddr)
 		go func() {
-			l, err := net.Listen("tcp", rpcAddr)
+			_, rpcPort, err := config.Interfaces.SplitRPCAddr()
+			if err != nil {
+				log.Fatalln("Failed to parse rpc port from rpc address:", err)
+			}
+
+			dmsgConf := snet.Config{
+				PubKey:     config.PK,
+				SecKey:     config.SK,
+				TpNetworks: []string{snet.DmsgType},
+
+				DmsgDiscAddr: config.DmsgDiscovery,
+				DmsgMinSrvs:  1,
+			}
+
+			network := snet.New(dmsgConf)
+
+			ctx := context.Background()
+			if err := network.Init(ctx); err != nil {
+				log.Fatalln("failed to init network: %v", err)
+			}
+			l, err := network.Listen(snet.DmsgType, rpcPort)
 			if err != nil {
 				log.Fatalln("Failed to bind tcp port:", err)
 			}

go.mod

@@ -29,4 +29,4 @@ require (
 )
 
 // Uncomment for tests with alternate branches of 'dmsg'
-//replace github.com/skycoin/dmsg => ../dmsg
+// replace github.com/skycoin/dmsg => ../dmsg

go.sum

@@ -148,6 +148,7 @@ golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3 h1:4y9KwBHBgBNwDbtu44R5o1fdOCQUEXhbk/P4A9WmJq0=
 golang.org/x/sys v0.0.0-20190801041406-cbf593c0f2f3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190627182818-9947fec5c3ab/go.mod h1:/rFqwRUd4F7ZHNgwSSTFct+R/Kf4OFW1sUzUTQQTgfc=

pkg/hypervisor/config.go

@@ -4,8 +4,10 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"time"
 
 	"github.com/skycoin/dmsg/cipher"
@@ -35,12 +37,13 @@ func (hk *Key) UnmarshalText(text []byte) error {
 
 // Config configures the hypervisor.
 type Config struct {
-	PK         cipher.PubKey   `json:"public_key"`
-	SK         cipher.SecKey   `json:"secret_key"`
-	DBPath     string          `json:"db_path"`     // Path to store database file.
-	EnableAuth bool            `json:"enable_auth"` // Whether to enable user management.
-	Cookies    CookieConfig    `json:"cookies"`     // Configures cookies (for session management).
-	Interfaces InterfaceConfig `json:"interfaces"`  // Configures exposed interfaces.
+	PK            cipher.PubKey   `json:"public_key"`
+	SK            cipher.SecKey   `json:"secret_key"`
+	DBPath        string          `json:"db_path"`        // Path to store database file.
+	EnableAuth    bool            `json:"enable_auth"`    // Whether to enable user management.
+	Cookies       CookieConfig    `json:"cookies"`        // Configures cookies (for session management).
+	Interfaces    InterfaceConfig `json:"interfaces"`     // Configures exposed interfaces.
+	DmsgDiscovery string          `json:"dmsg_discovery"` // DmsgDiscovery address for dmsg usage
 }
 
 func makeConfig() Config {
@@ -134,3 +137,18 @@ func (c *InterfaceConfig) FillDefaults() {
 	c.HTTPAddr = ":8080"
 	c.RPCAddr = ":7080"
 }
+
+// SplitRPCAddr returns host and port and whatever error results from parsing the rpc address interface
+func (c *InterfaceConfig) SplitRPCAddr() (host string, port uint16, err error) {
+	addr, err := url.Parse(c.RPCAddr)
+	if err != nil {
+		return
+	}
+
+	uint64port, err := strconv.ParseUint(addr.Port(), 10, 16)
+	if err != nil {
+		return
+	}
+
+	return addr.Host, uint16(uint64port), nil
+}

pkg/hypervisor/hypervisor.go

@@ -6,7 +6,6 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
-	"net"
 	"net/http"
 	"net/rpc"
 	"strconv"
@@ -16,9 +15,10 @@ import (
 	"github.com/go-chi/chi"
 	"github.com/go-chi/chi/middleware"
 	"github.com/google/uuid"
+	"github.com/skycoin/dmsg"
 	"github.com/skycoin/dmsg/cipher"
-	"github.com/skycoin/dmsg/noise"
 	"github.com/skycoin/skycoin/src/util/logging"
+	"github.com/skycoin/skywire/pkg/snet"
 
 	"github.com/skycoin/skywire/pkg/app"
 	"github.com/skycoin/skywire/pkg/httputil"
@@ -32,7 +32,7 @@ var (
 )
 
 type appNodeConn struct {
-	Addr   *noise.Addr
+	Addr   dmsg.Addr
 	Client visor.RPCClient
 }
 
@@ -61,13 +61,14 @@ func NewNode(config Config) (*Node, error) {
 }
 
 // ServeRPC serves RPC of a Node.
-func (m *Node) ServeRPC(lis net.Listener) error {
+func (m *Node) ServeRPC(lis *snet.Listener) error {
+
 	for {
-		conn, err := noise.WrapListener(lis, m.c.PK, m.c.SK, false, noise.HandshakeXK).Accept()
+		conn, err := lis.AcceptConn()
 		if err != nil {
 			return err
 		}
-		addr := conn.RemoteAddr().(*noise.Addr)
+		addr := conn.RemoteAddr().(dmsg.Addr)
 		m.mu.Lock()
 		m.nodes[addr.PK] = appNodeConn{
 			Addr:   addr,
@@ -100,9 +101,9 @@ func (m *Node) AddMockData(config MockConfig) error {
 		}
 		m.mu.Lock()
 		m.nodes[pk] = appNodeConn{
-			Addr: &noise.Addr{
+			Addr: dmsg.Addr{
 				PK:   pk,
-				Addr: mockAddr(fmt.Sprintf("0.0.0.0:%d", i)),
+				Port: uint16(i),
 			},
 			Client: client,
 		}
@@ -246,7 +247,7 @@ func (m *Node) getNodes() http.HandlerFunc {
 				summary = &visor.Summary{PubKey: pk}
 			}
 			summaries = append(summaries, summaryResp{
-				TCPAddr: c.Addr.Addr.String(),
+				TCPAddr: c.Addr.String(),
 				Summary: summary,
 			})
 		}
@@ -264,7 +265,7 @@ func (m *Node) getNode() http.HandlerFunc {
 			return
 		}
 		httputil.WriteJSON(w, r, http.StatusOK, summaryResp{
-			TCPAddr: ctx.Addr.Addr.String(),
+			TCPAddr: ctx.Addr.String(),
 			Summary: summary,
 		})
 	})
@@ -573,7 +574,7 @@ func (m *Node) getLoops() http.HandlerFunc {
 	<<< Helper functions >>>
 */
 
-func (m *Node) client(pk cipher.PubKey) (*noise.Addr, visor.RPCClient, bool) {
+func (m *Node) client(pk cipher.PubKey) (dmsg.Addr, visor.RPCClient, bool) {
 	m.mu.RLock()
 	conn, ok := m.nodes[pk]
 	m.mu.RUnlock()
@@ -583,7 +584,7 @@ func (m *Node) client(pk cipher.PubKey) (*noise.Addr, visor.RPCClient, bool) {
 type httpCtx struct {
 	// Node
 	PK   cipher.PubKey
-	Addr *noise.Addr
+	Addr dmsg.Addr
 	RPC  visor.RPCClient
 
 	// App

pkg/visor/config.go

@@ -168,7 +168,7 @@ func ensureDir(path string) (string, error) {
 // HypervisorConfig represents hypervisor configuration.
 type HypervisorConfig struct {
 	PubKey cipher.PubKey `json:"public_key"`
-	Addr   string        `json:"address"`
+	Port   uint16        `json:"port"`
 }
 
 // DmsgConfig represents dmsg configuration.

pkg/visor/visor.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/skycoin/dmsg"
 	"github.com/skycoin/dmsg/cipher"
-	"github.com/skycoin/dmsg/noise"
 	"github.com/skycoin/skycoin/src/util/logging"
 
 	"github.com/skycoin/skywire/pkg/app"
@@ -109,7 +108,6 @@ type Node struct {
 	pidMu sync.Mutex
 
 	rpcListener net.Listener
-	rpcDialers  []*noise.RPCClientDialer
 }
 
 // NewNode constructs new Node.
@@ -207,15 +205,6 @@ func NewNode(config *Config, masterLogger *logging.MasterLogger) (*Node, error)
 		}
 		node.rpcListener = l
 	}
-	node.rpcDialers = make([]*noise.RPCClientDialer, len(config.Hypervisors))
-	for i, entry := range config.Hypervisors {
-		node.rpcDialers[i] = noise.NewRPCClientDialer(entry.Addr, noise.HandshakeXK, noise.Config{
-			LocalPK:   pk,
-			LocalSK:   sk,
-			RemotePK:  entry.PubKey,
-			Initiator: true,
-		})
-	}
 
 	return node, err
 }
@@ -247,12 +236,15 @@ func (node *Node) Start() error {
 		node.logger.Info("Starting RPC interface on ", node.rpcListener.Addr())
 		go rpcSvr.Accept(node.rpcListener)
 	}
-	for _, dialer := range node.rpcDialers {
-		go func(dialer *noise.RPCClientDialer) {
-			if err := dialer.Run(rpcSvr, time.Second); err != nil {
-				node.logger.Errorf("Dialer exited with error: %v", err)
+	for _, hypervisor := range node.config.Hypervisors {
+		go func(hypervisor HypervisorConfig) {
+			conn, err := node.n.Dial(snet.DmsgType, hypervisor.PubKey, hypervisor.Port)
+			if err != nil {
+				node.logger.Errorf("Hypervisor dmsg Dial exited with error: %v", err)
+			} else {
+				rpcSvr.ServeConn(conn)
 			}
-		}(dialer)
+		}(hypervisor)
 	}
 
 	node.logger.Info("Starting packet router")
@@ -334,13 +326,6 @@ func (node *Node) Close() (err error) {
 			node.logger.Info("RPC interface stopped successfully")
 		}
 	}
-	for i, dialer := range node.rpcDialers {
-		if err = dialer.Close(); err != nil {
-			node.logger.WithError(err).Errorf("(%d) failed to stop RPC dialer", i)
-		} else {
-			node.logger.Infof("(%d) RPC dialer closed successfully", i)
-		}
-	}
 	node.startedMu.Lock()
 	for a, bind := range node.startedApps {
 		if err = node.stopApp(a, bind); err != nil {

vendor/modules.txt

@@ -66,8 +66,8 @@ github.com/sirupsen/logrus/hooks/syslog
 github.com/skycoin/dmsg/cipher
 github.com/skycoin/dmsg
 github.com/skycoin/dmsg/disc
-github.com/skycoin/dmsg/noise
 github.com/skycoin/dmsg/ioutil
+github.com/skycoin/dmsg/noise
 # github.com/skycoin/skycoin v0.26.0
 github.com/skycoin/skycoin/src/util/logging
 github.com/skycoin/skycoin/src/cipher