Feature/dmsg hypervisor

cmd/hypervisor/commands/root.go

@@ -1,11 +1,13 @@
 package commands
 
 import (
+	"context"
 	"fmt"
-	"net"
 	"net/http"
 	"os"
 
+	"github.com/SkycoinProject/dmsg"
+	"github.com/SkycoinProject/dmsg/disc"
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 	"github.com/spf13/cobra"
 
@@ -62,10 +64,22 @@ 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))
+
+			ctx := context.Background()
+			if err = dmsgC.InitiateServerConnections(ctx, 1); err != nil {
+				log.Fatalln("failed to initiate dmsg server connections:", err)
+			}
+			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.sum

@@ -1,5 +1,4 @@
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/SkycoinProject/dmsg v0.0.0-20190918162505-cc37af88b8bf/go.mod h1:WiVS8uAqzFsuWcRlDBsjrxqAiD41ftxS7S7CDTXZpko=
 github.com/SkycoinProject/dmsg v0.0.0-20190918181704-b7cccca1451e h1:JMIfHAjNSJLvKdnwXpUudbnYG627qD5d6aye+gmDups=
 github.com/SkycoinProject/dmsg v0.0.0-20190918181704-b7cccca1451e/go.mod h1:iB171maaQybaYptTK/mhzb02uonVRKQ8oCsJ7blquRw=
 github.com/SkycoinProject/skycoin v0.26.0 h1:8/ZRZb2VM2DM4YTIitRJMZ3Yo/3H1FFmbCMx5o6ekmA=
@@ -158,6 +157,7 @@ golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3 h1:7TYNF4UdlohbFwpNH04CoPMp1cHUZgO1Ebq5r2hIjfo=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/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/app/log_store_test.go

@@ -29,7 +29,6 @@ func TestLogStore(t *testing.T) {
 	require.NoError(t, err)
 
 	err = ls.Store(t1, "bar")
-	fmt.Println("original: ", t1.Format(time.RFC3339Nano))
 	require.NoError(t, err)
 
 	t2, err := time.Parse(time.RFC3339, "2000-02-01T00:00:00Z")

pkg/hypervisor/config.go

@@ -4,8 +4,10 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"net/http"
+	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"time"
 
 	"github.com/SkycoinProject/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,23 +6,23 @@ import (
 	"errors"
 	"fmt"
 	"math/rand"
-	"net"
 	"net/http"
 	"net/rpc"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
-	"github.com/go-chi/chi"
-	"github.com/go-chi/chi/middleware"
-	"github.com/google/uuid"
+	"github.com/SkycoinProject/dmsg"
 	"github.com/SkycoinProject/dmsg/cipher"
-	"github.com/SkycoinProject/dmsg/noise"
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/app"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/httputil"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/routing"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/visor"
+	"github.com/go-chi/chi"
+	"github.com/go-chi/chi/middleware"
+	"github.com/google/uuid"
 )
 
 var (
@@ -31,7 +31,7 @@ var (
 )
 
 type appNodeConn struct {
-	Addr   *noise.Addr
+	Addr   dmsg.Addr
 	Client visor.RPCClient
 }
 
@@ -60,13 +60,13 @@ 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)
 		m.mu.Lock()
 		m.nodes[addr.PK] = appNodeConn{
 			Addr:   addr,
@@ -99,9 +99,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,
 		}
@@ -245,7 +245,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,
 			})
 		}
@@ -263,7 +263,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,
 		})
 	})
@@ -337,6 +337,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 parseable or empty default to return all logs
 		t, err := time.Parse(time.RFC3339Nano, since)
@@ -572,7 +573,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()
@@ -582,7 +583,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/setup/idreservoir.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/SkycoinProject/dmsg/cipher"
-
 	"github.com/SkycoinProject/skywire-mainnet/pkg/routing"
 )
 

pkg/setup/node.go

@@ -12,9 +12,9 @@ import (
 	"github.com/SkycoinProject/dmsg/disc"
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 
-	"github.com/SkycoinProject/skywire-mainnet/pkg/snet"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/metrics"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/routing"
+	"github.com/SkycoinProject/skywire-mainnet/pkg/snet"
 )
 
 // Node performs routes setup operations over messaging channel.

pkg/setup/node_test.go

@@ -12,15 +12,14 @@ import (
 	"testing"
 	"time"
 
-"github.com/SkycoinProject/dmsg"
-"github.com/SkycoinProject/dmsg/cipher"
+	"github.com/SkycoinProject/dmsg"
+	"github.com/SkycoinProject/dmsg/cipher"
 	"github.com/SkycoinProject/dmsg/disc"
-	"github.com/stretchr/testify/require"
-	"golang.org/x/net/nettest"
-
 	"github.com/SkycoinProject/skywire-mainnet/pkg/metrics"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/routing"
 	"github.com/SkycoinProject/skywire-mainnet/pkg/snet"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/nettest"
 
 	"github.com/SkycoinProject/skycoin/src/util/logging"
 )

pkg/visor/config.go

@@ -167,7 +167,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.