Integrate http tunnel

cmd/start.go

@@ -9,6 +9,7 @@ import (
 	"github.com/NodeFactoryIo/vedran-daemon/internal/node"
 	"github.com/NodeFactoryIo/vedran-daemon/internal/run"
 	"github.com/NodeFactoryIo/vedran-daemon/internal/telemetry"
+	"github.com/NodeFactoryIo/vedran-daemon/internal/tunnel"
 	"github.com/NodeFactoryIo/vedran-daemon/pkg/logger"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -85,15 +86,19 @@ func start(cmd *cobra.Command, _ []string) error {
 	lbClient := lb.NewClient(lbURL)
 	nodeClient := node.NewClient(rpcURL, metricsURL)
 	telemetry := telemetry.NewTelemetry()
+	tunnel := &tunnel.Tunnel{
+		NodeRPCURL: rpcURL,
+	}
 
-	err := run.Start(lbClient, nodeClient, telemetry, id, payoutAddress)
+	err := run.Start(tunnel, lbClient, nodeClient, telemetry, id, payoutAddress)
 	if err != nil {
 		return fmt.Errorf("Failed starting vedran daemon because: %v", err)
 	}
 
 	return nil
 }
 
+// Execute runs command
 func Execute() {
 	if err := startCmd.Execute(); err != nil {
 		fmt.Println(err)

go.mod

@@ -3,6 +3,7 @@ module github.com/NodeFactoryIo/vedran-daemon
 go 1.15
 
 require (
+	github.com/NodeFactoryIo/vedran v0.1.2-0.20201012103221-1832974f2e38
 	github.com/go-co-op/gocron v0.3.1
 	github.com/mitchellh/mapstructure v1.3.3
 	github.com/prometheus/common v0.13.0

internal/lb/client.go

@@ -18,13 +18,13 @@ const (
 type RegisterRequest struct {
 	ID            string `json:"id"`
 	ConfigHash    string `json:"config_hash"`
-	NodeURL       string `json:"node_url"`
 	PayoutAddress string `json:"payout_address"`
 }
 
-// TokenResponse from lb register endpoint
-type TokenResponse struct {
-	Token string `json:"token"`
+// RegisterResponse from lb register endpoint
+type RegisterResponse struct {
+	Token               string `json:"token"`
+	TunnelServerAddress string `json:"tunnel_server_address"`
 }
 
 // Client used to communicate with vedran load balancer
@@ -49,22 +49,21 @@ func NewClient(baseURL *url.URL) *Client {
 }
 
 // Register daemon with load balancer and store token in client
-func (c *Client) Register(id string, nodeURL string, payoutAddress string, configHash string) error {
+func (c *Client) Register(id string, payoutAddress string, configHash string) (*RegisterResponse, error) {
 	body := &RegisterRequest{
 		ID:            id,
-		NodeURL:       nodeURL,
 		PayoutAddress: payoutAddress,
 		ConfigHash:    configHash,
 	}
 	req, _ := c.newRequest(http.MethodPost, registerEndpoint, body)
-	tokenResponse := new(TokenResponse)
-	_, err := c.do(req, tokenResponse)
+	registerResponse := new(RegisterResponse)
+	_, err := c.do(req, registerResponse)
 
-	if tokenResponse.Token != "" {
-		c.Token = tokenResponse.Token
+	if registerResponse.Token != "" {
+		c.Token = registerResponse.Token
 	}
 
-	return err
+	return registerResponse, err
 }
 
 // newRequest creates an API request. A relative URL should be provided in urlStr, which will be resolved to the

internal/lb/client_test.go

@@ -72,7 +72,6 @@ func TestClient_Register(t *testing.T) {
 	}
 	type args struct {
 		id            string
-		nodeURL       string
 		payoutAddress string
 		configHash    string
 	}
@@ -86,7 +85,7 @@ func TestClient_Register(t *testing.T) {
 	}{
 		{
 			name:    "Returns error if client does not return 200",
-			args:    args{"test-id", "http://localhost:3000", "0xtestaddress", "config-hash"},
+			args:    args{"test-id", "0xtestaddress", "config-hash"},
 			fields:  fields{http.DefaultClient, "valid", ""},
 			wantErr: true,
 			want:    "",
@@ -95,12 +94,12 @@ func TestClient_Register(t *testing.T) {
 			}},
 		{
 			name:    "Sets token on client if request valid",
-			args:    args{"test-id", "http://localhost:3000", "0xtestaddress", "config-hash"},
+			args:    args{"test-id", "0xtestaddress", "config-hash"},
 			fields:  fields{http.DefaultClient, "valid", ""},
 			wantErr: false,
 			want:    "test-token",
 			handleFunc: func(w http.ResponseWriter, r *http.Request) {
-				_, _ = io.WriteString(w, `{"token": "test-token"}`)
+				_, _ = io.WriteString(w, `{"token": "test-token", "tunnel_url": "192.168.1.31:5223"}`)
 			}},
 	}
 	for _, tt := range tests {
@@ -120,16 +119,15 @@ func TestClient_Register(t *testing.T) {
 			}
 			mux.HandleFunc("/api/v1/nodes", tt.handleFunc)
 
-			err := c.Register(tt.args.id, tt.args.nodeURL, tt.args.payoutAddress, tt.args.configHash)
+			registerResponse, err := c.Register(tt.args.id, tt.args.payoutAddress, tt.args.configHash)
 
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Client.Register() error = %v, wantErr %v", err, tt.wantErr)
 			}
 
-			if tt.want != c.Token {
+			if tt.want != c.Token || tt.want != registerResponse.Token {
 				t.Errorf("Client.Register() token = %s, want %s", c.Token, tt.want)
 			}
-
 		})
 
 		teardown()

internal/run/run.go

@@ -8,14 +8,15 @@ import (
 	"github.com/NodeFactoryIo/vedran-daemon/internal/lb"
 	"github.com/NodeFactoryIo/vedran-daemon/internal/node"
 	"github.com/NodeFactoryIo/vedran-daemon/internal/telemetry"
+	"github.com/NodeFactoryIo/vedran-daemon/internal/tunnel"
 	"github.com/go-co-op/gocron"
 	log "github.com/sirupsen/logrus"
 )
 
 var sleep = time.Sleep
 
 // Start registers to load balancer and starts sending telemetry
-func Start(lbClient *lb.Client, nodeClient node.Client, telemetry telemetry.Telemetry, id string, payoutAddress string) error {
+func Start(tunnel tunnel.Tunneler, lbClient *lb.Client, nodeClient node.Client, telemetry telemetry.Telemetry, id string, payoutAddress string) error {
 	var configHash hash.Hash32
 	for {
 		var err error
@@ -28,12 +29,14 @@ func Start(lbClient *lb.Client, nodeClient node.Client, telemetry telemetry.Tele
 		sleep(time.Second * 5)
 	}
 
-	err := lbClient.Register(id, nodeClient.GetRPCURL(), payoutAddress, base64.StdEncoding.EncodeToString(configHash.Sum(nil)[:]))
+	registerResponse, err := lbClient.Register(id, payoutAddress, base64.StdEncoding.EncodeToString(configHash.Sum(nil)[:]))
 	if err != nil {
 		return err
 	}
 	log.Infof("Registered to load balancer %s", lbClient.BaseURL.String())
 
+	go tunnel.StartTunnel(id, registerResponse.TunnelServerAddress, registerResponse.Token)
+
 	scheduler := gocron.NewScheduler(time.UTC)
 	telemetry.StartSendingTelemetry(scheduler, lbClient, nodeClient)
 	return nil

internal/run/run_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/NodeFactoryIo/vedran-daemon/internal/lb"
 	nodeMocks "github.com/NodeFactoryIo/vedran-daemon/mocks/node"
 	telemetryMocks "github.com/NodeFactoryIo/vedran-daemon/mocks/telemetry"
+	tunnelMocks "github.com/NodeFactoryIo/vedran-daemon/mocks/tunnel"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -60,6 +61,7 @@ func TestStart(t *testing.T) {
 		firstGetConfigHashError     error
 		secondGetConfigHashResult   hash.Hash32
 		secondGetConfigHashError    error
+		startTunnelCallCount        int
 	}{
 		{
 			name:    "Retries get config hash if get config hash fails and returns error if register fails",
@@ -71,16 +73,18 @@ func TestStart(t *testing.T) {
 			firstGetConfigHashError:   fmt.Errorf("Error"),
 			firstGetConfigHashResult:  nil,
 			secondGetConfigHashError:  nil,
-			secondGetConfigHashResult: testHash},
+			secondGetConfigHashResult: testHash,
+			startTunnelCallCount:      0},
 		{
 			name:    "Returns nil if startSendingTelemetry succeeds",
 			args:    args{lbClient, "test-id", "0xtestpayoutaddress"},
 			wantErr: false,
 			handleFunc: func(w http.ResponseWriter, r *http.Request) {
-				_, _ = io.WriteString(w, `{"token": "test-token"}`)
+				_, _ = io.WriteString(w, `{"token": "test-token", "tunnel_url": "url:5223"}`)
 			},
 			firstGetConfigHashError:  nil,
-			firstGetConfigHashResult: testHash},
+			firstGetConfigHashResult: testHash,
+			startTunnelCallCount:     1},
 	}
 
 	for _, tt := range tests {
@@ -89,6 +93,8 @@ func TestStart(t *testing.T) {
 
 			telemetryMock := &telemetryMocks.Telemetry{}
 			telemetryMock.On("StartSendingTelemetry", mock.Anything, mock.Anything, mock.Anything).Return()
+			tunnelMock := &tunnelMocks.Tunneler{}
+			tunnelMock.On("StartTunnel", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return()
 			nodeClient.On("GetRPCURL").Return("http://localhost:9933")
 			nodeClient.On("GetConfigHash").Once().Return(tt.firstGetConfigHashResult, tt.firstGetConfigHashError)
 			nodeClient.On("GetConfigHash").Once().Return(tt.secondGetConfigHashResult, tt.secondGetConfigHashError)
@@ -97,7 +103,7 @@ func TestStart(t *testing.T) {
 			lbClient.BaseURL = url
 			mux.HandleFunc("/api/v1/nodes", tt.handleFunc)
 
-			err := Start(tt.args.client, nodeClient, telemetryMock, tt.args.id, tt.args.payoutAddress)
+			err := Start(tunnelMock, tt.args.client, nodeClient, telemetryMock, tt.args.id, tt.args.payoutAddress)
 
 			if (err != nil) != tt.wantErr {
 				t.Errorf("Start() error = %v, wantErr %v", err, tt.wantErr)

internal/scheduler/scheduler.go

@@ -2,6 +2,7 @@ package scheduler
 
 import "github.com/go-co-op/gocron"
 
+// Scheduler schedules go cron jobs
 type Scheduler interface {
 	Every(interval uint64) *gocron.Scheduler
 	Do(jobFun interface{}, params ...interface{}) (*gocron.Job, error)

internal/tunnel/tunnel.go

@@ -0,0 +1,56 @@
+package tunnel
+
+import (
+	"net/url"
+	"time"
+
+	"github.com/NodeFactoryIo/vedran/pkg/http-tunnel/client"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	DefaultBackoffInterval    = 500 * time.Millisecond
+	DefaultBackoffMultiplier  = 1.5
+	DefaultBackoffMaxInterval = 60 * time.Second
+	DefaultBackoffMaxTime     = 15 * time.Minute
+	Protocol                  = "tcp"
+	RemoteAddr                = "0.0.0.0:AUTO"
+)
+
+// Tunnel is tunnel connection with load balancer
+type Tunnel struct {
+	NodeRPCURL *url.URL
+}
+
+// Tunneler defines methods for connecting to load balancer tunnel
+type Tunneler interface {
+	// StartTunnel connects to load balancer tunnel port and creates connection.
+	// nodeID is id that is passed in daemon,
+	// tunnelServerAddress is public address of load balancer tunnel server and
+	// token is jwt token given when registering with load balancer.
+	StartTunnel(nodeID string, tunnelServerAddress string, token string)
+}
+
+func (t *Tunnel) StartTunnel(nodeID string, tunnelServerAddress string, token string) {
+	c, err := client.NewClient(&client.ClientConfig{
+		ServerAddress: tunnelServerAddress,
+		Tunnels: map[string]*client.Tunnel{
+			"default": {
+				Protocol:   Protocol,
+				Addr:       t.NodeRPCURL.Host,
+				RemoteAddr: RemoteAddr,
+			},
+		},
+		Logger:    log.NewEntry(log.New()),
+		AuthToken: token,
+		IdName:    nodeID,
+	})
+	if err != nil {
+		log.Fatal("Failed to connect to tunnel: ", err)
+	}
+
+	err = c.Start()
+	if err != nil {
+		log.Fatal("Failed to start tunnels: ", err)
+	}
+}