Skip to content

Commit

Permalink
engine: GetSubsystemsStatus fix (thrasher-corp#773)
Browse files Browse the repository at this point in the history
* engine: GetSubsystemsStatus fix

* engine: force map literal to stop doubling up on keys, expanded test coverage

* engine: Deploy default for migration requirement.

* glorious: nits addr

* glorious: suggestion

* tests: fix
  • Loading branch information
shazbert authored Sep 3, 2021
1 parent a1a667b commit a54c510
Show file tree
Hide file tree
Showing 12 changed files with 252 additions and 73 deletions.
6 changes: 3 additions & 3 deletions communications/communications.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,13 @@ type Communications struct {
base.IComm
}

// ErrNoCommunicationRelayersEnabled returns when no relayers enabled
var ErrNoCommunicationRelayersEnabled = errors.New("no communication relayers enabled")
// ErrNoRelayersEnabled returns when no communication relayers are enabled
var ErrNoRelayersEnabled = errors.New("no communication relayers are enabled")

// NewComm sets up and returns a pointer to a Communications object
func NewComm(cfg *base.CommunicationsConfig) (*Communications, error) {
if !cfg.IsAnyEnabled() {
return nil, ErrNoCommunicationRelayersEnabled
return nil, ErrNoRelayersEnabled
}

var comm Communications
Expand Down
5 changes: 4 additions & 1 deletion dispatch/dispatch.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@ import (
"github.com/thrasher-corp/gocryptotrader/log"
)

// ErrNotRunning defines an error when the dispatcher is not running
var ErrNotRunning = errors.New("dispatcher not running")

// Name is an exported subsystem name
const Name = "dispatch"

Expand Down Expand Up @@ -117,7 +120,7 @@ func (d *Dispatcher) start(workers, channelCapacity int) error {
// stop stops the service and shuts down all worker routines
func (d *Dispatcher) stop() error {
if !atomic.CompareAndSwapUint32(&d.running, 1, 0) {
return errors.New("dispatcher not running")
return ErrNotRunning
}
close(d.shutdown)
ch := make(chan struct{})
Expand Down
4 changes: 2 additions & 2 deletions engine/communication_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,8 @@ func TestSetup(t *testing.T) {
}

_, err = SetupCommunicationManager(&base.CommunicationsConfig{})
if !errors.Is(err, communications.ErrNoCommunicationRelayersEnabled) {
t.Errorf("error '%v', expected '%v'", err, communications.ErrNoCommunicationRelayersEnabled)
if !errors.Is(err, communications.ErrNoRelayersEnabled) {
t.Errorf("error '%v', expected '%v'", err, communications.ErrNoRelayersEnabled)
}

m, err := SetupCommunicationManager(&base.CommunicationsConfig{
Expand Down
10 changes: 8 additions & 2 deletions engine/connection_manager.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package engine

import (
"errors"
"fmt"
"sync/atomic"

Expand All @@ -12,6 +13,8 @@ import (
// ConnectionManagerName is an exported subsystem name
const ConnectionManagerName = "internet_monitor"

var errConnectionCheckerIsNil = errors.New("connection checker is nil")

// connectionManager manages the connchecker
type connectionManager struct {
started int32
Expand Down Expand Up @@ -72,14 +75,17 @@ func (m *connectionManager) Start() error {
// Stop stops the connection manager
func (m *connectionManager) Stop() error {
if m == nil {
return fmt.Errorf("connection manager %w", ErrNilSubsystem)
return fmt.Errorf("connection manager: %w", ErrNilSubsystem)
}
if atomic.LoadInt32(&m.started) == 0 {
return fmt.Errorf("connection manager %w", ErrSubSystemNotStarted)
return fmt.Errorf("connection manager: %w", ErrSubSystemNotStarted)
}
defer func() {
atomic.CompareAndSwapInt32(&m.started, 1, 0)
}()
if m.conn == nil {
return fmt.Errorf("connection manager: %w", errConnectionCheckerIsNil)
}
log.Debugln(log.ConnectionMgr, "Connection manager shutting down...")
m.conn.Shutdown()
log.Debugln(log.ConnectionMgr, "Connection manager stopped.")
Expand Down
4 changes: 4 additions & 0 deletions engine/connection_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,10 @@ func TestConnectionMonitorStart(t *testing.T) {

func TestConnectionMonitorStop(t *testing.T) {
t.Parallel()
err := (&connectionManager{started: 1}).Stop()
if !errors.Is(err, errConnectionCheckerIsNil) {
t.Errorf("error '%v', expected '%v'", err, errConnectionCheckerIsNil)
}
m, err := setupConnectionManager(&config.ConnectionMonitorConfig{})
if !errors.Is(err, nil) {
t.Errorf("error '%v', expected '%v'", err, nil)
Expand Down
94 changes: 53 additions & 41 deletions engine/helpers.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,29 +34,31 @@ import (
)

var (
errCertExpired = errors.New("gRPC TLS certificate has expired")
errCertDataIsNil = errors.New("gRPC TLS certificate PEM data is nil")
errCertTypeInvalid = errors.New("gRPC TLS certificate type is invalid")
errCertExpired = errors.New("gRPC TLS certificate has expired")
errCertDataIsNil = errors.New("gRPC TLS certificate PEM data is nil")
errCertTypeInvalid = errors.New("gRPC TLS certificate type is invalid")
errSubsystemNotFound = errors.New("subsystem not found")
errGRPCManagementFault = errors.New("cannot manage GRPC subsystem via GRPC. Please manually change your config")
)

// GetSubsystemsStatus returns the status of various subsystems
func (bot *Engine) GetSubsystemsStatus() map[string]bool {
systems := make(map[string]bool)
systems[SyncManagerName] = bot.CommunicationsManager.IsRunning()
systems[ConnectionManagerName] = bot.connectionManager.IsRunning()
systems[OrderManagerName] = bot.OrderManager.IsRunning()
systems[PortfolioManagerName] = bot.portfolioManager.IsRunning()
systems[NTPManagerName] = bot.ntpManager.IsRunning()
systems[DatabaseConnectionManagerName] = bot.DatabaseManager.IsRunning()
systems[SyncManagerName] = bot.Settings.EnableExchangeSyncManager
systems[grpcName] = bot.Settings.EnableGRPC
systems[grpcProxyName] = bot.Settings.EnableGRPCProxy
systems[vm.Name] = bot.gctScriptManager.IsRunning()
systems[DeprecatedName] = bot.Settings.EnableDeprecatedRPC
systems[WebsocketName] = bot.Settings.EnableWebsocketRPC
systems[dispatch.Name] = dispatch.IsRunning()
systems[dataHistoryManagerName] = bot.dataHistoryManager.IsRunning()
return systems
return map[string]bool{
CommunicationsManagerName: bot.CommunicationsManager.IsRunning(),
ConnectionManagerName: bot.connectionManager.IsRunning(),
OrderManagerName: bot.OrderManager.IsRunning(),
PortfolioManagerName: bot.portfolioManager.IsRunning(),
NTPManagerName: bot.ntpManager.IsRunning(),
DatabaseConnectionManagerName: bot.DatabaseManager.IsRunning(),
SyncManagerName: bot.Settings.EnableExchangeSyncManager,
grpcName: bot.Settings.EnableGRPC,
grpcProxyName: bot.Settings.EnableGRPCProxy,
vm.Name: bot.gctScriptManager.IsRunning(),
DeprecatedName: bot.Settings.EnableDeprecatedRPC,
WebsocketName: bot.Settings.EnableWebsocketRPC,
dispatch.Name: dispatch.IsRunning(),
dataHistoryManagerName: bot.dataHistoryManager.IsRunning(),
}
}

// RPCEndpoint stores an RPC endpoint status and addr
Expand All @@ -66,29 +68,40 @@ type RPCEndpoint struct {
}

// GetRPCEndpoints returns a list of RPC endpoints and their listen addrs
func GetRPCEndpoints() map[string]RPCEndpoint {
endpoints := make(map[string]RPCEndpoint)
endpoints[grpcName] = RPCEndpoint{
Started: Bot.Settings.EnableGRPC,
ListenAddr: "grpc://" + Bot.Config.RemoteControl.GRPC.ListenAddress,
}
endpoints[grpcProxyName] = RPCEndpoint{
Started: Bot.Settings.EnableGRPCProxy,
ListenAddr: "http://" + Bot.Config.RemoteControl.GRPC.GRPCProxyListenAddress,
}
endpoints[DeprecatedName] = RPCEndpoint{
Started: Bot.Settings.EnableDeprecatedRPC,
ListenAddr: "http://" + Bot.Config.RemoteControl.DeprecatedRPC.ListenAddress,
}
endpoints[WebsocketName] = RPCEndpoint{
Started: Bot.Settings.EnableWebsocketRPC,
ListenAddr: "ws://" + Bot.Config.RemoteControl.WebsocketRPC.ListenAddress,
}
return endpoints
func (bot *Engine) GetRPCEndpoints() (map[string]RPCEndpoint, error) {
if bot.Config == nil {
return nil, errNilConfig
}
return map[string]RPCEndpoint{
grpcName: {
Started: bot.Settings.EnableGRPC,
ListenAddr: "grpc://" + bot.Config.RemoteControl.GRPC.ListenAddress,
},
grpcProxyName: {
Started: bot.Settings.EnableGRPCProxy,
ListenAddr: "http://" + bot.Config.RemoteControl.GRPC.GRPCProxyListenAddress,
},
DeprecatedName: {
Started: bot.Settings.EnableDeprecatedRPC,
ListenAddr: "http://" + bot.Config.RemoteControl.DeprecatedRPC.ListenAddress,
},
WebsocketName: {
Started: bot.Settings.EnableWebsocketRPC,
ListenAddr: "ws://" + bot.Config.RemoteControl.WebsocketRPC.ListenAddress,
},
}, nil
}

// SetSubsystem enables or disables an engine subsystem
func (bot *Engine) SetSubsystem(subSystemName string, enable bool) error {
if bot == nil {
return errNilBot
}

if bot.Config == nil {
return errNilConfig
}

var err error
switch strings.ToLower(subSystemName) {
case CommunicationsManagerName:
Expand Down Expand Up @@ -227,7 +240,7 @@ func (bot *Engine) SetSubsystem(subSystemName string, enable bool) error {
}
return bot.apiServer.StopWebsocketServer()
case grpcName, grpcProxyName:
return errors.New("cannot manage GRPC subsystem via GRPC. Please manually change your config")
return errGRPCManagementFault
case dataHistoryManagerName:
if enable {
if bot.dataHistoryManager == nil {
Expand All @@ -251,8 +264,7 @@ func (bot *Engine) SetSubsystem(subSystemName string, enable bool) error {
}
return bot.gctScriptManager.Stop()
}

return errors.New("subsystem not found")
return fmt.Errorf("%s: %w", subSystemName, errSubsystemNotFound)
}

// GetExchangeOTPs returns OTP codes for all exchanges which have a otpsecret
Expand Down
146 changes: 146 additions & 0 deletions engine/helpers_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,18 @@ import (
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/common/convert"
"github.com/thrasher-corp/gocryptotrader/common/file"
"github.com/thrasher-corp/gocryptotrader/communications"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/database"
"github.com/thrasher-corp/gocryptotrader/dispatch"
"github.com/thrasher-corp/gocryptotrader/exchanges/account"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
"github.com/thrasher-corp/gocryptotrader/exchanges/stats"
"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
"github.com/thrasher-corp/gocryptotrader/gctscript/vm"
"github.com/thrasher-corp/gocryptotrader/log"
)

var testExchange = "Bitstamp"
Expand Down Expand Up @@ -88,6 +93,147 @@ func CreateTestBot(t *testing.T) *Engine {
return bot
}

func TestGetSubsystemsStatus(t *testing.T) {
m := (&Engine{}).GetSubsystemsStatus()
if len(m) != 14 {
t.Fatalf("subsystem count is wrong expecting: %d but received: %d", 14, len(m))
}
}

func TestGetRPCEndpoints(t *testing.T) {
_, err := (&Engine{}).GetRPCEndpoints()
if !errors.Is(err, errNilConfig) {
t.Fatalf("received: %v, but expected: %v", err, errNilConfig)
}

m, err := (&Engine{Config: &config.Config{}}).GetRPCEndpoints()
if !errors.Is(err, nil) {
t.Fatalf("received: %v, but expected: %v", err, nil)
}
if len(m) != 4 {
t.Fatalf("expected length: %d but received: %d", 4, len(m))
}
}

func TestSetSubsystem(t *testing.T) {
testCases := []struct {
Subsystem string
Engine *Engine
EnableError error
DisableError error
}{
{Subsystem: "sillyBilly", EnableError: errNilBot, DisableError: errNilBot},
{Subsystem: "sillyBilly", Engine: &Engine{}, EnableError: errNilConfig, DisableError: errNilConfig},
{Subsystem: "sillyBilly", Engine: &Engine{Config: &config.Config{}}, EnableError: errSubsystemNotFound, DisableError: errSubsystemNotFound},
{
Subsystem: CommunicationsManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: communications.ErrNoRelayersEnabled,
DisableError: ErrNilSubsystem,
},
{
Subsystem: ConnectionManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: nil,
DisableError: nil,
},
{
Subsystem: OrderManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: nil,
DisableError: nil,
},
{
Subsystem: PortfolioManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: errNilExchangeManager,
DisableError: ErrNilSubsystem,
},
{
Subsystem: NTPManagerName,
Engine: &Engine{Config: &config.Config{Logging: log.Config{Enabled: convert.BoolPtr(false)}}},
EnableError: errNilNTPConfigValues,
DisableError: ErrNilSubsystem,
},
{
Subsystem: DatabaseConnectionManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: database.ErrDatabaseSupportDisabled,
DisableError: ErrSubSystemNotStarted,
},
{
Subsystem: SyncManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: errNoSyncItemsEnabled,
DisableError: ErrNilSubsystem,
},
{
Subsystem: dispatch.Name,
Engine: &Engine{Config: &config.Config{}},
EnableError: nil,
DisableError: nil,
},

{
Subsystem: DeprecatedName,
Engine: &Engine{Config: &config.Config{}, Settings: Settings{ConfigFile: config.DefaultFilePath()}},
EnableError: errServerDisabled,
DisableError: ErrSubSystemNotStarted,
},
{
Subsystem: WebsocketName,
Engine: &Engine{Config: &config.Config{}, Settings: Settings{ConfigFile: config.DefaultFilePath()}},
EnableError: errServerDisabled,
DisableError: ErrSubSystemNotStarted,
},
{
Subsystem: grpcName,
Engine: &Engine{Config: &config.Config{}},
EnableError: errGRPCManagementFault,
DisableError: errGRPCManagementFault},
{
Subsystem: grpcProxyName,
Engine: &Engine{Config: &config.Config{}},
EnableError: errGRPCManagementFault,
DisableError: errGRPCManagementFault},
{
Subsystem: dataHistoryManagerName,
Engine: &Engine{Config: &config.Config{}},
EnableError: database.ErrNilInstance,
DisableError: ErrNilSubsystem,
},
{
Subsystem: vm.Name,
Engine: &Engine{Config: &config.Config{}},
EnableError: nil,
DisableError: nil,
},
}

for _, tt := range testCases {
tt := tt
t.Run(tt.Subsystem, func(t *testing.T) {
t.Parallel()
err := tt.Engine.SetSubsystem(tt.Subsystem, true)
if !errors.Is(err, tt.EnableError) {
t.Fatalf(
"while enabled %s subsystem received: %#v, but expected: %v",
tt.Subsystem,
err,
tt.EnableError)
}
err = tt.Engine.SetSubsystem(tt.Subsystem, false)
if !errors.Is(err, tt.DisableError) {
t.Fatalf(
"while disabling %s subsystem received: %#v, but expected: %v",
tt.Subsystem,
err,
tt.DisableError)
}
})
}
}

func TestGetExchangeOTPs(t *testing.T) {
t.Parallel()
bot := CreateTestBot(t)
Expand Down
Loading

0 comments on commit a54c510

Please sign in to comment.