summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorValery Piashchynski <[email protected]>2021-04-18 17:31:52 +0300
committerValery Piashchynski <[email protected]>2021-04-18 17:31:52 +0300
commit4e6dfc00c5619c4e749602d345fd2829ab0a3f07 (patch)
tree12b4ce7644b02b6e76cfa46e8c80d54690d6e0dc
parent15b7a9a0fc074531f9b46bb87fb35819e248a58c (diff)
- Draft implementation of the service plugin
-rw-r--r--plugins/logger/config.go2
-rw-r--r--plugins/logger/plugin.go3
-rw-r--r--plugins/service/config.go34
-rw-r--r--plugins/service/plugin.go77
-rw-r--r--plugins/service/process.go134
-rw-r--r--tests/plugins/service/configs/.rr-service-init.yaml21
-rw-r--r--tests/plugins/service/php/loop.php6
-rw-r--r--tests/plugins/service/service_plugin_test.go81
8 files changed, 355 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))
+}
diff --git a/tests/plugins/service/configs/.rr-service-init.yaml b/tests/plugins/service/configs/.rr-service-init.yaml
new file mode 100644
index 00000000..9ca0bde5
--- /dev/null
+++ b/tests/plugins/service/configs/.rr-service-init.yaml
@@ -0,0 +1,21 @@
+service:
+ some_service_1: #<-- user defined name
+ command: "php php/loop.php" # command, can be any command (php, script, binary, exe)"
+ process_num: 10 # number of copies (processes)
+ exec_timeout: 5s # how long process allowed to run (until restart or stop), default - unlimited
+ restart_after_exit: false # run and restart after finish
+ restart_delay: 1s # delay between binary/script restarts, default 1minute
+# some_service_2: # exec_timeout is not set, by default - unlimited
+# command: "./test_binary"
+# process_num: 1
+# restart_after_exit: false
+# restart_delay: 10 # <--- ignored when restart_after_exit is false
+
+logs:
+ level: debug
+ mode: raw
+
+endure:
+ grace_period: 120s
+ print_graph: false
+ log_level: debug
diff --git a/tests/plugins/service/php/loop.php b/tests/plugins/service/php/loop.php
new file mode 100644
index 00000000..6ba488ef
--- /dev/null
+++ b/tests/plugins/service/php/loop.php
@@ -0,0 +1,6 @@
+<?php
+for ($x = 0; $x <= 1000; $x++) {
+ sleep(1);
+ error_log("The number is: $x", 0);
+}
+?>
diff --git a/tests/plugins/service/service_plugin_test.go b/tests/plugins/service/service_plugin_test.go
new file mode 100644
index 00000000..add6d374
--- /dev/null
+++ b/tests/plugins/service/service_plugin_test.go
@@ -0,0 +1,81 @@
+package service
+
+import (
+ "os"
+ "os/signal"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+
+ endure "github.com/spiral/endure/pkg/container"
+ "github.com/spiral/roadrunner/v2/plugins/config"
+ "github.com/spiral/roadrunner/v2/plugins/logger"
+ "github.com/spiral/roadrunner/v2/plugins/service"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestServiceInit(t *testing.T) {
+ cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.DebugLevel))
+ assert.NoError(t, err)
+
+ cfg := &config.Viper{
+ Path: "configs/.rr-service-init.yaml",
+ Prefix: "rr",
+ }
+
+ err = cont.RegisterAll(
+ cfg,
+ &logger.ZapLogger{},
+ &service.Plugin{},
+ )
+ assert.NoError(t, err)
+
+ err = cont.Init()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ ch, err := cont.Serve()
+ assert.NoError(t, err)
+
+ sig := make(chan os.Signal, 1)
+ signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+ wg := &sync.WaitGroup{}
+ wg.Add(1)
+
+ stopCh := make(chan struct{}, 1)
+
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case e := <-ch:
+ assert.Fail(t, "error", e.Error.Error())
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ return
+ case <-sig:
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ return
+ case <-stopCh:
+ // timeout
+ err = cont.Stop()
+ if err != nil {
+ assert.FailNow(t, "error", err.Error())
+ }
+ return
+ }
+ }
+ }()
+
+ time.Sleep(time.Second * 10)
+ stopCh <- struct{}{}
+ wg.Wait()
+}