Merge pull request #6099 from mysteriumnetwork/control-plane

Control plane messages support

Author: tzununbekov

cmd/commands/service/command.go

@@ -33,6 +33,7 @@ import (
 	"github.com/mysteriumnetwork/node/cmd/commands/cli/clio"
 	"github.com/mysteriumnetwork/node/config"
 	"github.com/mysteriumnetwork/node/config/urfavecli/clicontext"
+	"github.com/mysteriumnetwork/node/core/control"
 	"github.com/mysteriumnetwork/node/core/node"
 	"github.com/mysteriumnetwork/node/metadata"
 	"github.com/mysteriumnetwork/node/services"
@@ -80,7 +81,9 @@ func NewCommand(licenseCommandName string) *cli.Command {
 				errorChannel: quit,
 			}
 			go func() {
-				quit <- cmdService.Run(ctx)
+				cp := control.NewControlPlane(di.BrokerConnection, cmdService.tequilapi)
+				quit <- cmdService.Run(ctx, cp)
+				cp.Stop()
 			}()
 
 			return describeQuit(<-quit)
@@ -114,7 +117,7 @@ type serviceCommand struct {
 }
 
 // Run runs a command
-func (sc *serviceCommand) Run(ctx *cli.Context) (err error) {
+func (sc *serviceCommand) Run(ctx *cli.Context, cp *control.ControlPlane) (err error) {
 	serviceTypes := make([]string, 0)
 
 	activeServices := config.Current.GetString(config.FlagActiveServices.Name)
@@ -160,6 +163,9 @@ func (sc *serviceCommand) Run(ctx *cli.Context) (err error) {
 		go sc.runService(startRequest)
 	}
 
+	if err := cp.Start(providerID); err != nil {
+		return err
+	}
 	return <-sc.errorChannel
 }
 

core/control/consumer.go

@@ -0,0 +1,45 @@
+/*
+ * Copyright (C) 2019 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package control
+
+import (
+	"github.com/mysteriumnetwork/node/communication"
+)
+
+var _ communication.MessageConsumer = (*consumer)(nil)
+
+// consumer represents the pub/sub consumer of control messages
+type consumer struct {
+	callback func(controlMessage) error
+	topic    communication.MessageEndpoint
+}
+
+// GetMessageEndpoint returns endpoint where to receive messages
+func (c *consumer) GetMessageEndpoint() (communication.MessageEndpoint, error) {
+	return c.topic, nil
+}
+
+// NewMessage creates struct where message from endpoint will be serialized
+func (c *consumer) NewMessage() (messagePtr interface{}) {
+	return &controlMessage{}
+}
+
+// Consume handles messages from endpoint
+func (c *consumer) Consume(messagePtr interface{}) error {
+	return c.callback(*messagePtr.(*controlMessage))
+}

core/control/control.go

@@ -0,0 +1,60 @@
+/*
+ * Copyright (C) 2019 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package control
+
+import (
+	"fmt"
+
+	"github.com/mysteriumnetwork/node/communication"
+	"github.com/mysteriumnetwork/node/communication/nats"
+	"github.com/mysteriumnetwork/node/tequilapi/client"
+)
+
+type controlMessage []struct {
+	Service string `json:"service"`
+	Command string `json:"command"`
+}
+
+// ControlPlane is a struct that represents the control plane of the node
+type ControlPlane struct {
+	nats     communication.Receiver
+	api      *client.Client
+	identity string
+}
+
+// NewControlPlane creates a new control plane
+func NewControlPlane(connection nats.Connection, api *client.Client) *ControlPlane {
+	return &ControlPlane{
+		nats: nats.NewReceiver(connection, communication.NewCodecJSON(), ""),
+		api:  api,
+	}
+}
+
+// Start starts the control plane
+func (c *ControlPlane) Start(identity string) error {
+	c.identity = identity
+	return c.nats.Receive(communication.MessageConsumer(&consumer{
+		callback: c.handler,
+		topic:    communication.MessageEndpoint(fmt.Sprintf("%s.control-plane.v1", identity)),
+	}))
+}
+
+// Stop stops the control plane
+func (c *ControlPlane) Stop() {
+	c.nats.ReceiveUnsubscribe(communication.MessageEndpoint(fmt.Sprintf("%s.control-plane.v1", c.identity)))
+}

core/control/handler.go

@@ -0,0 +1,85 @@
+/*
+ * Copyright (C) 2019 The "MysteriumNetwork/node" Authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package control
+
+import (
+	"github.com/mysteriumnetwork/node/services"
+	"github.com/mysteriumnetwork/node/tequilapi/contract"
+	"github.com/rs/zerolog/log"
+)
+
+// handler is a function that handles control messages
+func (c *ControlPlane) handler(request controlMessage) error {
+	currentServices, err := c.api.Services()
+	if err != nil {
+		return err
+	}
+
+	for _, r := range request {
+		if r.Command == "start" {
+			log.Info().Str("service", r.Service).Msg("executing control start request")
+			if err := c.startService(r.Service, c.identity); err != nil {
+				log.Warn().AnErr("err", err).Msg("failed to start service")
+			}
+			continue
+		}
+
+		for _, service := range currentServices {
+			if service.Type != r.Service {
+				continue
+			}
+
+			if r.Command == "stop" {
+				log.Info().Str("service", service.Type).Msg("executing control stop request")
+				if err := c.stopService(service.ID); err != nil {
+					log.Warn().AnErr("err", err).Msg("failed to start service")
+				}
+			}
+
+			if r.Command == "restart" {
+				log.Info().Str("service", service.Type).Msg("executing control restart request")
+				if err := c.stopService(service.ID); err != nil {
+					log.Warn().AnErr("err", err).Msg("failed to stop service")
+				}
+				if err := c.startService(service.Type, c.identity); err != nil {
+					log.Warn().AnErr("err", err).Msg("failed to start service")
+				}
+			}
+		}
+	}
+	return nil
+}
+
+func (c *ControlPlane) startService(serviceType, providerID string) error {
+	serviceOpts, err := services.GetStartOptions(serviceType)
+	if err != nil {
+		return err
+	}
+	startRequest := contract.ServiceStartRequest{
+		ProviderID:     providerID,
+		Type:           serviceType,
+		AccessPolicies: &contract.ServiceAccessPolicies{IDs: serviceOpts.AccessPolicyList},
+		Options:        serviceOpts,
+	}
+	_, err = c.api.ServiceStart(startRequest)
+	return err
+}
+
+func (c *ControlPlane) stopService(id string) error {
+	return c.api.ServiceStop(id)
+}