- Draft implementation of the service plugin

5 files changed, 247 insertions, 3 deletions

diff --git a/plugins/logger/config.go b/plugins/logger/config.go
index eee5fb71..c435e8be 100644
--- a/plugins/logger/config.go
+++ b/plugins/logger/config.go
@@ -22,7 +22,7 @@ type Config struct {
 	// level of all loggers descended from this config.
 	Level string `mapstructure:"level"`
 
-	// Encoding sets the logger's encoding. Valid values are "json" and
+	// Encoding sets the logger's encoding. InitDefault values are "json" and
 	// "console", as well as any third-party encodings registered via
 	// RegisterEncoder.
 	Encoding string `mapstructure:"encoding"`
diff --git a/plugins/logger/plugin.go b/plugins/logger/plugin.go
index 08fc2454..e1066cba 100644
--- a/plugins/logger/plugin.go
+++ b/plugins/logger/plugin.go
@@ -69,7 +69,7 @@ func (z *ZapLogger) NamedLogger(name string) (Logger, error) {
 	return NewZapAdapter(z.base.Named(name)), nil
 }
 
-// NamedLogger returns logger dedicated to the specific channel. Similar to Named() but also reads the core params.
+// ServiceLogger returns logger dedicated to the specific channel. Similar to Named() but also reads the core params.
 func (z *ZapLogger) ServiceLogger(n endure.Named) (Logger, error) {
 	return z.NamedLogger(n.Name())
 }
@@ -78,5 +78,6 @@ func (z *ZapLogger) ServiceLogger(n endure.Named) (Logger, error) {
 func (z *ZapLogger) Provides() []interface{} {
 	return []interface{}{
 		z.ServiceLogger,
+		z.DefaultLogger,
 	}
 }
diff --git a/plugins/service/config.go b/plugins/service/config.go
new file mode 100644
index 00000000..b1099e06
--- /dev/null
+++ b/plugins/service/config.go
@@ -0,0 +1,34 @@
+package service
+
+import "time"
+
+// Service represents particular service configuration
+type Service struct {
+	Command          string        `mapstructure:"command"`
+	ProcessNum       int           `mapstructure:"process_num"`
+	ExecTimeout      time.Duration `mapstructure:"exec_timeout"`
+	RestartAfterExit bool          `mapstructure:"restart_after_exit"`
+	RestartDelay     time.Duration `mapstructure:"restart_delay"`
+}
+
+// Config for the services
+type Config struct {
+	Services map[string]Service `mapstructure:"service"`
+}
+
+func (c *Config) InitDefault() {
+	if len(c.Services) > 0 {
+		for k, v := range c.Services {
+			if v.ProcessNum == 0 {
+				val := c.Services[k]
+				val.ProcessNum = 1
+				c.Services[k] = val
+			}
+			if v.RestartDelay == 0 {
+				val := c.Services[k]
+				val.RestartDelay = time.Minute
+				c.Services[k] = val
+			}
+		}
+	}
+}
diff --git a/plugins/service/plugin.go b/plugins/service/plugin.go
index 858408e2..75e849a3 100644
--- a/plugins/service/plugin.go
+++ b/plugins/service/plugin.go
@@ -1,13 +1,88 @@
 package service
 
 import (
+	"sync"
+
+	"github.com/spiral/errors"
 	"github.com/spiral/roadrunner/v2/plugins/config"
 	"github.com/spiral/roadrunner/v2/plugins/logger"
 )
 
+const PluginName string = "service"
+
 type Plugin struct {
+	sync.Mutex
+
+	logger logger.Logger
+	cfg    Config
+
+	// all processes attached to the service
+	processes []*Process
 }
 
-func (p *Plugin) Init(cfg config.Configurer, log logger.Logger) error {
+func (service *Plugin) Init(cfg config.Configurer, log logger.Logger) error {
+	const op = errors.Op("service_plugin_init")
+	if !cfg.Has(PluginName) {
+		return errors.E(errors.Disabled)
+	}
+	err := cfg.UnmarshalKey(PluginName, &service.cfg.Services)
+	if err != nil {
+		return errors.E(op, err)
+	}
+
+	// init default parameters if not set by user
+	service.cfg.InitDefault()
+	// save the logger
+	service.logger = log
+
 	return nil
 }
+
+func (service *Plugin) Serve() chan error {
+	errCh := make(chan error, 1)
+
+	// start processing
+	go func() {
+		service.processes = make([]*Process, 0, len(service.cfg.Services))
+		// for the every service
+		for k := range service.cfg.Services {
+			// create needed number of the processes
+			for i := 0; i < service.cfg.Services[k].ProcessNum; i++ {
+				// create processor structure, which will process all the services
+				service.processes = append(service.processes, NewFatProcess(
+					service.cfg.Services[k].RestartAfterExit,
+					service.cfg.Services[k].ExecTimeout,
+					service.cfg.Services[k].RestartDelay,
+					service.cfg.Services[k].Command,
+					service.logger,
+					errCh,
+				))
+			}
+		}
+
+		service.Lock()
+		for i := 0; i < len(service.processes); i++ {
+			service.processes[i].start()
+		}
+		service.Unlock()
+	}()
+
+	return errCh
+}
+
+func (service *Plugin) Stop() error {
+	service.Lock()
+	defer service.Unlock()
+
+	if len(service.processes) > 0 {
+		for i := 0; i < len(service.processes); i++ {
+			service.processes[i].stop()
+		}
+	}
+	return nil
+}
+
+// Name contains service name.
+func (service *Plugin) Name() string {
+	return PluginName
+}
diff --git a/plugins/service/process.go b/plugins/service/process.go
new file mode 100644
index 00000000..449f005e
--- /dev/null
+++ b/plugins/service/process.go
@@ -0,0 +1,134 @@
+package service
+
+import (
+	"os/exec"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+	"unsafe"
+
+	"github.com/spiral/errors"
+	"github.com/spiral/roadrunner/v2/plugins/logger"
+)
+
+// Process structure contains an information about process, restart information, log, errors, etc
+type Process struct {
+	sync.Mutex
+	// command to execute
+	command *exec.Cmd
+	rawCmd  string
+
+	errCh chan error
+	log   logger.Logger
+
+	ExecTimeout      time.Duration
+	RestartAfterExit bool
+	RestartDelay     time.Duration
+
+	//
+	startTime time.Time
+	stopCh    chan struct{}
+}
+
+func NewFatProcess(restartAfterExit bool, execTimeout, restartDelay time.Duration, command string, l logger.Logger, errCh chan error) *Process {
+	p := &Process{
+		rawCmd:           command,
+		RestartDelay:     restartDelay,
+		ExecTimeout:      execTimeout,
+		RestartAfterExit: restartAfterExit,
+		errCh:            errCh,
+		stopCh:           make(chan struct{}),
+		log:              l,
+	}
+	// stderr redirect to the logger
+	return p
+}
+
+// write message to the log (stderr)
+func (p *Process) Write(b []byte) (int, error) {
+	p.log.Info(toString(b))
+	return len(b), nil
+}
+
+func (p *Process) start() {
+	p.Lock()
+	defer p.Unlock()
+
+	const op = errors.Op("processor_start")
+
+	// cmdArgs contain command arguments if the command in form of: php <command> or ls <command> -i -b
+	p.createProcess()
+
+	err := p.command.Start()
+	if err != nil {
+		p.errCh <- errors.E(op, err)
+		return
+	}
+
+	go p.wait()
+	go p.execHandler()
+	// save start time
+	p.startTime = time.Now()
+}
+
+func (p *Process) createProcess() {
+	var cmdArgs []string
+	cmdArgs = append(cmdArgs, strings.Split(p.rawCmd, " ")...)
+	if len(cmdArgs) < 2 {
+		p.command = exec.Command(p.rawCmd) //nolint:gosec
+	} else {
+		p.command = exec.Command(cmdArgs[0], cmdArgs[1:]...) //nolint:gosec
+	}
+	p.command.Stderr = p
+}
+
+func (p *Process) wait() {
+	// Wait error doesn't matter here
+	_ = p.command.Wait()
+
+	// wait for restart delay
+	if p.RestartAfterExit {
+		// wait for the delay
+		time.Sleep(p.RestartDelay)
+		// and start command again
+		p.start()
+	}
+}
+
+// stop can be only sent by the Endure when plugin stopped
+func (p *Process) stop() {
+	p.stopCh <- struct{}{}
+}
+
+func (p *Process) execHandler() {
+	tt := time.NewTicker(time.Second)
+	for {
+		select {
+		case <-tt.C:
+			p.Lock()
+			// if the exec timeout is set
+			if p.ExecTimeout != 0 {
+				// check the running time for the script
+				if time.Now().After(p.startTime.Add(p.ExecTimeout)) {
+					err := p.command.Process.Signal(syscall.SIGINT)
+					if err != nil {
+						_ = p.command.Process.Signal(syscall.SIGKILL)
+					}
+				}
+			}
+			p.Unlock()
+		case <-p.stopCh:
+			err := p.command.Process.Signal(syscall.SIGINT)
+			if err != nil {
+				_ = p.command.Process.Signal(syscall.SIGKILL)
+			}
+			tt.Stop()
+			return
+		}
+	}
+}
+
+func toString(data []byte) string {
+	return *(*string)(unsafe.Pointer(&data))
+}