Move all rmb handlers to api-gateway

.github/workflows/bins.yaml

@@ -65,12 +65,6 @@ jobs:
       package: cloudhypervisor
     secrets:
       token: ${{ secrets.HUB_JWT }}
-  rmb:
-    uses: ./.github/workflows/bin-package.yaml
-    with:
-      package: rmb
-    secrets:
-      token: ${{ secrets.HUB_JWT }}
   tailstream:
     uses: ./.github/workflows/bin-package.yaml
     with:

cmds/modules/api_gateway/main.go

@@ -3,16 +3,19 @@ package apigateway
 import (
 	"context"
 	"crypto/ed25519"
+	"encoding/hex"
 	"fmt"
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 	substrate "github.com/threefoldtech/tfchain/clients/tfchain-client-go"
+	"github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go/peer"
 	"github.com/threefoldtech/zbus"
-	apigateway "github.com/threefoldtech/zos/pkg/api_gateway"
 	"github.com/threefoldtech/zos/pkg/environment"
 	"github.com/threefoldtech/zos/pkg/stubs"
+	substrategw "github.com/threefoldtech/zos/pkg/substrate_gateway"
 	"github.com/threefoldtech/zos/pkg/utils"
+	zosapi "github.com/threefoldtech/zos/pkg/zos_api"
 	"github.com/urfave/cli/v2"
 )
 
@@ -65,17 +68,39 @@ func action(cli *cli.Context) error {
 		return fmt.Errorf("failed to create substrate manager: %w", err)
 	}
 
-	gw, err := apigateway.NewAPIGateway(manager, id)
+	router := peer.NewRouter()
+	gw, err := substrategw.NewSubstrateGateway(manager, id)
 	if err != nil {
 		return fmt.Errorf("failed to create api gateway: %w", err)
 	}
 
 	server.Register(zbus.ObjectID{Name: "api-gateway", Version: "0.0.1"}, gw)
+	api, err := zosapi.NewZosAPI(manager, redis, msgBrokerCon)
+	if err != nil {
+		return fmt.Errorf("failed to create zos api: %w", err)
+	}
+	api.SetupRoutes(router)
+
+	pair, err := id.KeyPair()
+	if err != nil {
+		return err
+	}
 
 	ctx, _ := utils.WithSignal(context.Background())
 	utils.OnDone(ctx, func(_ error) {
 		log.Info().Msg("shutting down")
 	})
+	_, err = peer.NewPeer(
+		ctx,
+		hex.EncodeToString(pair.Seed()),
+		manager,
+		router.Serve,
+		peer.WithKeyType(peer.KeyTypeEd25519),
+		peer.WithRelay(environment.MustGet().RelayURL...),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to start a new rmb peer: %w", err)
+	}
 
 	log.Info().
 		Str("broker", msgBrokerCon).

cmds/modules/networkd/main.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go"
 	"github.com/threefoldtech/zos/pkg/environment"
 	"github.com/threefoldtech/zos/pkg/network/dhcp"
 	"github.com/threefoldtech/zos/pkg/network/public"
@@ -156,25 +155,6 @@ func action(cli *cli.Context) error {
 		return errors.Wrap(err, "error creating network manager")
 	}
 
