Merge pull request #124 from Kifen/dmsg-hypervisor

Connect hypervisor and visor over dmsg

cmd/hypervisor/commands/gen-config.go

@@ -15,6 +15,7 @@ var (
 	output        string
 	replace       bool
 	configLocType = pathutil.WorkingDirLoc
+	testenv       bool
 )
 
 // nolint:gochecknoinits
@@ -24,6 +25,10 @@ func init() {
 	configLocTypeUsage := fmt.Sprintf("config generation mode. Valid values: %v", pathutil.AllConfigLocationTypes())
 
 	rootCmd.AddCommand(genConfigCmd)
+	genConfigCmd.Flags().StringVarP(&output, "output", "o", "", "path of output config file. Uses default of 'type' flag if unspecified.")
+	genConfigCmd.Flags().BoolVarP(&replace, "replace", "r", false, "whether to allow rewrite of a file that already exists.")
+	genConfigCmd.Flags().VarP(&configLocType, "type", "m", fmt.Sprintf("config generation mode. Valid values: %v", pathutil.AllConfigLocationTypes()))
+	genConfigCmd.Flags().BoolVarP(&testenv, "testing-environment", "t", false, "whether to use production or test deployment service.")
 	genConfigCmd.Flags().StringVarP(&output, "output", "o", "", outputUsage)
 	genConfigCmd.Flags().BoolVarP(&replace, "replace", "r", false, replaceUsage)
 	genConfigCmd.Flags().VarP(&configLocType, "type", "m", configLocTypeUsage)
@@ -47,11 +52,11 @@ var genConfigCmd = &cobra.Command{
 		var conf hypervisor.Config
 		switch configLocType {
 		case pathutil.WorkingDirLoc:
-			conf = hypervisor.GenerateWorkDirConfig()
+			conf = hypervisor.GenerateWorkDirConfig(testenv)
 		case pathutil.HomeLoc:
-			conf = hypervisor.GenerateHomeConfig()
+			conf = hypervisor.GenerateHomeConfig(testenv)
 		case pathutil.LocalLoc:
-			conf = hypervisor.GenerateLocalConfig()
+			conf = hypervisor.GenerateLocalConfig(testenv)
 		default:
 			log.Fatalln("invalid config type:", configLocType)
 		}

cmd/hypervisor/commands/root.go

@@ -2,9 +2,11 @@ package commands
 
 import (
 	"fmt"
-	"net"
 	"net/http"
 
+	"github.com/SkycoinProject/dmsg"
+	"github.com/SkycoinProject/dmsg/disc"
+
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 	"github.com/spf13/cobra"
 
@@ -65,10 +67,19 @@ 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)
+			}
+
+			dmsgC := dmsg.NewClient(config.PK, config.SK, disc.NewHTTP(config.DmsgDiscovery), dmsg.DefaultConfig())
+			go dmsgC.Serve()
+
+			l, err := dmsgC.Listen(rpcPort)
 			if err != nil {
 				log.Fatalln("Failed to bind tcp port:", err)
 			}
+
 			if err := m.ServeRPC(l); err != nil {
 				log.Fatalln("Failed to serve RPC:", err)
 			}

go.mod

@@ -24,6 +24,7 @@ require (
 	golang.org/x/crypto v0.0.0-20191106202628-ed6320f186d4
 	golang.org/x/lint v0.0.0-20191125180803-fdd1cda4f05f // indirect
 	golang.org/x/net v0.0.0-20191204025024-5ee1b9f4859a
+	golang.org/x/tools v0.0.0-20200124021010-5c352bb417e0 // indirect
 )
 
 //replace github.com/SkycoinProject/dmsg => ../dmsg

go.sum

@@ -200,7 +200,14 @@ golang.org/x/tools v0.0.0-20191206204035-259af5ff87bd/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20200110213125-a7a6caa82ab2 h1:V9r/14uGBqLgNlHRYWdVqjMdWkcOHnE2KG8DwVqQSEc=
 golang.org/x/tools v0.0.0-20200110213125-a7a6caa82ab2/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200115044656-831fdb1e1868/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200116062425-473961ec044c h1:D0OxfnjPaEGt7AluXNompYUYGhoY3u6+bValgqfd1vE=
 golang.org/x/tools v0.0.0-20200116062425-473961ec044c/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200123162537-45e69182d161 h1:96UK0Rx+sC4aQeuLGFxtAiCxAWzUu0OA4hhWlg4YvPQ=
+golang.org/x/tools v0.0.0-20200123162537-45e69182d161/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200123220707-24841a4f5f7d h1:KTOebTDPx6KUWl0wm5bjotr4ceAFMsScfXJderPyZ2c=
+golang.org/x/tools v0.0.0-20200123220707-24841a4f5f7d/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200124021010-5c352bb417e0 h1:G9K47VwP2wDdADV683EnkOYQHhb20LSa80C4AE+Gskw=
+golang.org/x/tools v0.0.0-20200124021010-5c352bb417e0/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=

pkg/httputil/httputil.go

@@ -6,6 +6,8 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"strconv"
+	"strings"
 
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 	"github.com/gorilla/handlers"
@@ -71,3 +73,14 @@ func WriteLog(writer io.Writer, params handlers.LogFormatterParams) {
 		log.WithError(err).Warn("Failed to write log")
 	}
 }
+
+// SplitRPCAddr returns host and port and whatever error results from parsing the rpc address interface
+func SplitRPCAddr(rpcAddr string) (host string, port uint16, err error) {
+	addrToken := strings.Split(rpcAddr, ":")
+	uint64port, err := strconv.ParseUint(addrToken[1], 10, 16)
+	if err != nil {
+		return
+	}
+
+	return addrToken[0], uint16(uint64port), nil
+}

pkg/hypervisor/config.go

@@ -10,6 +10,8 @@ import (
 
 	"github.com/SkycoinProject/dmsg/cipher"
 
+	"github.com/SkycoinProject/skywire-mainnet/internal/skyenv"
+	"github.com/SkycoinProject/skywire-mainnet/pkg/httputil"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/util/pathutil"
 )
 
@@ -42,20 +44,28 @@ func (hk *Key) UnmarshalText(text []byte) error {
 
 // Config configures the hypervisor.
 type Config struct {
-	Cookies    CookieConfig    `json:"cookies"`     // Configures cookies (for session management).
-	Interfaces InterfaceConfig `json:"interfaces"`  // Configures exposed interfaces.
-	DBPath     string          `json:"db_path"`     // Path to store database file.
-	EnableAuth bool            `json:"enable_auth"` // Whether to enable user management.
-	PK         cipher.PubKey   `json:"public_key"`
-	SK         cipher.SecKey   `json:"secret_key"`
+	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 {
+func makeConfig(testenv bool) Config {
 	var c Config
 
 	pk, sk := cipher.GenerateKeyPair()
 	c.PK = pk
 	c.SK = sk
+
+	if testenv {
+		c.DmsgDiscovery = skyenv.TestDmsgDiscAddr
+	} else {
+		c.DmsgDiscovery = skyenv.DefaultDmsgDiscAddr
+	}
+
 	c.EnableAuth = true
 	c.Cookies.HashKey = cipher.RandByte(hashKeyLen)
 	c.Cookies.BlockKey = cipher.RandByte(blockKeyLen)
@@ -66,29 +76,28 @@ func makeConfig() Config {
 }
 
 // GenerateWorkDirConfig generates a config with default values and uses db from current working directory.
-func GenerateWorkDirConfig() Config {
+func GenerateWorkDirConfig(testenv bool) Config {
 	dir, err := os.Getwd()
 	if err != nil {
 		log.Fatalf("failed to generate WD config: %s", dir)
 	}
-
-	c := makeConfig()
+	c := makeConfig(testenv)
 	c.DBPath = filepath.Join(dir, "users.db")
 
 	return c
 }
 
 // GenerateHomeConfig generates a config with default values and uses db from user's home folder.
-func GenerateHomeConfig() Config {
-	c := makeConfig()
+func GenerateHomeConfig(testenv bool) Config {
+	c := makeConfig(testenv)
 	c.DBPath = filepath.Join(pathutil.HomeDir(), ".skycoin/hypervisor/users.db")
 
 	return c
 }
 
 // GenerateLocalConfig generates a config with default values and uses db from shared folder.
-func GenerateLocalConfig() Config {
-	c := makeConfig()
+func GenerateLocalConfig(testenv bool) Config {
+	c := makeConfig(testenv)
 	c.DBPath = "/usr/local/SkycoinProject/hypervisor/users.db"
 
 	return c
@@ -155,3 +164,8 @@ 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) {
+	return httputil.SplitRPCAddr(c.RPCAddr)
+}

pkg/hypervisor/hypervisor.go

@@ -7,15 +7,15 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
-	"net"
 	"net/http"
 	"net/rpc"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/SkycoinProject/dmsg"
 	"github.com/SkycoinProject/dmsg/cipher"
-	"github.com/SkycoinProject/dmsg/noise"
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 	"github.com/go-chi/chi"
 	"github.com/go-chi/chi/middleware"
@@ -42,7 +42,7 @@ var (
 )
 
 type appNodeConn struct {
-	Addr   *noise.Addr
+	Addr   dmsg.Addr
 	Client visor.RPCClient
 }
 
@@ -72,15 +72,15 @@ 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 *dmsg.Listener) error {
 	for {
-		conn, err := noise.WrapListener(lis, m.c.PK, m.c.SK, false, noise.HandshakeXK).Accept()
+		conn, err := lis.Accept()
 		if err != nil {
 			return err
 		}
 
-		addr := conn.RemoteAddr().(*noise.Addr)
-
+		addr := conn.RemoteAddr().(dmsg.Addr)
+		log.Infoln("accepted: ", addr.PK)
 		m.mu.Lock()
 		m.nodes[addr.PK] = appNodeConn{
 			Addr:   addr,
@@ -90,11 +90,6 @@ func (m *Node) ServeRPC(lis net.Listener) error {
 	}
 }
 
-type mockAddr string
-
-func (mockAddr) Network() string  { return "mock" }
-func (a mockAddr) String() string { return string(a) }
-
 // MockConfig configures how mock data is to be added.
 type MockConfig struct {
 	Nodes            int
@@ -115,9 +110,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,
 		}
@@ -284,7 +279,7 @@ func (m *Node) getNodes() http.HandlerFunc {
 			}
 
 			summaries = append(summaries, summaryResp{
-				TCPAddr: c.Addr.Addr.String(),
+				TCPAddr: c.Addr.String(),
 				Online:  err == nil,
 				Summary: summary,
 			})
@@ -305,7 +300,7 @@ func (m *Node) getNode() http.HandlerFunc {
 		}
 
 		httputil.WriteJSON(w, r, http.StatusOK, summaryResp{
-			TCPAddr: ctx.Addr.Addr.String(),
+			TCPAddr: ctx.Addr.String(),
 			Summary: summary,
 		})
 	})
@@ -412,6 +407,7 @@ type LogsRes struct {
 func (m *Node) appLogsSince() http.HandlerFunc {
 	return m.withCtx(m.appCtx, func(w http.ResponseWriter, r *http.Request, ctx *httpCtx) {
 		since := r.URL.Query().Get("since")
+		since = strings.Replace(since, " ", "+", 1) // we need to put '+' again that was replaced in the query string
 
 		// if time is not parsable or empty default to return all logs
 		t, err := time.Parse(time.RFC3339Nano, since)
@@ -696,7 +692,7 @@ func (m *Node) restart() 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()
@@ -705,6 +701,11 @@ func (m *Node) client(pk cipher.PubKey) (*noise.Addr, visor.RPCClient, bool) {
 }
 
 type httpCtx struct {
+	// Node
+	PK   cipher.PubKey
+	Addr dmsg.Addr
+	RPC  visor.RPCClient
+
 	// App
 	App *visor.AppState
 
@@ -713,11 +714,6 @@ type httpCtx struct {
 
 	// Route
 	RtKey routing.RouteID
-
-	// Node
-	PK   cipher.PubKey
-	Addr *noise.Addr
-	RPC  visor.RPCClient
 }
 
 type (

pkg/hypervisor/hypervisor_test.go

@@ -43,7 +43,7 @@ const (
 )
 
 func TestNewNode(t *testing.T) {
-	config := makeConfig()
+	config := makeConfig(false)
 
 	confDir, err := ioutil.TempDir(os.TempDir(), "SWHV")
 	require.NoError(t, err)