-	mBus, err := rmb.NewRouter(broker)
-	if err != nil {
-		return errors.Wrap(err, "failed to initialize message bus")
-	}
-
-	zosRouter := mBus.Subroute("zos")
-	zosRouter.Use(rmb.LoggerMiddleware)
-
-	if _, err := network.NewNetworkMessageBus(zosRouter, networker, stubs.NewAPIGatewayStub(client)); err != nil {
-		return errors.Wrap(err, "failed to initialize rmb api")
-	}
-
-	// we need to start both rmb server and zbus server.
-	go func(ctx context.Context) {
-		if err := startRmbServer(ctx, mBus); err != nil {
-			log.Fatal().Err(err).Msg("rmb exited unexpectedly")
-		}
-	}(ctx)
-
 	log.Info().Msg("start zbus server")
 	if err := startZBusServer(ctx, broker, networker); err != nil {
 		return errors.Wrap(err, "unexpected error")
@@ -183,21 +163,6 @@ func action(cli *cli.Context) error {
 	return nil
 }
 
-func startRmbServer(ctx context.Context, bus *rmb.DefaultRouter) error {
-	log.Info().
-		Msg("starting networkd rmb module")
-
-	for _, handler := range bus.Handlers() {
-		log.Debug().Msgf("registered handler: %s", handler)
-	}
-
-	if err := bus.Run(ctx); err != nil && err != context.Canceled {
-		return err
-	}
-
-	return nil
-}
-
 func startZBusServer(ctx context.Context, broker string, networker pkg.Networker) error {
 
 	server, err := zbus.NewRedisServer(module, broker, 1)

cmds/modules/noded/main.go

@@ -3,21 +3,16 @@ package noded
 import (
 	"context"
 	"crypto/ed25519"
-	"encoding/json"
 	"fmt"
-	"os/exec"
-	"strings"
 	"time"
 
 	"github.com/cenkalti/backoff"
 	"github.com/pkg/errors"
 	"github.com/urfave/cli/v2"
 
 	substrate "github.com/threefoldtech/tfchain/clients/tfchain-client-go"
-	"github.com/threefoldtech/tfgrid-sdk-go/rmb-sdk-go"
 	"github.com/threefoldtech/zos/pkg/app"
 	"github.com/threefoldtech/zos/pkg/capacity"
-	"github.com/threefoldtech/zos/pkg/diagnostics"
 	"github.com/threefoldtech/zos/pkg/environment"
 	"github.com/threefoldtech/zos/pkg/events"
 	"github.com/threefoldtech/zos/pkg/monitord"
@@ -166,48 +161,6 @@ func action(cli *cli.Context) error {
 		WithVirtualized(len(hypervisor) != 0)
 
 	go registerationServer(ctx, msgBrokerCon, env, info)
-	bus, err := rmb.NewRouter(msgBrokerCon)
-	if err != nil {
-		return errors.Wrap(err, "failed to initialize message bus server")
-	}
-
-	bus.WithHandler("zos.system.version", func(ctx context.Context, payload []byte) (interface{}, error) {
-		ver := stubs.NewVersionMonitorStub(redis)
-		output, err := exec.CommandContext(ctx, "zinit", "-V").CombinedOutput()
-		var zInitVer string
-		if err != nil {
-			zInitVer = err.Error()
-		} else {
-			zInitVer = strings.TrimSpace(strings.TrimPrefix(string(output), "zinit"))
-		}
-
-		version := struct {
-			ZOS   string `json:"zos"`
-			ZInit string `json:"zinit"`
-		}{
-			ZOS:   ver.GetVersion(ctx).String(),
-			ZInit: zInitVer,
-		}
-
-		return version, nil
-	})
-
-	diagnosticsMgr, err := diagnostics.NewDiagnosticsManager(msgBrokerCon, redis)
-	if err != nil {
-		return errors.Wrap(err, "failed to create a new Diagnostics manager")
-	}
-	bus.WithHandler("zos.system.diagnostics", func(ctx context.Context, payload []byte) (interface{}, error) {
-		return diagnosticsMgr.GetSystemDiagnostics(ctx)
-	})
-
-	bus.WithHandler("zos.system.dmi", func(ctx context.Context, payload []byte) (interface{}, error) {
-		return dmi, nil
-	})
-
-	bus.WithHandler("zos.system.hypervisor", func(ctx context.Context, payload []byte) (interface{}, error) {
-		return hypervisor, nil
-	})
-
 	log.Info().Msg("start perf scheduler")
 
 	perfMon, err := perf.NewPerformanceMonitor(msgBrokerCon)
@@ -224,28 +177,6 @@ func action(cli *cli.Context) error {
 	if err = perfMon.Run(ctx); err != nil {
 		return errors.Wrap(err, "failed to run the scheduler")
 	}
-	bus.WithHandler("zos.perf.get", func(ctx context.Context, payload []byte) (interface{}, error) {
-		type Payload struct {
-			Name string
-		}
-		var request Payload
-		err := json.Unmarshal(payload, &request)
-		if err != nil {
-			return nil, errors.Wrapf(err, "failed to unmarshal payload: %v", payload)
-		}
-
-		return perfMon.Get(request.Name)
-	})
-	bus.WithHandler("zos.perf.get_all", func(ctx context.Context, payload []byte) (interface{}, error) {
-		return perfMon.GetAll()
-	})
-
-	// answer calls for dmi
-	go func() {
-		if err := bus.Run(ctx); err != nil {
-			log.Fatal().Err(err).Msg("message bus handler failure")
-		}
-	}()
 
 	// block indefinietly, and other modules will get an error
 	// when calling the registrar NodeID
@@ -298,12 +229,7 @@ func action(cli *cli.Context) error {
 
 	server.Register(zbus.ObjectID{Name: "host", Version: "0.0.1"}, host)
 	server.Register(zbus.ObjectID{Name: "system", Version: "0.0.1"}, system)
-
-	go func() {
-		if err := server.Run(ctx); err != nil && err != context.Canceled {
-			log.Fatal().Err(err).Msg("unexpected error")
-		}
-	}()
+	server.Register(zbus.ObjectID{Name: "performance-monitor", Version: "0.0.1"}, perfMon)
 
 	log.Info().Uint32("node", node).Uint32("twin", twin).Msg("node registered")
 
@@ -326,19 +252,10 @@ func action(cli *cli.Context) error {
 	if err != nil {
 		return err
 	}
-
-	log.Debug().Msg("start message bus")
-	for {
-		err := runMsgBus(ctx, sk, env.SubstrateURL, env.RelayURL, msgBrokerCon)
-
-		if ctxErr := ctx.Err(); ctxErr != nil {
-			// if context is cancelled, then it's a normal shutdown
-			return nil
-		}
-
-		log.Error().Err(err).Msg("rmb-peer exited with an error, restarting")
-		<-time.After(1 * time.Second)
+	if err := server.Run(ctx); err != nil && err != context.Canceled {
+		log.Fatal().Err(err).Msg("unexpected error")
 	}
+	return nil
 }
 
 func retryNotify(err error, d time.Duration) {

cmds/modules/noded/public.go

@@ -37,11 +37,11 @@ func public(ctx context.Context, nodeID uint32, env environment.Environment, cl
 		return errors.Wrap(err, "failed to subscribe to node events")
 	}
 
-	apiGateway := stubs.NewAPIGatewayStub(cl)
+	substrateGateway := stubs.NewSubstrateGatewayStub(cl)
 
 reapply:
 	for {
-		node, err := apiGateway.GetNode(ctx, nodeID)
+		node, err := substrateGateway.GetNode(ctx, nodeID)
 		if err != nil {
 			return errors.Wrap(err, "failed to get node public config")
 		}